完善异机备份告警与同日恢复点取代（07-27-automatic-offhost-backup）

scripts/send-wecom-group-alert.py

@@ -0,0 +1,188 @@
+#!/usr/bin/env python3
+"""Send a redacted, deduplicated OA operations alert to a WeCom group robot."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import stat
+import sys
+import tempfile
+import urllib.parse
+import urllib.request
+from pathlib import Path
+
+
+SAFE_TOKEN = re.compile(r"^[A-Za-z0-9_-]{1,160}$")
+ALLOWED_STATUSES = {
+    "FAIL_AUTOMATIC_BACKUP",
+    "FAIL_OFFHOST_DEADMAN",
+    "PASS_AUTOMATIC_BACKUP_COMPLETE",
+    "PASS_AUTOMATIC_BACKUP_FRESH",
+    "PASS_OFFHOST_DEADMAN_FRESH",
+    "PASS_WITH_FINDINGS",
+}
+
+
+def fail(message: str) -> "NoReturn":
+    print(f"[wecom-group-alert][ERROR] {message}", file=sys.stderr)
+    raise SystemExit(1)
+
+
+def validate_private_file(path: Path, label: str) -> None:
+    try:
+        info = path.lstat()
+    except FileNotFoundError:
+        fail(f"{label} is missing")
+    if stat.S_ISLNK(info.st_mode) or not stat.S_ISREG(info.st_mode):
+        fail(f"{label} must be a regular non-symbolic file")
+    if stat.S_IMODE(info.st_mode) != 0o600:
+        fail(f"{label} mode must be 0600")
+    if info.st_uid != os.geteuid():
+        fail(f"{label} must be owned by the executing account")
+
+
+def read_webhook(path: Path) -> str:
+    validate_private_file(path, "webhook file")
+    try:
+        value = path.read_text(encoding="utf-8").strip()
+    except OSError:
+        fail("webhook file cannot be read")
+    if "\n" in value or "\r" in value:
+        fail("webhook file must contain one line")
+    parsed = urllib.parse.urlparse(value)
+    query = urllib.parse.parse_qs(parsed.query, strict_parsing=True)
+    if (
+        parsed.scheme != "https"
+        or parsed.hostname != "qyapi.weixin.qq.com"
+        or parsed.netloc != "qyapi.weixin.qq.com"
+        or parsed.path != "/cgi-bin/webhook/send"
+        or set(query) != {"key"}
+        or len(query["key"]) != 1
+        or not re.fullmatch(r"[A-Za-z0-9-]{16,128}", query["key"][0])
+        or parsed.fragment
+    ):
+        fail("webhook file format is invalid")
+    return value
+
+
+def load_summary(path: Path) -> tuple[str, str, int]:
+    validate_private_file(path, "summary file")
+    try:
+        payload = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        fail("summary file is invalid")
+    status = payload.get("status")
+    findings = payload.get("findings") or []
+    finding_count = payload.get("findingCount")
+    if status not in ALLOWED_STATUSES or not isinstance(finding_count, int):
+        fail("summary status is invalid")
+    if finding_count != len(findings) or finding_count not in {0, 1}:
+        fail("summary finding count is invalid")
+    category = "healthy"
+    if findings:
+        category = findings[0].get("detail", "")
+        if not isinstance(category, str) or not SAFE_TOKEN.fullmatch(category):
+            fail("summary category is invalid")
+    return status, category, finding_count
+
+
+def load_previous(path: Path) -> dict[str, object]:
+    if not path.exists():
+        return {}
+    validate_private_file(path, "alert state file")
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        fail("alert state file is invalid")
+    return value if isinstance(value, dict) else {}
+
+
+def atomic_write(path: Path, payload: dict[str, object]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    os.chmod(path.parent, 0o700)
+    fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
+    try:
+        os.fchmod(fd, 0o600)
+        with os.fdopen(fd, "w", encoding="utf-8") as handle:
+            json.dump(payload, handle, ensure_ascii=True, indent=2)
+            handle.write("\n")
+        os.replace(temp_name, path)
+    except BaseException:
+        try:
+            os.unlink(temp_name)
+        except FileNotFoundError:
+            pass
+        raise
+
+
+def should_send(previous: dict[str, object], status: str, category: str) -> bool:
+    previous_status = previous.get("status")
+    previous_category = previous.get("category")
+    if status.startswith("FAIL_") or status == "PASS_WITH_FINDINGS":
+        return previous_status != status or previous_category != category
+    return isinstance(previous_status, str) and (
+        previous_status.startswith("FAIL_") or previous_status == "PASS_WITH_FINDINGS"
+    )
+
+
+def safe_text(status: str, category: str) -> str:
+    if status.startswith("FAIL_"):
+        headline = "内部 OA 运维告警"
+    elif status == "PASS_WITH_FINDINGS":
+        headline = "内部 OA 运维提醒"
+    else:
+        headline = "内部 OA 运维恢复通知"
+    return f"{headline}\n状态：{status}\n类别：{category}\n请按恢复卡检查，不要在群内粘贴原始日志。"
+
+
+def send(webhook: str, text: str) -> None:
+    body = json.dumps({"msgtype": "text", "text": {"content": text}}).encode("utf-8")
+    capture = os.environ.get("HYB_WECOM_ALERT_TEST_CAPTURE")
+    if capture:
+        if os.environ.get("HYB_WECOM_ALERT_TEST_MODE") != "1":
+            fail("test capture is disabled")
+        capture_path = Path(capture)
+        with capture_path.open("ab") as handle:
+            handle.write(body + b"\n")
+        os.chmod(capture_path, 0o600)
+        return
+    request = urllib.request.Request(
+        webhook,
+        data=body,
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        with urllib.request.urlopen(request, timeout=10) as response:
+            result = json.loads(response.read().decode("utf-8"))
+    except Exception:
+        fail("group robot request failed")
+    if result.get("errcode") != 0:
+        fail("group robot rejected the alert")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--webhook-file", required=True, type=Path)
+    parser.add_argument("--summary-file", required=True, type=Path)
+    parser.add_argument("--state-file", required=True, type=Path)
+    args = parser.parse_args()
+
+    try:
+        status, category, _ = load_summary(args.summary_file)
+        previous = load_previous(args.state_file)
+        send_now = should_send(previous, status, category)
+        if send_now:
+            webhook = read_webhook(args.webhook_file)
+            send(webhook, safe_text(status, category))
+        atomic_write(args.state_file, {"schemaVersion": 1, "status": status, "category": category})
+    except OSError:
+        fail("local alert file operation failed")
+    print(f"[wecom-group-alert][PASS] evaluated sent={str(send_now).lower()}")
+
+
+if __name__ == "__main__":
+    main()