chore: import upstream snapshot with attribution

scripts/catalog_models_dev.py

@@ -0,0 +1,378 @@
+#!/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()