Hunter B
b1794f8881
Promote the expanded underwater release from the 0.8.68 candidate number to 0.9.0 across Cargo, npm, generated facts, changelogs, current docs, and the authoritative release ledger.\n\nDocument the final runtime, Operate, Fleet, motion, Inkling, community, and deferred-work cutlines without rewriting historical fixtures. Version and release-helper checks pass.
379 lines
13 KiB
Python
Executable File
379 lines
13 KiB
Python
Executable File
#!/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.9.0 (+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()
|