73 lines
2.6 KiB
Python
73 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Pre-work Slack inbox — 작업 시작 전 읽어야 할 Slack 메시지를 파일로.
|
|
|
|
subagent/hook은 MCP(Slack)를 직접 못 부른다. 그래서 Orchestrator(메인 세션, MCP 보유)가
|
|
`mcp__slack__slack_get_channel_history`로 관련 메시지를 가져와 이 도구에 JSON으로 파이프하면,
|
|
`slack-inbox/<workflow>.md`로 정리해 준다. 그 파일을 워커 context-package의 must-read로 넣는다.
|
|
|
|
Usage:
|
|
<slack history json> | slack_inbox.py --workflow WF [--channel NAME] [--title T]
|
|
stdin JSON: {"messages":[{"user":..,"text":..,"ts":..}, ...]} 또는 메시지 배열
|
|
"""
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
ROOT = os.environ.get("CLAUDE_PROJECT_DIR") or os.path.dirname(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
)
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import _workspace as W # noqa: E402
|
|
INBOX = W.slack_inbox()
|
|
|
|
|
|
def fmt_ts(ts):
|
|
try:
|
|
return datetime.fromtimestamp(float(ts), timezone.utc).strftime("%Y-%m-%d %H:%M")
|
|
except Exception:
|
|
return str(ts or "-")
|
|
|
|
|
|
def main():
|
|
a = sys.argv[1:]
|
|
opt = {}
|
|
i = 0
|
|
while i < len(a):
|
|
if a[i].startswith("--"):
|
|
opt[a[i][2:]] = a[i + 1] if i + 1 < len(a) else ""
|
|
i += 2
|
|
else:
|
|
i += 1
|
|
wf = opt.get("workflow", "adhoc")
|
|
raw = sys.stdin.read().strip()
|
|
msgs = []
|
|
if raw:
|
|
try:
|
|
data = json.loads(raw)
|
|
msgs = data.get("messages", data) if isinstance(data, dict) else data
|
|
except json.JSONDecodeError:
|
|
msgs = []
|
|
os.makedirs(INBOX, exist_ok=True)
|
|
out = os.path.join(INBOX, f"{wf}.md")
|
|
now = datetime.now().strftime("%Y-%m-%d %H:%M")
|
|
L = [f"# 📥 작업 전 Slack 인박스 — {wf}", "",
|
|
f"채널: {opt.get('channel', '#clean-architecture-전체')} · 가져온 시각: {now} · {len(msgs)}건",
|
|
"> 작업 시작 전 읽고, 내 작업과 관련된 결정·요청·제약을 반영한다. 관련 없으면 무시.", ""]
|
|
for m in (msgs or []):
|
|
if not isinstance(m, dict):
|
|
L.append(f"- {m}")
|
|
continue
|
|
who = m.get("user") or m.get("username") or m.get("bot_id") or "?"
|
|
text = str(m.get("text", "")).replace("\n", " ").strip()
|
|
L.append(f"- `{fmt_ts(m.get('ts'))}` **{who}**: {text}")
|
|
if not msgs:
|
|
L.append("- (관련 메시지 없음)")
|
|
with open(out, "w") as f:
|
|
f.write("\n".join(L) + "\n")
|
|
print(f"[slack_inbox] {os.path.relpath(out, ROOT)} ({len(msgs)} msgs)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|