path: root/scripts/notify_completion.py

#!/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 <sha256>``. 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-<timestamp>.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:]))