126 lines
3.5 KiB
Python
126 lines
3.5 KiB
Python
#!/usr/bin/env python3
|
|
import json
|
|
import sys
|
|
import time
|
|
import urllib.request
|
|
|
|
NTFY_URL = "https://ntfy.unbl.ink/hr-agents"
|
|
|
|
|
|
def read_jsonl(path):
|
|
entries = []
|
|
try:
|
|
with open(path) as f:
|
|
for line in f:
|
|
line = line.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
entries.append(json.loads(line))
|
|
except json.JSONDecodeError:
|
|
continue
|
|
except OSError:
|
|
pass
|
|
return entries
|
|
|
|
|
|
def load_jsonl(path, has_assistant_text=None):
|
|
if not path:
|
|
return []
|
|
entries = read_jsonl(path)
|
|
if has_assistant_text is not None:
|
|
for _ in range(4):
|
|
if has_assistant_text(entries):
|
|
break
|
|
time.sleep(0.5)
|
|
entries = read_jsonl(path)
|
|
return entries
|
|
|
|
|
|
def text_blocks(entry):
|
|
content = entry.get("message", {}).get("content")
|
|
if isinstance(content, str):
|
|
yield content
|
|
elif isinstance(content, list):
|
|
for c in content:
|
|
if isinstance(c, dict) and c.get("type") == "text":
|
|
yield c.get("text", "")
|
|
|
|
|
|
def first_text(entries, role):
|
|
for e in entries:
|
|
if e.get("type") != role:
|
|
continue
|
|
for t in text_blocks(e):
|
|
if t.strip():
|
|
return t.strip()
|
|
return ""
|
|
|
|
|
|
def last_text(entries, role):
|
|
for e in reversed(entries):
|
|
if e.get("type") != role:
|
|
continue
|
|
for t in text_blocks(e):
|
|
if t.strip():
|
|
return t.strip()
|
|
return ""
|
|
|
|
|
|
def truncate(s, n):
|
|
s = " ".join(s.split())
|
|
return s if len(s) <= n else s[: n - 1] + "…"
|
|
|
|
|
|
def post(title, message, tags, priority=None):
|
|
req = urllib.request.Request(
|
|
NTFY_URL, data=message.encode("utf-8"), method="POST"
|
|
)
|
|
req.add_header("Title", title)
|
|
req.add_header("Tags", tags)
|
|
if priority:
|
|
req.add_header("Priority", priority)
|
|
urllib.request.urlopen(req, timeout=10)
|
|
|
|
|
|
def main():
|
|
payload = json.load(sys.stdin)
|
|
is_subagent = bool(payload.get("agent_transcript_path")) or (
|
|
payload.get("hook_event_name") == "SubagentStop"
|
|
)
|
|
|
|
if is_subagent:
|
|
agent_type = payload.get("agent_type", "subagent")
|
|
entries = load_jsonl(
|
|
payload.get("agent_transcript_path"),
|
|
has_assistant_text=lambda e: bool(last_text(e, "assistant")),
|
|
)
|
|
task = truncate(first_text(entries, "user"), 200)
|
|
result = truncate(last_text(entries, "assistant"), 500)
|
|
message = (
|
|
"\n\n".join(p for p in (f"Task: {task}" if task else "", f"Result: {result}" if result else ""))
|
|
or "(no transcript content found)"
|
|
)
|
|
post(f"Subagent finished: {agent_type}", message, "robot", priority="low")
|
|
else:
|
|
entries = load_jsonl(
|
|
payload.get("transcript_path"),
|
|
has_assistant_text=lambda e: bool(last_text(e, "assistant")),
|
|
)
|
|
request = truncate(last_text(entries, "user"), 200)
|
|
response = truncate(last_text(entries, "assistant"), 500)
|
|
message = (
|
|
"\n\n".join(p for p in (f"Last request: {request}" if request else "", f"Last reply: {response}" if response else ""))
|
|
or "(no transcript content found)"
|
|
)
|
|
reason = payload.get("reason", "")
|
|
title = f"Claude Code: session ended ({reason})" if reason else "Claude Code: session ended"
|
|
post(title, message, "white_check_mark")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
main()
|
|
except Exception:
|
|
pass
|