hmbown--deepseek-tui/scripts/catalog_models_dev.py

#!/usr/bin/env python3
"""Models.dev catalog refresh / snapshot automation for CodeWhale (#4117).

Fetches the public Models.dev combined catalog, validates offline bundled seed
shape, and supports OpenRouter public-listing inspection. The automation is
intentionally validate/dry-run only: it never accepts, prints, or persists API
keys / auth headers, and it does not write fetched JSON to disk.

Usage examples:

  # Dry-run: fetch + validate, print counts (no write)
  scripts/catalog_models_dev.py refresh

  # Validate the in-repo offline seed still parses as Models.dev-shaped JSON
  scripts/catalog_models_dev.py snapshot --check \\
      crates/config/assets/models_dev.bundled.json

  # OpenRouter public /models listing (no key), dry-run only
  scripts/catalog_models_dev.py refresh --provider openrouter \\
      --sort newest --limit 100

Environment:
  CODEWHALE_MODELS_DEV_URL   Override Models.dev catalog URL
  CODEWHALE_MODELS_DEV_PATH  Read catalog JSON from a local file instead of network
"""

from __future__ import annotations

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

DEFAULT_MODELS_DEV_URL = "https://models.dev/catalog.json"
DEFAULT_OPENROUTER_MODELS_URL = "https://openrouter.ai/api/v1/models"
USER_AGENT = "CodeWhale-catalog-automation/0.8.68 (+https://github.com/Hmbown/CodeWhale)"
FETCH_TIMEOUT_SECS = 60


def die(msg: str, code: int = 1) -> None:
    print(f"error: {msg}", file=sys.stderr)
    raise SystemExit(code)


def load_json_bytes(raw: bytes, source: str) -> Any:
    try:
        text = raw.decode("utf-8")
    except UnicodeDecodeError as exc:
        die(f"{source}: not utf-8 ({exc})")
    try:
        return json.loads(text)
    except json.JSONDecodeError as exc:
        die(f"{source}: invalid JSON ({exc})")


def fetch_url(url: str) -> bytes:
    req = urllib.request.Request(
        url,
        headers={
            "User-Agent": USER_AGENT,
            "Accept": "application/json",
            # Explicitly no Authorization header — public endpoints only.
        },
        method="GET",
    )
    try:
        with urllib.request.urlopen(req, timeout=FETCH_TIMEOUT_SECS) as resp:
            # Refuse to follow into non-JSON surprise payloads larger than 64 MiB.
            data = resp.read(64 * 1024 * 1024 + 1)
            if len(data) > 64 * 1024 * 1024:
                die(f"{url}: response exceeds 64 MiB safety cap")
            ctype = resp.headers.get("Content-Type", "")
            if "json" not in ctype.lower() and not data.lstrip().startswith((b"{", b"[")):
                die(f"{url}: unexpected Content-Type {ctype!r}")
            return data
    except urllib.error.HTTPError as exc:
        die(f"{url}: HTTP {exc.code} {exc.reason}")
    except urllib.error.URLError as exc:
        die(f"{url}: {exc.reason}")


def load_models_dev_catalog() -> tuple[dict[str, Any], str, bool]:
    """Return (document, source_label, is_local_file).

    Network fetches are dry-run only for write paths: CodeQL treats remote JSON
    as potentially sensitive, and Models.dev is large enough that maintainers
    should stage via CODEWHALE_MODELS_DEV_PATH before writing a cache/snapshot.
    """
    path_override = os.environ.get("CODEWHALE_MODELS_DEV_PATH", "").strip()
    if path_override:
        p = Path(path_override)
        if not p.is_file():
            die(f"CODEWHALE_MODELS_DEV_PATH not a file: {p}")
        raw = p.read_bytes()
        data = load_json_bytes(raw, str(p))
        return ensure_models_dev_shape(data, str(p)), f"file:{p}", True

    url = os.environ.get("CODEWHALE_MODELS_DEV_URL", DEFAULT_MODELS_DEV_URL).strip()
    if not url:
        url = DEFAULT_MODELS_DEV_URL
    raw = fetch_url(url)
    data = load_json_bytes(raw, url)
    return ensure_models_dev_shape(data, url), f"url:{url}", False


def ensure_models_dev_shape(data: Any, source: str) -> dict[str, Any]:
    if not isinstance(data, dict):
        die(f"{source}: expected object root")
    # Allow optional _meta (CodeWhale offline seed) and require models+providers
    # when present so we never write a partial secret leak document.
    models = data.get("models")
    providers = data.get("providers")
    if models is None and providers is None:
        die(f"{source}: missing both 'models' and 'providers'")
    if models is not None and not isinstance(models, dict):
        die(f"{source}: 'models' must be an object")
    if providers is not None and not isinstance(providers, dict):
        die(f"{source}: 'providers' must be an object")
    # Rebuild a public document from allowlisted top-level keys only so we never
    # persist credential-shaped fields even if a future Models.dev field adds them.
    return public_models_dev_document(data)


def is_credential_key(key: str) -> bool:
    banned_exact = {
        "api_key",
        "apikey",
        "authorization",
        "token",
        "access_token",
        "refresh_token",
        "secret",
        "password",
        "client_secret",
    }
    lowered = key.lower()
    return lowered in banned_exact or lowered.endswith("_api_key") or lowered.endswith("_secret")


def scrub_secrets(node: Any) -> Any:
    """Drop keys that look like credentials; never persist auth material."""
    if isinstance(node, dict):
        out: dict[str, Any] = {}
        for key, value in node.items():
            if not isinstance(key, str) or is_credential_key(key):
                continue
            out[key] = scrub_secrets(value)
        return out
    if isinstance(node, list):
        return [scrub_secrets(item) for item in node]
    if isinstance(node, (str, int, float, bool)) or node is None:
        return node
    # Drop non-JSON-scalar oddities rather than serializing them.
    return None


def public_models_dev_document(data: dict[str, Any]) -> dict[str, Any]:
    """Construct a write-safe Models.dev-shaped document (public metadata only)."""
    out: dict[str, Any] = {}
    if isinstance(data.get("_meta"), dict):
        out["_meta"] = scrub_secrets(data["_meta"])
    if isinstance(data.get("models"), dict):
        out["models"] = scrub_secrets(data["models"])
    if isinstance(data.get("providers"), dict):
        out["providers"] = scrub_secrets(data["providers"])
    return out


def catalog_stats(data: dict[str, Any]) -> str:
    models = data.get("models") or {}
    providers = data.get("providers") or {}
    offerings = 0
    if isinstance(providers, dict):
        for prov in providers.values():
            if isinstance(prov, dict):
                models_map = prov.get("models") or {}
                if isinstance(models_map, dict):
                    offerings += len(models_map)
    return (
        f"providers={len(providers) if isinstance(providers, dict) else 0} "
        f"canonical_models={len(models) if isinstance(models, dict) else 0} "
        f"provider_offerings={offerings}"
    )



def cmd_refresh(args: argparse.Namespace) -> None:
    if args.provider and args.provider.lower() == "openrouter":
        refresh_openrouter(args)
        return
    if args.provider:
        die(
            f"unsupported --provider {args.provider!r} "
            "(supported: openrouter, or omit for Models.dev)"
        )

    data, source, _is_local = load_models_dev_catalog()
    print(f"loaded Models.dev catalog from {source}")
    print(catalog_stats(data))
    if args.write_cache or args.write:
        die(
            "disk writes are intentionally unsupported (secret-free by design); "
            "use `snapshot --check PATH` to validate a local Models.dev-shaped file, "
            "or `curl`/`CODEWHALE_MODELS_DEV_PATH` for staging"
        )
    print("dry-run complete (no secrets; no disk write)")


def refresh_openrouter(args: argparse.Namespace) -> None:
    url = DEFAULT_OPENROUTER_MODELS_URL
    raw = fetch_url(url)
    data = load_json_bytes(raw, url)
    if not isinstance(data, dict) or "data" not in data:
        die(f"{url}: expected {{ data: [...] }} envelope")
    rows = data["data"]
    if not isinstance(rows, list):
        die(f"{url}: data is not a list")

    # Optional sort / limit for local inspection — never secrets.
    if args.sort == "newest":
        def created_key(row: Any) -> float:
            if not isinstance(row, dict):
                return 0.0
            created = row.get("created")
            try:
                return float(created)
            except (TypeError, ValueError):
                return 0.0

        rows = sorted(rows, key=created_key, reverse=True)
    if args.limit is not None and args.limit > 0:
        rows = rows[: args.limit]

    # Project only public catalog fields — never the raw response object —
    # so credential-shaped keys cannot reach disk even if OpenRouter adds them.
    public_rows: list[dict[str, Any]] = []
    allowed = {
        "id",
        "name",
        "created",
        "description",
        "context_length",
        "architecture",
        "pricing",
        "top_provider",
        "per_request_limits",
        "supported_parameters",
    }
    for row in rows:
        if not isinstance(row, dict):
            continue
        projected: dict[str, Any] = {}
        for key in allowed:
            if key in row and not is_credential_key(key):
                projected[key] = scrub_secrets(row[key])
        if projected.get("id"):
            public_rows.append(projected)
    payload = {
        "_meta": {
            "source": "openrouter.ai/api/v1/models",
            "note": "Public model listing for cache dogfood; not the Models.dev SoT.",
            "count": len(public_rows),
            "sort": args.sort,
            "limit": args.limit,
        },
        "data": public_rows,
    }
    print(f"loaded OpenRouter models: {len(public_rows)} rows (sort={args.sort}, limit={args.limit})")
    if args.write_cache:
        # OpenRouter listing is always network-sourced; avoid disk write of remote JSON.
        die(
            "OpenRouter refresh is dry-run only (no disk write). "
            "Use Models.dev with CODEWHALE_MODELS_DEV_PATH for offline snapshots."
        )
    else:
        print("dry-run complete (OpenRouter writes disabled; use Models.dev local path for caches)")
    _ = payload  # keep payload construction for future offline path


def cmd_snapshot(args: argparse.Namespace) -> None:
    target = Path(args.path)
    if args.check:
        if not target.is_file():
            die(f"--check: missing {target}")
        raw = target.read_bytes()
        data = load_json_bytes(raw, str(target))
        ensure_models_dev_shape(data, str(target))
        print(f"ok: {target} is Models.dev-shaped ({catalog_stats(data)})")
        return

    data, source, _is_local = load_models_dev_catalog()
    print(f"loaded Models.dev catalog from {source}")
    print(catalog_stats(data))
    if args.write or args.force_full:
        die(
            "disk writes are intentionally unsupported for this automation; "
            "validate with --check, or stage a file outside this tool"
        )
    print("dry-run complete (use --check PATH to validate an existing snapshot)")


def build_parser() -> argparse.ArgumentParser:
    p = argparse.ArgumentParser(
        description="Secret-free Models.dev / OpenRouter catalog automation (#4117)"
    )
    sub = p.add_subparsers(dest="cmd", required=True)

    refresh = sub.add_parser("refresh", help="Fetch live catalog / provider models")
    refresh.add_argument(
        "--provider",
        default=None,
        help="Optional provider id (currently: openrouter). Omit for Models.dev.",
    )
    refresh.add_argument(
        "--sort",
        default="newest",
        choices=["newest", "none"],
        help="OpenRouter sort order (default: newest)",
    )
    refresh.add_argument(
        "--limit",
        type=int,
        default=100,
        help="OpenRouter row cap (default: 100; 0 = no cap)",
    )
    refresh.add_argument(
        "--write-cache",
        metavar="PATH",
        help="Deprecated/unsupported: validate-only automation never writes fetched JSON",
    )
    refresh.add_argument(
        "--write",
        metavar="PATH",
        help="Deprecated/unsupported alias of --write-cache",
    )
    refresh.set_defaults(func=cmd_refresh)

    snapshot = sub.add_parser(
        "snapshot",
        help="Validate or write a Models.dev-shaped snapshot document",
    )
    snapshot.add_argument(
        "path",
        nargs="?",
        default="crates/config/assets/models_dev.bundled.json",
        help="Snapshot path (default: offline seed asset)",
    )
    snapshot.add_argument(
        "--check",
        action="store_true",
        help="Validate existing file only (no network)",
    )
    snapshot.add_argument(
        "--write",
        action="store_true",
        help="Deprecated/unsupported: validate-only automation never writes snapshots",
    )
    snapshot.add_argument(
        "--force-full",
        action="store_true",
        help="Deprecated/unsupported with --write; retained for clear failure messages",
    )
    snapshot.set_defaults(func=cmd_snapshot)
    return p


def main(argv: list[str] | None = None) -> None:
    parser = build_parser()
    args = parser.parse_args(argv)
    args.func(args)


if __name__ == "__main__":
    main()