refresh_active_scenarios.py

#!/usr/bin/env python3
"""
refresh_active_scenarios.py - regenerate the curated/public scenario list.

The source of truth is the live database's `scenario.is_active` column,
read here via GET https://api.coeqwal.org/api/scenarios. To change which
scenarios are active, run `etl/ingestion/tools/set_scenario_active.py`
(which flips the DB and chains this refresh). This script is read-only
against the API and never mutates the DB itself.

Pulls every short_code where `is_active` is true, sorts them, then:

  1. Rewrites the inline list in the top-level `README.md` between the markers:

       <!-- ACTIVE_SCENARIOS:BEGIN -->
       ...
       <!-- ACTIVE_SCENARIOS:END -->

  2. Regenerates `etl/common/active_scenarios.py` so Python consumers can
     `from etl.common.active_scenarios import ACTIVE_SCENARIOS` without
     reaching out to the API at runtime.

If the README markers are missing the script errors out rather than
guessing where to insert. It never edits anything else in the README.

Usage:
  python etl/ingestion/tools/refresh_active_scenarios.py
  python etl/ingestion/tools/refresh_active_scenarios.py --api-url https://api.coeqwal.org
  python etl/ingestion/tools/refresh_active_scenarios.py --dry-run
"""

from __future__ import annotations

import argparse
import json
import logging
import urllib.error
import urllib.request
from datetime import datetime, timezone
from pathlib import Path
from typing import List

DEFAULT_API_URL = "https://api.coeqwal.org"
ETL_DIR = Path(__file__).resolve().parents[2]
REPO_ROOT = ETL_DIR.parent
README_PATH = REPO_ROOT / "README.md"
ACTIVE_SCENARIOS_PY = ETL_DIR / "common" / "active_scenarios.py"

BEGIN_MARKER = "<!-- ACTIVE_SCENARIOS:BEGIN -->"
END_MARKER = "<!-- ACTIVE_SCENARIOS:END -->"

log = logging.getLogger("refresh_active_scenarios")


def _fetch_scenarios(api_url: str) -> List[dict]:
    """GET <api>/api/scenarios and return the parsed JSON list."""
    url = api_url.rstrip("/") + "/api/scenarios"
    log.info("GET %s ...", url)
    req = urllib.request.Request(url, headers={"Accept": "application/json"})
    try:
        with urllib.request.urlopen(req, timeout=30) as resp:
            body = resp.read()
    except urllib.error.URLError as e:
        raise SystemExit(f"\nFailed to reach API at {url}: {e}\n")
    try:
        data = json.loads(body)
    except json.JSONDecodeError as e:
        raise SystemExit(f"\nAPI returned non-JSON: {e}\n")
    if not isinstance(data, list):
        raise SystemExit(f"\nExpected a list from {url}, got {type(data).__name__}\n")
    return data


def _build_block(short_codes: List[str], api_url: str) -> str:
    """Render the new ACTIVE_SCENARIOS block."""
    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    inline = ", ".join(short_codes) if short_codes else "(none)"
    return (
        f"{BEGIN_MARKER}\n"
        f"\n"
        f"**Active scenarios ({len(short_codes)})**: {inline}\n"
        f"\n"
        f"_Last refreshed {now} from `{api_url}/api/scenarios`. "
        f"Regenerate with `python etl/ingestion/tools/refresh_active_scenarios.py`._\n"
        f"\n"
        f"{END_MARKER}"
    )


def _render_active_scenarios_py(short_codes: List[str], api_url: str) -> str:
    """Render the contents of etl/common/active_scenarios.py."""
    now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
    api_endpoint = api_url.rstrip("/") + "/api/scenarios"
    if short_codes:
        body = "\n".join(f'    "{code}",' for code in short_codes)
        literal = "frozenset({\n" + body + "\n})"
    else:
        literal = "frozenset()"
    return (
        '"""Curated/public scenarios that are live on the production website.\n'
        "\n"
        f"Auto-generated by `etl/ingestion/tools/refresh_active_scenarios.py`\n"
        f"on {now} from `{api_endpoint}` (rows where `is_active` is true).\n"
        "Do not edit by hand. Re-run the refresh script to regenerate.\n"
        "\n"
        "To change which scenarios are in this set, run\n"
        "`etl/ingestion/tools/set_scenario_active.py --activate sXXX` (or\n"
        "`--deactivate sXXX`), which flips `scenario.is_active` in the DB\n"
        "and chains the refresh.\n"
        "\n"
        "Use this set for anything that must match what the website serves:\n"
        "tier uploads, API verification, tier verification. For the larger\n"
        "set of scenarios that have ETL outputs in S3 (a superset that\n"
        "includes drafts and inactive runs), import `ETL_SCENARIOS` from\n"
        "`etl.common.etl_scenarios` instead.\n"
        '"""\n'
        "\n"
        "from __future__ import annotations\n"
        "\n"
        f"ACTIVE_SCENARIOS: frozenset[str] = {literal}\n"
    )


def _replace_block(content: str, new_block: str) -> str:
    """Replace the existing ACTIVE_SCENARIOS block with new_block."""
    begin_idx = content.find(BEGIN_MARKER)
    end_idx = content.find(END_MARKER)
    if begin_idx == -1 or end_idx == -1:
        raise SystemExit(
            f"\nMarkers not found in {README_PATH}.\n"
            f"Expected both '{BEGIN_MARKER}' and '{END_MARKER}'.\n"
            f"Add an empty block to the README and re-run this script.\n"
        )
    if end_idx < begin_idx:
        raise SystemExit(
            f"\nMarker order is wrong in {README_PATH} (END before BEGIN). Fix manually.\n"
        )
    end_full = end_idx + len(END_MARKER)
    return content[:begin_idx] + new_block + content[end_full:]


def main():
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s %(levelname)-5s %(message)s",
        datefmt="%Y-%m-%dT%H:%M:%SZ",
    )
    parser = argparse.ArgumentParser(
        description="Refresh the curated scenario list (top-level README block + etl/common/active_scenarios.py) from the live API."
    )
    parser.add_argument("--api-url", default=DEFAULT_API_URL,
                        help=f"API base URL (default: {DEFAULT_API_URL})")
    parser.add_argument("--readme", default=str(README_PATH),
                        help=f"Path to the README to edit (default: {README_PATH})")
    parser.add_argument("--active-py", default=str(ACTIVE_SCENARIOS_PY),
                        help=f"Path to write the ACTIVE_SCENARIOS Python module (default: {ACTIVE_SCENARIOS_PY})")
    parser.add_argument("--dry-run", action="store_true",
                        help="Print the new block and the generated module to stdout, don't write.")
    args = parser.parse_args()

    scenarios = _fetch_scenarios(args.api_url)
    active = sorted(
        s["short_code"]
        for s in scenarios
        if isinstance(s, dict) and s.get("is_active") and s.get("short_code")
    )
    log.info("Fetched %d scenarios; %d are active", len(scenarios), len(active))

    new_block = _build_block(active, args.api_url)
    new_py = _render_active_scenarios_py(active, args.api_url)

    readme = Path(args.readme)
    if not readme.exists():
        raise SystemExit(f"\nREADME not found at {readme}\n")
    current_readme = readme.read_text()
    new_readme = _replace_block(current_readme, new_block)

    active_py = Path(args.active_py)
    current_py = active_py.read_text() if active_py.exists() else ""

    if args.dry_run:
        print("--- new README block ---")
        print(new_block)
        print("\n--- new etl/common/active_scenarios.py ---")
        print(new_py)
        print("\n--- diff summary ---")
        print(f"README: {'(no change)' if current_readme == new_readme else '(would change)'}")
        print(f"active_scenarios.py: {'(no change)' if current_py == new_py else '(would change)'}")
        return

    readme_changed = current_readme != new_readme
    py_changed = current_py != new_py

    if readme_changed:
        readme.write_text(new_readme)
        log.info("Updated %s with %d active scenarios", readme, len(active))
    else:
        log.info("README already up to date (%d active scenarios)", len(active))

    if py_changed:
        active_py.parent.mkdir(parents=True, exist_ok=True)
        active_py.write_text(new_py)
        log.info("Wrote %s (%d scenarios)", active_py, len(active))
    else:
        log.info("etl/common/active_scenarios.py already up to date")


if __name__ == "__main__":
    main()