1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
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())
|
