#!/usr/bin/env python3 """Quiet, once-only completion notifier for the Fund Lab supervision workflow. Reads a notices directory and, when a marker-stamped ``completion-notice.md`` is present, sends its body to a Slack webhook exactly once and then renames the file so subsequent runs stay silent. Everything here is stdlib-only and single-file. Marker contract --------------- The notice file must start with a line of the form ``# NOTICE ``. The hash covers the file content *excluding* the marker line. When the stored marker does not match the body the script rewrites the marker (first run establishes it) and still sends. After sending, the file is renamed to ``completion-notice.sent-.md``. Environment ----------- NOTICES_DIR notices directory (default: /home/somhairle/.hermes/cache/fundlab-notices) SLACK_WEBHOOK_URL Slack incoming webhook; when unset/empty the body is not sent and no output is produced, but the file is still marked as sent. Exit codes ---------- 0 nothing to do, or delivered (quiet, no stdout) 2 file read / marker rewrite / send / rename failure (message on stderr) """ import hashlib import json import os import re import sys import tempfile import urllib.request from datetime import datetime, timezone DEFAULT_NOTICES_DIR = "/home/somhairle/.hermes/cache/fundlab-notices" NOTICE_NAME = "completion-notice.md" SENT_PREFIX = "completion-notice.sent-" MARKER_RE = re.compile(r"^# NOTICE ([0-9a-fA-F]{64})$") WEBHOOK_TIMEOUT_SECONDS = 5 def body_hash(body: str) -> str: """sha256 of the notice body (file content with the marker line removed).""" return hashlib.sha256(body.encode("utf-8")).hexdigest() def render_notice(digest: str, body: str) -> str: return "# NOTICE %s\n%s" % (digest, body) def split_notice(content: str): """Return (marker_hash, body) when the first line is a marker, else None. The body keeps everything after the marker line byte-for-byte, including trailing newlines, so the hash is stable across reads. """ lines = content.splitlines(keepends=True) if not lines: return None match = MARKER_RE.match(lines[0].rstrip("\r\n")) if match is None: return None return match.group(1).lower(), "".join(lines[1:]) def send_slack(url: str, text: str) -> None: payload = json.dumps({"text": text}).encode("utf-8") request = urllib.request.Request( url, data=payload, headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(request, timeout=WEBHOOK_TIMEOUT_SECONDS) as response: response.read() def sent_target(notices_dir: str) -> str: stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") target = os.path.join(notices_dir, "%s%s.md" % (SENT_PREFIX, stamp)) if os.path.exists(target): target = "%s-%s.md" % (target[:-3], os.getpid()) return target def process_notice(notices_dir: str, send) -> bool: """Deliver a pending notice once. Returns True when a notice was delivered.""" path = os.path.join(notices_dir, NOTICE_NAME) if not os.path.isfile(path): return False with open(path, "r", encoding="utf-8") as handle: content = handle.read() parsed = split_notice(content) if parsed is None: return False marker, body = parsed digest = body_hash(body) if digest != marker: with open(path, "w", encoding="utf-8") as handle: handle.write(render_notice(digest, body)) url = (os.environ.get("SLACK_WEBHOOK_URL") or "").strip() if url: send(url, body) os.replace(path, sent_target(notices_dir)) return True def _report(name: str, ok: bool, failures: list) -> None: if ok: print("PASS %s" % name) else: print("FAIL %s" % name) failures.append(name) def selftest() -> int: failures = [] saved_webhook = os.environ.get("SLACK_WEBHOOK_URL") os.environ["SLACK_WEBHOOK_URL"] = "https://example.invalid/selftest" try: with tempfile.TemporaryDirectory(prefix="notify-selftest-") as tmp: # 1) no notice at all -> silent, nothing sent, nothing renamed sent = [] result = process_notice(tmp, lambda url, text: sent.append((url, text))) _report( "no notice is silent", result is False and sent == [] and os.listdir(tmp) == [], failures, ) # 2) healthy marker -> body sent once, file renamed, replay silent body = "标题\n正文 100%\n" path = os.path.join(tmp, NOTICE_NAME) with open(path, "w", encoding="utf-8") as handle: handle.write(render_notice(body_hash(body), body)) sent = [] result = process_notice(tmp, lambda url, text: sent.append((url, text))) files = sorted(os.listdir(tmp)) renamed = any(name.startswith(SENT_PREFIX) and name.endswith(".md") for name in files) replay = process_notice(tmp, lambda url, text: sent.append((url, text))) _report( "notice is sent and renamed once", result is True and sent == [("https://example.invalid/selftest", body)] and NOTICE_NAME not in files and renamed and replay is False and len(sent) == 1, failures, ) # 3) broken marker -> repaired, then sent and renamed body = "损坏 marker\n" path = os.path.join(tmp, NOTICE_NAME) repaired = render_notice(body_hash(body), body) with open(path, "w", encoding="utf-8") as handle: handle.write(render_notice("0" * 64, body)) sent = [] result = process_notice(tmp, lambda url, text: sent.append((url, text))) files = sorted(os.listdir(tmp)) corrected = [ name for name in files if name.startswith(SENT_PREFIX) and open(os.path.join(tmp, name), "r", encoding="utf-8").read() == repaired ] _report( "broken marker is repaired and sent", result is True and sent == [("https://example.invalid/selftest", body)] and len(corrected) == 1 and NOTICE_NAME not in files, failures, ) finally: if saved_webhook is None: os.environ.pop("SLACK_WEBHOOK_URL", None) else: os.environ["SLACK_WEBHOOK_URL"] = saved_webhook return 0 if not failures else 1 def main(argv) -> int: if "--selftest" in argv: return selftest() notices_dir = (os.environ.get("NOTICES_DIR") or "").strip() or DEFAULT_NOTICES_DIR try: process_notice(notices_dir, send_slack) return 0 except Exception as error: # noqa: BLE001 - classification is by exit code print("notify_completion: %s" % error, file=sys.stderr) return 2 if __name__ == "__main__": sys.exit(main(sys.argv[1:]))