| 1 | #!/usr/bin/env python3 |
| 2 | """Models.dev catalog refresh / snapshot automation for CodeWhale (#4117). |
| 3 | |
| 4 | Fetches the public Models.dev combined catalog, validates offline bundled seed |
| 5 | shape, and supports OpenRouter public-listing inspection. The automation is |
| 6 | intentionally validate/dry-run only: it never accepts, prints, or persists API |
| 7 | keys / auth headers, and it does not write fetched JSON to disk. |
| 8 | |
| 9 | Usage examples: |
| 10 | |
| 11 | # Dry-run: fetch + validate, print counts (no write) |
| 12 | scripts/catalog_models_dev.py refresh |
| 13 | |
| 14 | # Validate the in-repo offline seed still parses as Models.dev-shaped JSON |
| 15 | scripts/catalog_models_dev.py snapshot --check \\ |
| 16 | crates/config/assets/models_dev.bundled.json |
| 17 | |
| 18 | # OpenRouter public /models listing (no key), dry-run only |
| 19 | scripts/catalog_models_dev.py refresh --provider openrouter \\ |
| 20 | --sort newest --limit 100 |
| 21 | |
| 22 | Environment: |
| 23 | CODEWHALE_MODELS_DEV_URL Override Models.dev catalog URL |
| 24 | CODEWHALE_MODELS_DEV_PATH Read catalog JSON from a local file instead of network |
| 25 | """ |
| 26 | |
| 27 | from __future__ import annotations |
| 28 | |
| 29 | import argparse |
| 30 | import json |
| 31 | import os |
| 32 | import sys |
| 33 | import urllib.error |
| 34 | import urllib.request |
| 35 | from pathlib import Path |
| 36 | from typing import Any |
| 37 | |
| 38 | DEFAULT_MODELS_DEV_URL = "https://models.dev/catalog.json" |
| 39 | DEFAULT_OPENROUTER_MODELS_URL = "https://openrouter.ai/api/v1/models" |
| 40 | USER_AGENT = "CodeWhale-catalog-automation/0.9.0 (+https://github.com/Hmbown/CodeWhale)" |
| 41 | FETCH_TIMEOUT_SECS = 60 |
| 42 | |
| 43 | |
| 44 | def die(msg: str, code: int = 1) -> None: |
| 45 | print(f"error: {msg}", file=sys.stderr) |
| 46 | raise SystemExit(code) |
| 47 | |
| 48 | |
| 49 | def load_json_bytes(raw: bytes, source: str) -> Any: |
| 50 | try: |
| 51 | text = raw.decode("utf-8") |
| 52 | except UnicodeDecodeError as exc: |
| 53 | die(f"{source}: not utf-8 ({exc})") |
| 54 | try: |
| 55 | return json.loads(text) |
| 56 | except json.JSONDecodeError as exc: |
| 57 | die(f"{source}: invalid JSON ({exc})") |
| 58 | |
| 59 | |
| 60 | def fetch_url(url: str) -> bytes: |
| 61 | req = urllib.request.Request( |
| 62 | url, |
| 63 | headers={ |
| 64 | "User-Agent": USER_AGENT, |
| 65 | "Accept": "application/json", |
| 66 | # Explicitly no Authorization header — public endpoints only. |
| 67 | }, |
| 68 | method="GET", |
| 69 | ) |
| 70 | try: |
| 71 | with urllib.request.urlopen(req, timeout=FETCH_TIMEOUT_SECS) as resp: |
| 72 | # Refuse to follow into non-JSON surprise payloads larger than 64 MiB. |
| 73 | data = resp.read(64 * 1024 * 1024 + 1) |
| 74 | if len(data) > 64 * 1024 * 1024: |
| 75 | die(f"{url}: response exceeds 64 MiB safety cap") |
| 76 | ctype = resp.headers.get("Content-Type", "") |
| 77 | if "json" not in ctype.lower() and not data.lstrip().startswith((b"{", b"[")): |
| 78 | die(f"{url}: unexpected Content-Type {ctype!r}") |
| 79 | return data |
| 80 | except urllib.error.HTTPError as exc: |
| 81 | die(f"{url}: HTTP {exc.code} {exc.reason}") |
| 82 | except urllib.error.URLError as exc: |
| 83 | die(f"{url}: {exc.reason}") |
| 84 | |
| 85 | |
| 86 | def load_models_dev_catalog() -> tuple[dict[str, Any], str, bool]: |
| 87 | """Return (document, source_label, is_local_file). |
| 88 | |
| 89 | Network fetches are dry-run only for write paths: CodeQL treats remote JSON |
| 90 | as potentially sensitive, and Models.dev is large enough that maintainers |
| 91 | should stage via CODEWHALE_MODELS_DEV_PATH before writing a cache/snapshot. |
| 92 | """ |
| 93 | path_override = os.environ.get("CODEWHALE_MODELS_DEV_PATH", "").strip() |
| 94 | if path_override: |
| 95 | p = Path(path_override) |
| 96 | if not p.is_file(): |
| 97 | die(f"CODEWHALE_MODELS_DEV_PATH not a file: {p}") |
| 98 | raw = p.read_bytes() |
| 99 | data = load_json_bytes(raw, str(p)) |
| 100 | return ensure_models_dev_shape(data, str(p)), f"file:{p}", True |
| 101 | |
| 102 | url = os.environ.get("CODEWHALE_MODELS_DEV_URL", DEFAULT_MODELS_DEV_URL).strip() |
| 103 | if not url: |
| 104 | url = DEFAULT_MODELS_DEV_URL |
| 105 | raw = fetch_url(url) |
| 106 | data = load_json_bytes(raw, url) |
| 107 | return ensure_models_dev_shape(data, url), f"url:{url}", False |
| 108 | |
| 109 | |
| 110 | def ensure_models_dev_shape(data: Any, source: str) -> dict[str, Any]: |
| 111 | if not isinstance(data, dict): |
| 112 | die(f"{source}: expected object root") |
| 113 | # Allow optional _meta (CodeWhale offline seed) and require models+providers |
| 114 | # when present so we never write a partial secret leak document. |
| 115 | models = data.get("models") |
| 116 | providers = data.get("providers") |
| 117 | if models is None and providers is None: |
| 118 | die(f"{source}: missing both 'models' and 'providers'") |
| 119 | if models is not None and not isinstance(models, dict): |
| 120 | die(f"{source}: 'models' must be an object") |
| 121 | if providers is not None and not isinstance(providers, dict): |
| 122 | die(f"{source}: 'providers' must be an object") |
| 123 | # Rebuild a public document from allowlisted top-level keys only so we never |
| 124 | # persist credential-shaped fields even if a future Models.dev field adds them. |
| 125 | return public_models_dev_document(data) |
| 126 | |
| 127 | |
| 128 | def is_credential_key(key: str) -> bool: |
| 129 | banned_exact = { |
| 130 | "api_key", |
| 131 | "apikey", |
| 132 | "authorization", |
| 133 | "token", |
| 134 | "access_token", |
| 135 | "refresh_token", |
| 136 | "secret", |
| 137 | "password", |
| 138 | "client_secret", |
| 139 | } |
| 140 | lowered = key.lower() |
| 141 | return lowered in banned_exact or lowered.endswith("_api_key") or lowered.endswith("_secret") |
| 142 | |
| 143 | |
| 144 | def scrub_secrets(node: Any) -> Any: |
| 145 | """Drop keys that look like credentials; never persist auth material.""" |
| 146 | if isinstance(node, dict): |
| 147 | out: dict[str, Any] = {} |
| 148 | for key, value in node.items(): |
| 149 | if not isinstance(key, str) or is_credential_key(key): |
| 150 | continue |
| 151 | out[key] = scrub_secrets(value) |
| 152 | return out |
| 153 | if isinstance(node, list): |
| 154 | return [scrub_secrets(item) for item in node] |
| 155 | if isinstance(node, (str, int, float, bool)) or node is None: |
| 156 | return node |
| 157 | # Drop non-JSON-scalar oddities rather than serializing them. |
| 158 | return None |
| 159 | |
| 160 | |
| 161 | def public_models_dev_document(data: dict[str, Any]) -> dict[str, Any]: |
| 162 | """Construct a write-safe Models.dev-shaped document (public metadata only).""" |
| 163 | out: dict[str, Any] = {} |
| 164 | if isinstance(data.get("_meta"), dict): |
| 165 | out["_meta"] = scrub_secrets(data["_meta"]) |
| 166 | if isinstance(data.get("models"), dict): |
| 167 | out["models"] = scrub_secrets(data["models"]) |
| 168 | if isinstance(data.get("providers"), dict): |
| 169 | out["providers"] = scrub_secrets(data["providers"]) |
| 170 | return out |
| 171 | |
| 172 | |
| 173 | def catalog_stats(data: dict[str, Any]) -> str: |
| 174 | models = data.get("models") or {} |
| 175 | providers = data.get("providers") or {} |
| 176 | offerings = 0 |
| 177 | if isinstance(providers, dict): |
| 178 | for prov in providers.values(): |
| 179 | if isinstance(prov, dict): |
| 180 | models_map = prov.get("models") or {} |
| 181 | if isinstance(models_map, dict): |
| 182 | offerings += len(models_map) |
| 183 | return ( |
| 184 | f"providers={len(providers) if isinstance(providers, dict) else 0} " |
| 185 | f"canonical_models={len(models) if isinstance(models, dict) else 0} " |
| 186 | f"provider_offerings={offerings}" |
| 187 | ) |
| 188 | |
| 189 | |
| 190 | |
| 191 | def cmd_refresh(args: argparse.Namespace) -> None: |
| 192 | if args.provider and args.provider.lower() == "openrouter": |
| 193 | refresh_openrouter(args) |
| 194 | return |
| 195 | if args.provider: |
| 196 | die( |
| 197 | f"unsupported --provider {args.provider!r} " |
| 198 | "(supported: openrouter, or omit for Models.dev)" |
| 199 | ) |
| 200 | |
| 201 | data, source, _is_local = load_models_dev_catalog() |
| 202 | print(f"loaded Models.dev catalog from {source}") |
| 203 | print(catalog_stats(data)) |
| 204 | if args.write_cache or args.write: |
| 205 | die( |
| 206 | "disk writes are intentionally unsupported (secret-free by design); " |
| 207 | "use `snapshot --check PATH` to validate a local Models.dev-shaped file, " |
| 208 | "or `curl`/`CODEWHALE_MODELS_DEV_PATH` for staging" |
| 209 | ) |
| 210 | print("dry-run complete (no secrets; no disk write)") |
| 211 | |
| 212 | |
| 213 | def refresh_openrouter(args: argparse.Namespace) -> None: |
| 214 | url = DEFAULT_OPENROUTER_MODELS_URL |
| 215 | raw = fetch_url(url) |
| 216 | data = load_json_bytes(raw, url) |
| 217 | if not isinstance(data, dict) or "data" not in data: |
| 218 | die(f"{url}: expected {{ data: [...] }} envelope") |
| 219 | rows = data["data"] |
| 220 | if not isinstance(rows, list): |
| 221 | die(f"{url}: data is not a list") |
| 222 | |
| 223 | # Optional sort / limit for local inspection — never secrets. |
| 224 | if args.sort == "newest": |
| 225 | def created_key(row: Any) -> float: |
| 226 | if not isinstance(row, dict): |
| 227 | return 0.0 |
| 228 | created = row.get("created") |
| 229 | try: |
| 230 | return float(created) |
| 231 | except (TypeError, ValueError): |
| 232 | return 0.0 |
| 233 | |
| 234 | rows = sorted(rows, key=created_key, reverse=True) |
| 235 | if args.limit is not None and args.limit > 0: |
| 236 | rows = rows[: args.limit] |
| 237 | |
| 238 | # Project only public catalog fields — never the raw response object — |
| 239 | # so credential-shaped keys cannot reach disk even if OpenRouter adds them. |
| 240 | public_rows: list[dict[str, Any]] = [] |
| 241 | allowed = { |
| 242 | "id", |
| 243 | "name", |
| 244 | "created", |
| 245 | "description", |
| 246 | "context_length", |
| 247 | "architecture", |
| 248 | "pricing", |
| 249 | "top_provider", |
| 250 | "per_request_limits", |
| 251 | "supported_parameters", |
| 252 | } |
| 253 | for row in rows: |
| 254 | if not isinstance(row, dict): |
| 255 | continue |
| 256 | projected: dict[str, Any] = {} |
| 257 | for key in allowed: |
| 258 | if key in row and not is_credential_key(key): |
| 259 | projected[key] = scrub_secrets(row[key]) |
| 260 | if projected.get("id"): |
| 261 | public_rows.append(projected) |
| 262 | payload = { |
| 263 | "_meta": { |
| 264 | "source": "openrouter.ai/api/v1/models", |
| 265 | "note": "Public model listing for cache dogfood; not the Models.dev SoT.", |
| 266 | "count": len(public_rows), |
| 267 | "sort": args.sort, |
| 268 | "limit": args.limit, |
| 269 | }, |
| 270 | "data": public_rows, |
| 271 | } |
| 272 | print(f"loaded OpenRouter models: {len(public_rows)} rows (sort={args.sort}, limit={args.limit})") |
| 273 | if args.write_cache: |
| 274 | # OpenRouter listing is always network-sourced; avoid disk write of remote JSON. |
| 275 | die( |
| 276 | "OpenRouter refresh is dry-run only (no disk write). " |
| 277 | "Use Models.dev with CODEWHALE_MODELS_DEV_PATH for offline snapshots." |
| 278 | ) |
| 279 | else: |
| 280 | print("dry-run complete (OpenRouter writes disabled; use Models.dev local path for caches)") |
| 281 | _ = payload # keep payload construction for future offline path |
| 282 | |
| 283 | |
| 284 | def cmd_snapshot(args: argparse.Namespace) -> None: |
| 285 | target = Path(args.path) |
| 286 | if args.check: |
| 287 | if not target.is_file(): |
| 288 | die(f"--check: missing {target}") |
| 289 | raw = target.read_bytes() |
| 290 | data = load_json_bytes(raw, str(target)) |
| 291 | ensure_models_dev_shape(data, str(target)) |
| 292 | print(f"ok: {target} is Models.dev-shaped ({catalog_stats(data)})") |
| 293 | return |
| 294 | |
| 295 | data, source, _is_local = load_models_dev_catalog() |
| 296 | print(f"loaded Models.dev catalog from {source}") |
| 297 | print(catalog_stats(data)) |
| 298 | if args.write or args.force_full: |
| 299 | die( |
| 300 | "disk writes are intentionally unsupported for this automation; " |
| 301 | "validate with --check, or stage a file outside this tool" |
| 302 | ) |
| 303 | print("dry-run complete (use --check PATH to validate an existing snapshot)") |
| 304 | |
| 305 | |
| 306 | def build_parser() -> argparse.ArgumentParser: |
| 307 | p = argparse.ArgumentParser( |
| 308 | description="Secret-free Models.dev / OpenRouter catalog automation (#4117)" |
| 309 | ) |
| 310 | sub = p.add_subparsers(dest="cmd", required=True) |
| 311 | |
| 312 | refresh = sub.add_parser("refresh", help="Fetch live catalog / provider models") |
| 313 | refresh.add_argument( |
| 314 | "--provider", |
| 315 | default=None, |
| 316 | help="Optional provider id (currently: openrouter). Omit for Models.dev.", |
| 317 | ) |
| 318 | refresh.add_argument( |
| 319 | "--sort", |
| 320 | default="newest", |
| 321 | choices=["newest", "none"], |
| 322 | help="OpenRouter sort order (default: newest)", |
| 323 | ) |
| 324 | refresh.add_argument( |
| 325 | "--limit", |
| 326 | type=int, |
| 327 | default=100, |
| 328 | help="OpenRouter row cap (default: 100; 0 = no cap)", |
| 329 | ) |
| 330 | refresh.add_argument( |
| 331 | "--write-cache", |
| 332 | metavar="PATH", |
| 333 | help="Deprecated/unsupported: validate-only automation never writes fetched JSON", |
| 334 | ) |
| 335 | refresh.add_argument( |
| 336 | "--write", |
| 337 | metavar="PATH", |
| 338 | help="Deprecated/unsupported alias of --write-cache", |
| 339 | ) |
| 340 | refresh.set_defaults(func=cmd_refresh) |
| 341 | |
| 342 | snapshot = sub.add_parser( |
| 343 | "snapshot", |
| 344 | help="Validate or write a Models.dev-shaped snapshot document", |
| 345 | ) |
| 346 | snapshot.add_argument( |
| 347 | "path", |
| 348 | nargs="?", |
| 349 | default="crates/config/assets/models_dev.bundled.json", |
| 350 | help="Snapshot path (default: offline seed asset)", |
| 351 | ) |
| 352 | snapshot.add_argument( |
| 353 | "--check", |
| 354 | action="store_true", |
| 355 | help="Validate existing file only (no network)", |
| 356 | ) |
| 357 | snapshot.add_argument( |
| 358 | "--write", |
| 359 | action="store_true", |
| 360 | help="Deprecated/unsupported: validate-only automation never writes snapshots", |
| 361 | ) |
| 362 | snapshot.add_argument( |
| 363 | "--force-full", |
| 364 | action="store_true", |
| 365 | help="Deprecated/unsupported with --write; retained for clear failure messages", |
| 366 | ) |
| 367 | snapshot.set_defaults(func=cmd_snapshot) |
| 368 | return p |
| 369 | |
| 370 | |
| 371 | def main(argv: list[str] | None = None) -> None: |
| 372 | parser = build_parser() |
| 373 | args = parser.parse_args(argv) |
| 374 | args.func(args) |
| 375 | |
| 376 | |
| 377 | if __name__ == "__main__": |
| 378 | main() |
| 379 |