state-hub/scripts/capture_legacy_meter_evidence.py

#!/usr/bin/env python3
"""Capture legacy-meter review payload as dated evidence JSON.

Usage:
    python scripts/capture_legacy_meter_evidence.py [--hours 8] [--days N] [--api-base URL] [--dry-run]

Default review window is 8 hours (post-deploy monitoring). Pass --days for the
weekly activity-core cadence (typically 7).

Writes:
    docs/evidence/legacy-meter-weekly-review-YYYYMMDD.json

When the hub is reachable, optionally marks interfaces retired via PATCH when
--retire-keys is supplied (comma-separated legacy-meter keys).
"""
from __future__ import annotations

import argparse
import datetime
import json
import os
import sys
import urllib.error
import urllib.request
from pathlib import Path

SCRIPT_DIR = Path(__file__).parent
REPO_ROOT = SCRIPT_DIR.parent
EVIDENCE_DIR = REPO_ROOT / "docs" / "evidence"


def _api_get(base: str, path: str, timeout: float = 30.0) -> dict:
    url = f"{base.rstrip('/')}{path}"
    with urllib.request.urlopen(url, timeout=timeout) as resp:
        return json.loads(resp.read())


def _api_patch(base: str, interface_id: str, body: dict, timeout: float = 30.0) -> dict:
    url = f"{base.rstrip('/')}/legacy-meter/interfaces/{interface_id}"
    data = json.dumps(body).encode()
    req = urllib.request.Request(
        url,
        data=data,
        method="PATCH",
        headers={"Content-Type": "application/json"},
    )
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        return json.loads(resp.read())


def _retire_interfaces(base: str, keys: list[str]) -> list[dict]:
    interfaces = _api_get(base, "/legacy-meter/interfaces")
    by_key = {item["interface_key"]: item for item in interfaces}
    retired: list[dict] = []
    for key in keys:
        interface = by_key.get(key)
        if interface is None:
            print(f"skip retire: {key} not registered", file=sys.stderr)
            continue
        if interface.get("status") == "retired":
            print(f"skip retire: {key} already retired")
            retired.append(interface)
            continue
        updated = _api_patch(
            base,
            interface["id"],
            {"status": "retired", "notes": "Retired by STATE-WP-0069 closeout"},
        )
        print(f"retired: {key}")
        retired.append(updated)
    return retired


DEFAULT_REVIEW_DAYS = 7


def _review_query(*, days: int | None, hours: int | None) -> str:
    """Build the review-window query.

    Defaults to a 7-day window to match the ``weekly-review`` endpoint and the
    ``cadence: weekly`` label in the payload. An hours-scoped window only
    samples part of each day, so an interface called outside that band reads as
    unused and is falsely reported as a retirement candidate.
    """
    if hours is not None:
        return f"hours={hours}"
    return f"days={days or DEFAULT_REVIEW_DAYS}"


def capture(
    *,
    api_base: str,
    days: int | None,
    hours: int | None,
    retire_keys: list[str],
    dry_run: bool,
) -> Path:
    query = _review_query(days=days, hours=hours)
    review = _api_get(api_base, f"/legacy-meter/weekly-review?{query}")
    retired: list[dict] = []
    if retire_keys and not dry_run:
        retired = _retire_interfaces(api_base, retire_keys)
        review = _api_get(api_base, f"/legacy-meter/weekly-review?{query}")

    today = datetime.date.today().strftime("%Y%m%d")
    out_path = EVIDENCE_DIR / f"legacy-meter-weekly-review-{today}.json"
    payload: dict = {
        "captured_at": datetime.datetime.now(datetime.timezone.utc).isoformat(),
        "api_base": api_base,
        "workplan": "STATE-WP-0070",
        "retired_interfaces": retired,
        "weekly_review": review,
    }
    if hours is not None:
        payload["hours"] = hours
    else:
        payload["days"] = days or DEFAULT_REVIEW_DAYS

    if dry_run:
        print(json.dumps(payload, indent=2))
        return out_path

    EVIDENCE_DIR.mkdir(parents=True, exist_ok=True)
    out_path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8")
    candidates = len(review.get("retirement_candidates", []))
    interfaces = len(review.get("interfaces", []))
    window = query.replace("=", " ")
    print(f"wrote {out_path} ({interfaces} interfaces, {candidates} retirement candidates, {window})")
    return out_path


def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--hours",
        type=int,
        default=None,
        help="Review window in hours; only samples part of each day — use for "
        "spot checks, never as retirement evidence",
    )
    parser.add_argument(
        "--days",
        type=int,
        default=None,
        help=f"Review window in days (default: {DEFAULT_REVIEW_DAYS}, weekly cadence)",
    )
    parser.add_argument(
        "--api-base",
        default=os.environ.get("API_BASE", "http://127.0.0.1:8000"),
    )
    parser.add_argument(
        "--retire-keys",
        default="event_subject:org.statehub.workstream.completed",
        help="Comma-separated legacy-meter keys to mark retired before capture",
    )
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()

    retire_keys = [key.strip() for key in args.retire_keys.split(",") if key.strip()]
    try:
        capture(
            api_base=args.api_base,
            days=args.days,
            hours=args.hours,
            retire_keys=retire_keys,
            dry_run=args.dry_run,
        )
    except urllib.error.URLError as exc:
        print(f"Error: could not reach State Hub API at {args.api_base}: {exc}", file=sys.stderr)
        sys.exit(1)


if __name__ == "__main__":
    main()