| 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 public_source_label(source: str) -> str: |
| 174 | """Log a catalog origin without query/fragment (tokens live there).""" |
| 175 | if source.startswith("url:"): |
| 176 | url = source[4:] |
| 177 | for sep in ("?", "#"): |
| 178 | url = url.split(sep, 1)[0] |
| 179 | return f"url:{url}" |
| 180 | return source |
| 181 | |
| 182 | |
| 183 | def public_limit_value(value: Any) -> str: |
| 184 | """Format a catalog limit for logs. Never print credential-shaped strings. |
| 185 | |
| 186 | Remote catalog JSON is tainted for clear-text-logging rules. Only numeric |
| 187 | limits are meaningful here; anything else (including token-shaped strings) |
| 188 | is replaced with a constant so the raw value cannot reach stdout. |
| 189 | """ |
| 190 | if isinstance(value, bool): |
| 191 | return "redacted" |
| 192 | if value is None: |
| 193 | return "null" |
| 194 | if isinstance(value, int): |
| 195 | return str(value) |
| 196 | if isinstance(value, float): |
| 197 | return format(value, ".6g") |
| 198 | return "redacted" |
| 199 | |
| 200 | |
| 201 | def catalog_stats(data: dict[str, Any]) -> str: |
| 202 | models = data.get("models") or {} |
| 203 | providers = data.get("providers") or {} |
| 204 | offerings = 0 |
| 205 | if isinstance(providers, dict): |
| 206 | for prov in providers.values(): |
| 207 | if isinstance(prov, dict): |
| 208 | models_map = prov.get("models") or {} |
| 209 | if isinstance(models_map, dict): |
| 210 | offerings += len(models_map) |
| 211 | return ( |
| 212 | f"providers={len(providers) if isinstance(providers, dict) else 0} " |
| 213 | f"canonical_models={len(models) if isinstance(models, dict) else 0} " |
| 214 | f"provider_offerings={offerings}" |
| 215 | ) |
| 216 | |
| 217 | |
| 218 | |
| 219 | def cmd_refresh(args: argparse.Namespace) -> None: |
| 220 | if args.provider and args.provider.lower() == "openrouter": |
| 221 | refresh_openrouter(args) |
| 222 | return |
| 223 | if args.provider: |
| 224 | die( |
| 225 | f"unsupported --provider {args.provider!r} " |
| 226 | "(supported: openrouter, or omit for Models.dev)" |
| 227 | ) |
| 228 | |
| 229 | data, source, _is_local = load_models_dev_catalog() |
| 230 | print(f"loaded Models.dev catalog from {source}") |
| 231 | print(catalog_stats(data)) |
| 232 | if args.write_cache or args.write: |
| 233 | die( |
| 234 | "disk writes are intentionally unsupported (secret-free by design); " |
| 235 | "use `snapshot --check PATH` to validate a local Models.dev-shaped file, " |
| 236 | "or `curl`/`CODEWHALE_MODELS_DEV_PATH` for staging" |
| 237 | ) |
| 238 | print("dry-run complete (no secrets; no disk write)") |
| 239 | |
| 240 | |
| 241 | def refresh_openrouter(args: argparse.Namespace) -> None: |
| 242 | url = DEFAULT_OPENROUTER_MODELS_URL |
| 243 | raw = fetch_url(url) |
| 244 | data = load_json_bytes(raw, url) |
| 245 | if not isinstance(data, dict) or "data" not in data: |
| 246 | die(f"{url}: expected {{ data: [...] }} envelope") |
| 247 | rows = data["data"] |
| 248 | if not isinstance(rows, list): |
| 249 | die(f"{url}: data is not a list") |
| 250 | |
| 251 | # Optional sort / limit for local inspection — never secrets. |
| 252 | if args.sort == "newest": |
| 253 | def created_key(row: Any) -> float: |
| 254 | if not isinstance(row, dict): |
| 255 | return 0.0 |
| 256 | created = row.get("created") |
| 257 | try: |
| 258 | return float(created) |
| 259 | except (TypeError, ValueError): |
| 260 | return 0.0 |
| 261 | |
| 262 | rows = sorted(rows, key=created_key, reverse=True) |
| 263 | if args.limit is not None and args.limit > 0: |
| 264 | rows = rows[: args.limit] |
| 265 | |
| 266 | # Project only public catalog fields — never the raw response object — |
| 267 | # so credential-shaped keys cannot reach disk even if OpenRouter adds them. |
| 268 | public_rows: list[dict[str, Any]] = [] |
| 269 | allowed = { |
| 270 | "id", |
| 271 | "name", |
| 272 | "created", |
| 273 | "description", |
| 274 | "context_length", |
| 275 | "architecture", |
| 276 | "pricing", |
| 277 | "top_provider", |
| 278 | "per_request_limits", |
| 279 | "supported_parameters", |
| 280 | } |
| 281 | for row in rows: |
| 282 | if not isinstance(row, dict): |
| 283 | continue |
| 284 | projected: dict[str, Any] = {} |
| 285 | for key in allowed: |
| 286 | if key in row and not is_credential_key(key): |
| 287 | projected[key] = scrub_secrets(row[key]) |
| 288 | if projected.get("id"): |
| 289 | public_rows.append(projected) |
| 290 | payload = { |
| 291 | "_meta": { |
| 292 | "source": "openrouter.ai/api/v1/models", |
| 293 | "note": "Public model listing for cache dogfood; not the Models.dev SoT.", |
| 294 | "count": len(public_rows), |
| 295 | "sort": args.sort, |
| 296 | "limit": args.limit, |
| 297 | }, |
| 298 | "data": public_rows, |
| 299 | } |
| 300 | print(f"loaded OpenRouter models: {len(public_rows)} rows (sort={args.sort}, limit={args.limit})") |
| 301 | if args.write_cache: |
| 302 | # OpenRouter listing is always network-sourced; avoid disk write of remote JSON. |
| 303 | die( |
| 304 | "OpenRouter refresh is dry-run only (no disk write). " |
| 305 | "Use Models.dev with CODEWHALE_MODELS_DEV_PATH for offline snapshots." |
| 306 | ) |
| 307 | else: |
| 308 | print("dry-run complete (OpenRouter writes disabled; use Models.dev local path for caches)") |
| 309 | _ = payload # keep payload construction for future offline path |
| 310 | |
| 311 | |
| 312 | def cmd_snapshot(args: argparse.Namespace) -> None: |
| 313 | target = Path(args.path) |
| 314 | if args.check: |
| 315 | if not target.is_file(): |
| 316 | die(f"--check: missing {target}") |
| 317 | raw = target.read_bytes() |
| 318 | data = load_json_bytes(raw, str(target)) |
| 319 | ensure_models_dev_shape(data, str(target)) |
| 320 | print(f"ok: {target} is Models.dev-shaped ({catalog_stats(data)})") |
| 321 | return |
| 322 | |
| 323 | data, source, _is_local = load_models_dev_catalog() |
| 324 | print(f"loaded Models.dev catalog from {source}") |
| 325 | print(catalog_stats(data)) |
| 326 | if args.write or args.force_full: |
| 327 | die( |
| 328 | "disk writes are intentionally unsupported for this automation; " |
| 329 | "validate with --check, or stage a file outside this tool" |
| 330 | ) |
| 331 | print("dry-run complete (use --check PATH to validate an existing snapshot)") |
| 332 | |
| 333 | |
| 334 | def _limit_fields(entry: Any) -> dict[str, Any]: |
| 335 | if not isinstance(entry, dict): |
| 336 | return {} |
| 337 | limit = entry.get("limit") |
| 338 | return limit if isinstance(limit, dict) else {} |
| 339 | |
| 340 | |
| 341 | def _collect_limit_drift( |
| 342 | seed_value: Any, |
| 343 | upstream_value: Any, |
| 344 | path: str, |
| 345 | drift: list[str], |
| 346 | ) -> None: |
| 347 | seed_limit = _limit_fields(seed_value) |
| 348 | upstream_limit = _limit_fields(upstream_value) |
| 349 | for field in ("context", "output"): |
| 350 | bundled = seed_limit.get(field) |
| 351 | upstream = upstream_limit.get(field) |
| 352 | if bundled != upstream: |
| 353 | drift.append( |
| 354 | f"{path}: limit.{field} " |
| 355 | f"bundled={public_limit_value(bundled)} " |
| 356 | f"upstream={public_limit_value(upstream)}" |
| 357 | ) |
| 358 | |
| 359 | |
| 360 | def cmd_drift(args: argparse.Namespace) -> None: |
| 361 | """Diff the bundled seed against upstream for limit.output / limit.context. |
| 362 | |
| 363 | Dry-run only: no API key, no disk write. Network access (unless |
| 364 | CODEWHALE_MODELS_DEV_PATH points at a local file) reads the public |
| 365 | Models.dev catalog. Wiring this as a CI gate is a maintainer decision and |
| 366 | is deliberately not activated here. |
| 367 | """ |
| 368 | seed_path = Path(args.seed) |
| 369 | if not seed_path.is_file(): |
| 370 | die(f"drift: missing bundled seed {seed_path}") |
| 371 | seed = load_json_bytes(seed_path.read_bytes(), str(seed_path)) |
| 372 | ensure_models_dev_shape(seed, str(seed_path)) |
| 373 | |
| 374 | upstream, source, _is_local = load_models_dev_catalog() |
| 375 | |
| 376 | drift: list[str] = [] |
| 377 | missing_upstream: list[str] = [] |
| 378 | |
| 379 | seed_models = seed.get("models") or {} |
| 380 | upstream_models = upstream.get("models") or {} |
| 381 | if isinstance(seed_models, dict) and isinstance(upstream_models, dict): |
| 382 | for model_id in sorted(seed_models): |
| 383 | if model_id not in upstream_models: |
| 384 | missing_upstream.append(f"models.{model_id}") |
| 385 | continue |
| 386 | _collect_limit_drift( |
| 387 | seed_models[model_id], |
| 388 | upstream_models[model_id], |
| 389 | f"models.{model_id}", |
| 390 | drift, |
| 391 | ) |
| 392 | |
| 393 | seed_providers = seed.get("providers") or {} |
| 394 | upstream_providers = upstream.get("providers") or {} |
| 395 | if isinstance(seed_providers, dict) and isinstance(upstream_providers, dict): |
| 396 | for provider_id in sorted(seed_providers): |
| 397 | seed_provider = seed_providers[provider_id] |
| 398 | seed_offerings = ( |
| 399 | seed_provider.get("models") if isinstance(seed_provider, dict) else {} |
| 400 | ) |
| 401 | if not isinstance(seed_offerings, dict): |
| 402 | continue |
| 403 | upstream_provider = upstream_providers.get(provider_id) |
| 404 | upstream_offerings = ( |
| 405 | upstream_provider.get("models") |
| 406 | if isinstance(upstream_provider, dict) |
| 407 | else {} |
| 408 | ) |
| 409 | if not isinstance(upstream_offerings, dict): |
| 410 | upstream_offerings = {} |
| 411 | for model_id in sorted(seed_offerings): |
| 412 | path = f"providers.{provider_id}.models.{model_id}" |
| 413 | if model_id not in upstream_offerings: |
| 414 | missing_upstream.append(path) |
| 415 | continue |
| 416 | _collect_limit_drift( |
| 417 | seed_offerings[model_id], |
| 418 | upstream_offerings[model_id], |
| 419 | path, |
| 420 | drift, |
| 421 | ) |
| 422 | |
| 423 | print(f"bundled seed: {seed_path}") |
| 424 | print(f"upstream: {public_source_label(source)}") |
| 425 | if missing_upstream: |
| 426 | print("removed upstream (bundled id no longer present):") |
| 427 | for path in missing_upstream: |
| 428 | print(f" - {path}") |
| 429 | if drift: |
| 430 | print(f"limit drift detected ({len(drift)}):") |
| 431 | for path in drift: |
| 432 | print(f" - {path}") |
| 433 | # Non-zero so a future CI gate can fail on drift; activating that gate |
| 434 | # is a maintainer decision. |
| 435 | die( |
| 436 | "bundled seed has drifted from upstream for limit.output / limit.context", |
| 437 | code=2, |
| 438 | ) |
| 439 | if missing_upstream: |
| 440 | print("note: removed-upstream ids are reported above (non-fatal).") |
| 441 | print("no limit.output / limit.context drift") |
| 442 | |
| 443 | |
| 444 | def build_parser() -> argparse.ArgumentParser: |
| 445 | p = argparse.ArgumentParser( |
| 446 | description="Secret-free Models.dev / OpenRouter catalog automation (#4117)" |
| 447 | ) |
| 448 | sub = p.add_subparsers(dest="cmd", required=True) |
| 449 | |
| 450 | refresh = sub.add_parser("refresh", help="Fetch live catalog / provider models") |
| 451 | refresh.add_argument( |
| 452 | "--provider", |
| 453 | default=None, |
| 454 | help="Optional provider id (currently: openrouter). Omit for Models.dev.", |
| 455 | ) |
| 456 | refresh.add_argument( |
| 457 | "--sort", |
| 458 | default="newest", |
| 459 | choices=["newest", "none"], |
| 460 | help="OpenRouter sort order (default: newest)", |
| 461 | ) |
| 462 | refresh.add_argument( |
| 463 | "--limit", |
| 464 | type=int, |
| 465 | default=100, |
| 466 | help="OpenRouter row cap (default: 100; 0 = no cap)", |
| 467 | ) |
| 468 | refresh.add_argument( |
| 469 | "--write-cache", |
| 470 | metavar="PATH", |
| 471 | help="Deprecated/unsupported: validate-only automation never writes fetched JSON", |
| 472 | ) |
| 473 | refresh.add_argument( |
| 474 | "--write", |
| 475 | metavar="PATH", |
| 476 | help="Deprecated/unsupported alias of --write-cache", |
| 477 | ) |
| 478 | refresh.set_defaults(func=cmd_refresh) |
| 479 | |
| 480 | snapshot = sub.add_parser( |
| 481 | "snapshot", |
| 482 | help="Validate or write a Models.dev-shaped snapshot document", |
| 483 | ) |
| 484 | snapshot.add_argument( |
| 485 | "path", |
| 486 | nargs="?", |
| 487 | default="crates/config/assets/models_dev.bundled.json", |
| 488 | help="Snapshot path (default: offline seed asset)", |
| 489 | ) |
| 490 | snapshot.add_argument( |
| 491 | "--check", |
| 492 | action="store_true", |
| 493 | help="Validate existing file only (no network)", |
| 494 | ) |
| 495 | snapshot.add_argument( |
| 496 | "--write", |
| 497 | action="store_true", |
| 498 | help="Deprecated/unsupported: validate-only automation never writes snapshots", |
| 499 | ) |
| 500 | snapshot.add_argument( |
| 501 | "--force-full", |
| 502 | action="store_true", |
| 503 | help="Deprecated/unsupported with --write; retained for clear failure messages", |
| 504 | ) |
| 505 | snapshot.set_defaults(func=cmd_snapshot) |
| 506 | |
| 507 | drift = sub.add_parser( |
| 508 | "drift", |
| 509 | help="Diff bundled seed against upstream for limit.output / limit.context", |
| 510 | ) |
| 511 | drift.add_argument( |
| 512 | "--seed", |
| 513 | default="crates/config/assets/models_dev.bundled.json", |
| 514 | help="Bundled seed path (default: crates/config/assets/models_dev.bundled.json)", |
| 515 | ) |
| 516 | drift.set_defaults(func=cmd_drift) |
| 517 | return p |
| 518 | |
| 519 | |
| 520 | def main(argv: list[str] | None = None) -> None: |
| 521 | parser = build_parser() |
| 522 | args = parser.parse_args(argv) |
| 523 | args.func(args) |
| 524 | |
| 525 | |
| 526 | if __name__ == "__main__": |
| 527 | main() |
| 528 |