1 files changed, 132 insertions, 0 deletions
diff --git a/scripts/completion_notify.py b/scripts/completion_notify.py
new file mode 100644
index 0000000..e18796b
--- /dev/null
+++ b/scripts/completion_notify.py
@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+"""completion-notice.md 变更通知：供 Hermes cron 调用。
+
+行为（严格）:
+- 通知文件缺失: 静默退出 0，无任何输出
+- 文件存在: 计算 sha256；与上次已通知 hash 相同 -> 静默退出 0
+- hash 为新值: 把文件内容原样打印到 stdout 一次（供 cron 捕获后发 Slack），并原子记录该 hash
+- 每个新内容 hash 只打印一次；内容再次变化会再次打印
+- 状态文件损坏/不可读: 视为未通知过（打印一次）
+- 不依赖任何模型凭据；纯本地文件与 hash
+
+用法:
+  completion_notify.py [--notice PATH] [--state PATH] [--selftest]
+默认 notice=<cwd>/docs/completion-notice.md, state=<脚本所在目录>/completion_notify_state.json。
+脚本放项目中，由 Hermes 复制到 cron 脚本目录时建议显式传 --notice/--state。
+
+--selftest: 全部合成测试（临时文件模拟 缺失/新hash/重复/变更/损坏状态），不伪造真实结果。
+"""
+import argparse
+import hashlib
+import json
+import os
+import sys
+import tempfile
+from pathlib import Path
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+DEFAULT_NOTICE = Path.cwd() / "docs/completion-notice.md"
+DEFAULT_STATE = SCRIPT_DIR / "completion_notify_state.json"
+
+
+def content_hash(data: bytes) -> str:
+    return hashlib.sha256(data).hexdigest()
+
+
+def load_state(path: Path):
+    try:
+        v = json.loads(Path(path).read_text())
+        if isinstance(v, dict) and isinstance(v.get("hash"), str):
+            return v["hash"]
+    except Exception:
+        pass
+    return None
+
+
+def save_state(path: Path, h: str):
+    p = Path(path)
+    p.parent.mkdir(parents=True, exist_ok=True)
+    fd, tmp = tempfile.mkstemp(dir=str(p.parent), prefix=".notify_state.", suffix=".tmp")
+    try:
+        with os.fdopen(fd, "w") as fh:
+            fh.write(json.dumps({"hash": h}, indent=1) + "\n")
+        os.replace(tmp, p)
+    except BaseException:
+        try:
+            os.unlink(tmp)
+        except OSError:
+            pass
+        raise
+
+
+def run_once(notice_path, state_path, out=sys.stdout):
+    """返回 'printed' | 'silent' | 'missing'。打印发生在 out 上。"""
+    p = Path(notice_path)
+    try:
+        data = p.read_bytes()
+    except OSError:
+        return "missing"  # 缺失静默
+    h = content_hash(data)
+    if load_state(state_path) == h:
+        return "silent"
+    out.write(data.decode("utf-8", "replace"))
+    save_state(state_path, h)
+    return "printed"
+
+
+def run_once_str(notice_path, state_path):
+    """测试用: 返回 (action, printed_text)。"""
+    import io
+    buf = io.StringIO()
+    action = run_once(notice_path, state_path, out=buf)
+    return action, buf.getvalue()
+
+
+def selftest():
+    ok = True
+
+    def check(name, cond):
+        nonlocal ok
+        print(("PASS " if cond else "FAIL ") + name)
+        ok = ok and cond
+
+    import tempfile
+    with tempfile.TemporaryDirectory() as td:
+        td = Path(td)
+        notice = td / "completion-notice.md"
+        state = td / "state" / "notify.json"
+        action, text = run_once_str(notice, state)
+        check("missing -> silent no output", action == "missing" and text == "")
+        notice.write_text("# 完成\n- 测试内容\n")
+        action, text = run_once_str(notice, state)
+        check("new content printed once", action == "printed" and text == notice.read_text())
+        action, text = run_once_str(notice, state)
+        check("same hash silent", action == "silent" and text == "")
+        notice.write_text("# 完成 v2\n")
+        action, text = run_once_str(notice, state)
+        check("changed content printed again", action == "printed" and text == "# 完成 v2\n")
+        action, text = run_once_str(notice, state)
+        check("second view silent", action == "silent" and text == "")
+        state.write_text("not-json{{{")
+        action, text = run_once_str(notice, state)
+        check("corrupt state treated as new", action == "printed" and text == "# 完成 v2\n")
+        check("state saved after corrupt", load_state(state) == content_hash("# 完成 v2\n".encode()))
+
+    print("SELFTEST=" + ("PASS" if ok else "FAIL"))
+    return 0 if ok else 1
+
+
+def main():
+    ap = argparse.ArgumentParser(description="completion-notice.md new-hash notifier")
+    ap.add_argument("--notice", default=str(DEFAULT_NOTICE))
+    ap.add_argument("--state", default=str(DEFAULT_STATE))
+    ap.add_argument("--selftest", action="store_true")
+    args = ap.parse_args()
+    if args.selftest:
+        return selftest()
+    action = run_once(args.notice, args.state)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())