the-custodian/scripts/weekly_allocation_check.py

#!/usr/bin/env python3
"""CUST-WP-0071-T05 missed/failed-run check for the weekly allocation assessment.

Flags `failed` when the latest receipt failed, and `missing` when no retained
report exists for the due ISO week. The current week falls due on Monday 12:00
Europe/Berlin; before then the previous week is checked. Week labels follow
weekly_allocation_incluster.week_of (ISO week of UTC time minus 6h).

Receipts source (inactive when unset, so harmless before the CronJob runs):
- a path to a copy of the job's receipts.jsonl (failed and missing), or
- `hub`: the job's delivery messages (from custodian-allocation-review to
  the-custodian). Only retained weeks are posted, so a failed run shows as missing.
"""
from __future__ import annotations

import argparse
from datetime import datetime, timedelta, timezone
import json
from pathlib import Path
from urllib.parse import urlencode
from urllib.request import urlopen
from zoneinfo import ZoneInfo

BERLIN = ZoneInfo("Europe/Berlin")
OK = {"retained", "already-retained"}
WINDOW_END_HOUR = 6


def week_label(moment: datetime) -> str:
    year, week, _ = (moment.astimezone(timezone.utc) - timedelta(hours=WINDOW_END_HOUR)).isocalendar()
    return f"{year}-W{week:02d}"


def due_week(now: datetime) -> str:
    local = now.astimezone(BERLIN)
    monday = (local - timedelta(days=local.weekday())).replace(hour=12, minute=0, second=0, microsecond=0)
    return week_label(monday if local >= monday else monday - timedelta(days=7))


def evaluate(receipts: list[dict], now: datetime) -> dict:
    week = due_week(now)
    ordered = sorted(receipts, key=lambda r: r.get("run_at") or "")
    flags = []
    if ordered and ordered[-1].get("status") == "failed":
        flags.append({"flag": "failed", "week": ordered[-1].get("week"), "run_at": ordered[-1].get("run_at"),
                      "error": ordered[-1].get("error")})
    if not any(r.get("week") == week and r.get("status") in OK for r in receipts):
        flags.append({"flag": "missing", "week": week,
                      "detail": "no retained weekly allocation report for the due ISO week"})
    return {"status": "flagged" if flags else "ok", "due_week": week, "flags": flags,
            "recovery": "inspect CronJob custodian-assessment/custodian-allocation-review; rerun with "
                        "kubectl create job --from=cronjob/custodian-allocation-review (same-week rerun is idempotent)"
            if flags else None}


def from_file(path: Path) -> list[dict]:
    out = []
    for line in path.read_text().splitlines():
        try:
            out.append(json.loads(line))
        except ValueError:
            continue
    return out


def from_hub(hub: str, opener=urlopen) -> list[dict]:
    query = urlencode({"to_agent": "the-custodian", "from_agent": "custodian-allocation-review", "limit": 50})
    with opener(hub.rstrip("/") + "/messages/?" + query, timeout=30) as response:
        messages = json.load(response)
    out = []
    for message in messages:
        try:
            body = json.loads(message.get("body") or "{}")
        except ValueError:
            continue
        if body.get("week"):
            out.append({"week": body["week"], "status": "retained", "run_at": message.get("created_at")})
    return out


def check(source: str | None, hub: str, now: datetime | None = None, opener=urlopen) -> dict:
    if not source:
        return {"status": "inactive", "reason": "no weekly allocation receipts source configured"}
    receipts = from_hub(hub, opener) if source == "hub" else from_file(Path(source))
    return evaluate(receipts, now or datetime.now(timezone.utc))


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("--receipts", help="receipts.jsonl path or 'hub'")
    parser.add_argument("--hub", default="http://127.0.0.1:8000")
    args = parser.parse_args()
    result = check(args.receipts, args.hub)
    print(json.dumps(result, indent=2))
    return 1 if result["status"] == "flagged" else 0


if __name__ == "__main__":
    raise SystemExit(main())