168 lines
6.6 KiB
Python
168 lines
6.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Harness -> Slack notification layer (redact -> format -> deliver).
|
|
|
|
Notifies only the events org-os policy allows: blocker / human-review / critical
|
|
/ digest / task / review. Sensitive data is masked before it ever leaves.
|
|
|
|
Delivery (Claude Code hooks cannot call MCP directly, so two paths):
|
|
- if $SLACK_WEBHOOK_URL set -> POST directly (fully autonomous, headless-safe)
|
|
- else -> enqueue to slack-outbox/*.json
|
|
(main session flushes via mcp__slack__slack_post_message)
|
|
|
|
Usage:
|
|
notify_slack.py <event> [report.yaml] [--title "..."] [--channel C0BCN9H9ABH]
|
|
<event> in: task | review | blocker | human-review | critical | digest
|
|
"""
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
import urllib.request
|
|
from datetime import datetime
|
|
|
|
ROOT = os.environ.get("CLAUDE_PROJECT_DIR") or os.path.dirname(
|
|
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
)
|
|
DEFAULT_CHANNEL = "C0BCN9H9ABH" # #clean-architecture-전체 (사전 승인)
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
import _workspace as W # noqa: E402
|
|
OUTBOX = W.slack_outbox()
|
|
|
|
# redact-before-slack: mask secrets/PII before anything leaves the harness
|
|
REDACT = [
|
|
(re.compile(r"[\w.+-]+@[\w-]+\.[\w.-]+"), "***@masked"),
|
|
(re.compile(r"xox[baprs]-[A-Za-z0-9-]+"), "***REDACTED***"),
|
|
(re.compile(r"sk-(ant-)?[A-Za-z0-9._-]{20,}"), "***REDACTED***"),
|
|
(re.compile(r"AKIA[0-9A-Z]{16}"), "***REDACTED***"),
|
|
(re.compile(r"gh[opsu]_[A-Za-z0-9]{30,}"), "***REDACTED***"),
|
|
(re.compile(r"(?i)bearer\s+[A-Za-z0-9._-]{10,}"), "bearer ***REDACTED***"),
|
|
(re.compile(r"(?i)\b(password|passwd|secret|token|api[_-]?key)\b\s*[:=]\s*\S+"), r"\1: ***REDACTED***"),
|
|
(re.compile(r"\b[0-9a-fA-F]{32,}\b"), "***REDACTED***"),
|
|
]
|
|
EMOJI = {"task": ":memo:", "review": ":mag:", "blocker": ":rotating_light:",
|
|
"human-review": ":raising_hand:", "critical": ":red_circle:", "digest": ":bar_chart:",
|
|
"report": ":round_pushpin:"}
|
|
|
|
|
|
def redact(text):
|
|
for pat, repl in REDACT:
|
|
text = pat.sub(repl, text)
|
|
return text
|
|
|
|
|
|
def header():
|
|
ts = datetime.now().strftime("%Y-%m-%d %H:%M")
|
|
return f"`repo: {os.path.basename(ROOT)}` · `branch: (no git)` · `{ts}`"
|
|
|
|
|
|
def load_doc(report_path):
|
|
if not report_path or report_path == "-" or not os.path.exists(report_path):
|
|
return {}
|
|
import yaml
|
|
d = yaml.safe_load(open(report_path)) or {}
|
|
return d if isinstance(d, dict) else {}
|
|
|
|
|
|
def load_header(report_path):
|
|
return load_doc(report_path).get("report-header", {})
|
|
|
|
|
|
def _max_grade(rh):
|
|
gs = [str(e.get("grade", "")) for e in (rh.get("evidence") or []) if isinstance(e, dict)]
|
|
gs = [g for g in gs if g.startswith("E")]
|
|
return max(gs) if gs else None
|
|
|
|
|
|
def build_report(doc, title):
|
|
"""템플릿 3(agent-report): 직무 정체성 → BLUF → 근거·산출물 → 결정필요/승인자 → 동료 cc."""
|
|
rh = doc.get("report-header", {}) or {}
|
|
role = doc.get("role-name") or doc.get("role-id") or "-"
|
|
lens = doc.get("lens") or "-"
|
|
fam = doc.get("role-id") or doc.get("synthesized-by") or "AGENT"
|
|
lines = [f":round_pushpin: *[{fam}] {title}*",
|
|
f"`role: {role} · lens: {lens}` · {header()}", ""]
|
|
bl = rh.get("bottom-line")
|
|
conf = rh.get("confidence") or {}
|
|
cval = conf.get("value", "?") if isinstance(conf, dict) else str(conf)
|
|
g = _max_grade(rh)
|
|
if bl:
|
|
lines += [f"*BLUF* — {str(bl).strip()} (신뢰도: {cval}{' · 근거 ' + g if g else ''})", ""]
|
|
did = doc.get("findings") or ([doc["work-summary"]] if doc.get("work-summary") else []) \
|
|
or ([doc["recommendation"]] if doc.get("recommendation") else [])
|
|
if did:
|
|
lines += ["*무엇을 했나 / 근거*"] + [f"• {str(d).strip()}" for d in did[:4]] + [""]
|
|
outs = doc.get("output-artifacts") or doc.get("linked-reports") or []
|
|
if outs:
|
|
lines += ["*산출물*"] + [f"• `{o.get('uri') if isinstance(o, dict) else o}`" for o in outs[:4]] + [""]
|
|
dn = rh.get("decision-needed") or {}
|
|
if isinstance(dn, dict) and dn.get("needed"):
|
|
lines.append(f":vertical_traffic_light: *결정 필요* — 승인자: *{dn.get('approver', '?')}*")
|
|
risks = rh.get("risks") or []
|
|
if risks:
|
|
lines.append(f":warning: *리스크* — {str(risks[0]).strip()}")
|
|
tags = doc.get("tags") or []
|
|
if tags:
|
|
lines += ["", f":handshake: cc {' '.join('*#' + str(t) + '*' for t in tags[:5])}"]
|
|
return redact("\n".join(lines).rstrip())
|
|
|
|
|
|
def build(event, rh, title):
|
|
emoji = EMOJI.get(event, ":memo:")
|
|
label = {"blocker": "Blocker", "human-review": "검토 요청", "critical": "Critical",
|
|
"digest": "Daily Digest", "review": "리뷰", "task": "작업"}.get(event, event)
|
|
lines = [f"{emoji} *[{label}] {title}*", header(), ""]
|
|
bl = rh.get("bottom-line")
|
|
if bl:
|
|
lines += ["*핵심(BLUF)*", f"• {str(bl).strip()}", ""]
|
|
dn = rh.get("decision-needed") or {}
|
|
if isinstance(dn, dict) and dn.get("needed"):
|
|
lines += [f"*결정 필요* — 승인자: `{dn.get('approver', '?')}`", ""]
|
|
risks = rh.get("risks") or []
|
|
if risks:
|
|
lines += ["*리스크*"] + [f"• {r}" for r in risks[:3]] + [""]
|
|
return redact("\n".join(lines).rstrip())
|
|
|
|
|
|
def deliver(channel, text):
|
|
webhook = os.environ.get("SLACK_WEBHOOK_URL")
|
|
if webhook:
|
|
req = urllib.request.Request(webhook, data=json.dumps({"text": text}).encode(),
|
|
headers={"Content-Type": "application/json"})
|
|
urllib.request.urlopen(req, timeout=10)
|
|
return "webhook"
|
|
os.makedirs(OUTBOX, exist_ok=True)
|
|
stamp = datetime.now().strftime("%Y%m%d-%H%M%S-%f")
|
|
with open(os.path.join(OUTBOX, f"{stamp}.json"), "w") as f:
|
|
json.dump({"channel_id": channel, "text": text}, f, ensure_ascii=False, indent=2)
|
|
return "outbox"
|
|
|
|
|
|
def main():
|
|
args = [a for a in sys.argv[1:]]
|
|
if not args:
|
|
sys.stderr.write("usage: notify_slack.py <event> [report.yaml] [--title ..] [--channel ..]\n")
|
|
sys.exit(1)
|
|
event = args[0]
|
|
report = None
|
|
title = None
|
|
channel = DEFAULT_CHANNEL
|
|
i = 1
|
|
while i < len(args):
|
|
if args[i] == "--title":
|
|
title = args[i + 1]; i += 2
|
|
elif args[i] == "--channel":
|
|
channel = args[i + 1]; i += 2
|
|
else:
|
|
report = args[i]; i += 1
|
|
title = title or (report and os.path.basename(report)) or event
|
|
if event == "report":
|
|
text = build_report(load_doc(report), title)
|
|
else:
|
|
text = build(event, load_header(report), title)
|
|
mode = deliver(channel, text)
|
|
print(f"[notify_slack] {event} -> {mode}\n{text}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|