#!/usr/bin/env python """Check every AI Labs catalogue entry against the live DashScope key. Why this exists as a repo script rather than a one-off: the catalogue's whole value is that every model and every option in it was *measured*, and that property decays. A vendor deprecates a model, a console switch gets flipped back on, a region stops offering something — and the first person to notice is an employee whose click fails. This makes "is the catalogue still true?" a two-minute command. It is also the intended pre-merge and post-console-change gate. The two failure modes it was written for, both hit for real on 2026-09-03: * ``AllocationQuota.FreeTierOnly`` — a per-model "免费额度用完即停" switch in the Bailian console, whose error text talks about payment information and sends you to the billing page instead of the switch. * a **silently ignored** parameter — ``wan3.0-video-prime`` accepts ``size`` and then produces a completely unrelated aspect ratio. A catalogue option that does nothing is worse than a missing one, so image sizes are checked against the dimensions actually returned. Usage ----- # against the configured production connector (read-only on the DB) DATABASE_URL_SYNC='mysql+pymysql://...' .venv/bin/python \ scripts/verify_ai_labs_catalog.py # or with an explicit key, no database at all .venv/bin/python scripts/verify_ai_labs_catalog.py --api-key sk-xxx # cheap mode: skip the paid generation probes, check text + existence .venv/bin/python scripts/verify_ai_labs_catalog.py --no-generate Exit code is 0 when every checked entry passed, 1 otherwise, so it can gate a release step. Cost: the default run submits one small image and one 5-second 480P clip per generation capability. Measured at a few cents in total against the org's vendor credit; ``--no-generate`` reduces it to token-level noise. """ from __future__ import annotations import argparse import asyncio import json import os import sys import time from pathlib import Path from typing import Any # Allow ``python scripts/verify_ai_labs_catalog.py`` from the server dir. sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import httpx # noqa: E402 from app.domains.ai_labs import catalog # noqa: E402 from app.domains.bailian.llm_client import ( # noqa: E402 compat_endpoint_for_region, ) from app.domains.bailian.media_client import ( # noqa: E402 PATH_MULTIMODAL, PATH_VIDEO, native_root_for_region, ) OK = "PASS" BAD = "FAIL" SKIP = "SKIP" _FREE_TIER = "AllocationQuota.FreeTierOnly" def _resolve_key_from_db() -> tuple[str, str]: """``(api_key, region)`` from ``platform_settings``. Synchronous pymysql on purpose: an async engine against the production TiDB endpoint hangs from a laptop, and this script only ever reads two rows. """ from sqlalchemy import create_engine, text url = os.environ.get("DATABASE_URL_SYNC") if not url: raise SystemExit( "Set DATABASE_URL_SYNC (mysql+pymysql://…) or pass --api-key.", ) engine = create_engine( url, connect_args={"connect_timeout": 10, "read_timeout": 60}, ) with engine.connect() as conn: rows = dict( conn.execute( text( "SELECT `key`, value FROM platform_settings " "WHERE `key` IN " "('ai_spending.llm.api_key','ai_spending.llm.region')", ), ).all(), ) def _clean(value: Any) -> str: if isinstance(value, bytes | bytearray): value = value.decode() if isinstance(value, str): try: value = json.loads(value) except ValueError: pass return str(value or "").strip() key = _clean(rows.get("ai_spending.llm.api_key")) if not key: raise SystemExit("No ai_spending.llm.api_key configured.") return key, (_clean(rows.get("ai_spending.llm.region")) or "intl") class Reporter: """Collects verdicts and prints them as they arrive.""" def __init__(self) -> None: self.rows: list[tuple[str, str, str, str]] = [] def add(self, verdict: str, kind: str, name: str, detail: str = "") -> None: self.rows.append((verdict, kind, name, detail)) marker = {OK: "✓", BAD: "✗", SKIP: "-"}[verdict] line = f" {marker} {kind:<14} {name:<34} {detail}" print(line.rstrip(), flush=True) @property def failed(self) -> list[tuple[str, str, str, str]]: return [row for row in self.rows if row[0] == BAD] def summary(self) -> str: passed = sum(1 for row in self.rows if row[0] == OK) skipped = sum(1 for row in self.rows if row[0] == SKIP) return ( f"{passed} passed, {len(self.failed)} failed, {skipped} skipped" ) def _describe_error(payload: dict[str, Any], status: int) -> str: code = payload.get("code") or ( (payload.get("error") or {}).get("type") if isinstance( payload.get("error"), dict, ) else None ) message = payload.get("message") or ( (payload.get("error") or {}).get("message") if isinstance( payload.get("error"), dict, ) else "" ) if code == _FREE_TIER or "free quota has been exhausted" in str(message): return ( "BLOCKED by the console switch 「免费额度用完即停」 — turn it off " "for this model (用量 & 费用 → 免费额度). NOT a billing problem." ) return f"HTTP {status} {code or ''} {str(message)[:120]}".strip() async def _check_chat_models( client: httpx.AsyncClient, *, headers: dict, region: str, report: Reporter, ) -> None: endpoint = compat_endpoint_for_region(region) for model in catalog.CHAT_MODELS: body = { "model": model.vendor_model, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 16, } if model.supports_thinking: body["enable_thinking"] = False try: resp = await client.post( endpoint, headers=headers, json=body, timeout=120, ) except httpx.HTTPError as exc: report.add(BAD, "chat", model.key, f"network: {exc}") continue payload = _json(resp) if resp.status_code != 200: report.add(BAD, "chat", model.key, _describe_error(payload, resp.status_code)) continue report.add(OK, "chat", model.key, "text ok") if not model.supports_vision: continue vision_body = { "model": model.vendor_model, "max_tokens": 16, "messages": [{"role": "user", "content": [ {"type": "image_url", "image_url": {"url": _probe_image()}}, {"type": "text", "text": "ok?"}, ]}], } try: resp = await client.post( endpoint, headers=headers, json=vision_body, timeout=120, ) except httpx.HTTPError as exc: report.add(BAD, "chat-vision", model.key, f"network: {exc}") continue if resp.status_code != 200: report.add( BAD, "chat-vision", model.key, "catalogue claims vision: " + _describe_error(_json(resp), resp.status_code), ) else: report.add(OK, "chat-vision", model.key, "image part accepted") _PROBE_IMAGE_CACHE: str | None = None def _probe_image() -> str: """A 64x64 PNG data URI, generated once per run. Deliberately not a 1x1 pixel: the first version of this script used one and three models rejected it with "The image length and width do not meet the model restrictions", which looked exactly like "this model cannot do vision" and would have had the catalogue edited to remove a capability those models actually have. Generated with Pillow rather than embedded as a blob so the size is visible and adjustable here. """ global _PROBE_IMAGE_CACHE if _PROBE_IMAGE_CACHE is None: import base64 import io from PIL import Image buffer = io.BytesIO() Image.new("RGB", (64, 64), (200, 200, 210)).save(buffer, format="PNG") encoded = base64.b64encode(buffer.getvalue()).decode() _PROBE_IMAGE_CACHE = f"data:image/png;base64,{encoded}" return _PROBE_IMAGE_CACHE def _json(resp: httpx.Response) -> dict[str, Any]: try: data = resp.json() except ValueError: return {"message": resp.text[:200]} return data if isinstance(data, dict) else {} async def _check_translate( client: httpx.AsyncClient, *, headers: dict, region: str, report: Reporter, ) -> None: endpoint = compat_endpoint_for_region(region) for option in catalog.TRANSLATE_LANGUAGES: body = { "model": catalog.TRANSLATE_MODEL, "messages": [{"role": "user", "content": "unified IT platform"}], "translation_options": { "source_lang": "auto", "target_lang": option.value, }, "max_tokens": 64, } try: resp = await client.post( endpoint, headers=headers, json=body, timeout=90, ) except httpx.HTTPError as exc: report.add(BAD, "translate", option.value, f"network: {exc}") continue if resp.status_code != 200: report.add( BAD, "translate", option.value, _describe_error(_json(resp), resp.status_code), ) continue payload = _json(resp) text = ( ((payload.get("choices") or [{}])[0].get("message") or {}) .get("content") or "" ) report.add(OK, "translate", option.value, text[:28]) async def _check_image_sizes( client: httpx.AsyncClient, *, headers: dict, root: str, report: Reporter, ) -> None: """Every offered size, checked against the dimensions actually returned. The check is ``requested == produced``, not "the call succeeded": the vendor accepting a parameter and honouring it are different things, and this catalogue exists because they came apart once already. """ capability = catalog.capability(catalog.CAP_IMAGE) if capability is None: return param = capability.param("size") if param is None: return for option in param.options: body = { "model": capability.vendor_model, "input": {"messages": [{ "role": "user", "content": [{"text": "a plain light grey square, flat"}], }]}, "parameters": { **capability.fixed_params, "size": option.value, }, } started = time.monotonic() try: resp = await client.post( f"{root}/{PATH_MULTIMODAL}", headers=headers, json=body, timeout=240, ) except httpx.HTTPError as exc: report.add(BAD, "image-size", str(option.value), f"network: {exc}") continue payload = _json(resp) if resp.status_code != 200 or payload.get("code"): report.add( BAD, "image-size", str(option.value), _describe_error(payload, resp.status_code), ) continue usage = payload.get("usage") or {} produced = f"{usage.get('output_width')}*{usage.get('output_height')}" elapsed = time.monotonic() - started if produced != option.value: report.add( BAD, "image-size", str(option.value), f"SILENTLY IGNORED: asked {option.value}, got {produced}", ) else: report.add( OK, "image-size", str(option.value), f"{produced} in {elapsed:.0f}s", ) async def _check_video( client: httpx.AsyncClient, *, headers: dict, root: str, report: Reporter, ) -> None: """Each resolution submitted and polled to a terminal state. Only the shortest duration is rendered per resolution — the point is to prove the option is accepted and produces a file, not to re-measure render time. """ capability = catalog.capability(catalog.CAP_VIDEO) if capability is None: return resolution = capability.param("resolution") duration = capability.param("duration") if resolution is None or duration is None: return shortest = min(option.value for option in duration.options) async_headers = {**headers, "X-DashScope-Async": "enable"} for option in resolution.options: body = { "model": capability.vendor_model, "input": {"prompt": "a grey cube rotating slowly"}, "parameters": { **capability.fixed_params, "resolution": option.value, "duration": shortest, }, } try: resp = await client.post( f"{root}/{PATH_VIDEO}", headers=async_headers, json=body, timeout=60, ) except httpx.HTTPError as exc: report.add(BAD, "video-res", str(option.value), f"network: {exc}") continue payload = _json(resp) task_id = (payload.get("output") or {}).get("task_id") if resp.status_code != 200 or not task_id: report.add( BAD, "video-res", str(option.value), _describe_error(payload, resp.status_code), ) continue started = time.monotonic() verdict = "timed out in this probe" state = "" while time.monotonic() - started < capability.deadline_seconds: await asyncio.sleep(6) try: poll = await client.get( f"{root}/tasks/{task_id}", headers=headers, timeout=30, ) except httpx.HTTPError as exc: verdict = f"network while polling: {exc}" break output = (_json(poll).get("output") or {}) state = str(output.get("task_status") or "") if state in ("SUCCEEDED", "FAILED", "CANCELED", "UNKNOWN"): usage = (_json(poll).get("usage") or {}) if state == "SUCCEEDED" and output.get("video_url"): seconds = time.monotonic() - started verdict = ( f"SR={usage.get('SR')} " f"{usage.get('output_video_duration')}s " f"in {seconds:.0f}s" ) else: verdict = f"{state} {str(output.get('message') or '')[:80]}" break good = state == "SUCCEEDED" and "SR=" in verdict report.add( OK if good else BAD, "video-res", str(option.value), verdict, ) async def _check_speech( client: httpx.AsyncClient, *, headers: dict, root: str, report: Reporter, ) -> None: capability = catalog.capability(catalog.CAP_SPEECH) if capability is None: return param = capability.param("voice") if param is None: return for option in param.options: body = { "model": capability.vendor_model, "input": {"text": "Helix AI Labs check.", "voice": option.value}, } try: resp = await client.post( f"{root}/{PATH_MULTIMODAL}", headers=headers, json=body, timeout=120, ) except httpx.HTTPError as exc: report.add(BAD, "voice", str(option.value), f"network: {exc}") continue payload = _json(resp) url = ((payload.get("output") or {}).get("audio") or {}).get("url") if resp.status_code != 200 or payload.get("code") or not url: report.add( BAD, "voice", str(option.value), _describe_error(payload, resp.status_code), ) else: report.add(OK, "voice", str(option.value), "audio produced") async def _check_transcribe( client: httpx.AsyncClient, *, headers: dict, root: str, report: Reporter, ) -> None: """Synthesise a known sentence, then transcribe it back. A round trip is the only check that proves the whole path rather than just that the endpoint answers: it exercises the request shape, the inline-audio encoding and the transcript extraction together. """ speech = catalog.capability(catalog.CAP_SPEECH) asr = catalog.capability(catalog.CAP_TRANSCRIBE) if speech is None or asr is None: return sentence = "今天天气很好，我们去公园散步吧。" try: resp = await client.post( f"{root}/{PATH_MULTIMODAL}", headers=headers, timeout=120, json={ "model": speech.vendor_model, "input": {"text": sentence, "voice": "Cherry"}, }, ) url = ((_json(resp).get("output") or {}).get("audio") or {}).get("url") if not url: report.add(SKIP, "transcribe", asr.key, "no audio to test with") return audio = (await client.get(url, timeout=120)).content except httpx.HTTPError as exc: report.add(BAD, "transcribe", asr.key, f"network: {exc}") return import base64 uri = "data:audio/wav;base64," + base64.b64encode(audio).decode() try: resp = await client.post( f"{root}/{PATH_MULTIMODAL}", headers=headers, timeout=180, json={ "model": asr.vendor_model, "input": {"messages": [{ "role": "user", "content": [{"audio": uri}], }]}, }, ) except httpx.HTTPError as exc: report.add(BAD, "transcribe", asr.key, f"network: {exc}") return payload = _json(resp) if resp.status_code != 200 or payload.get("code"): report.add( BAD, "transcribe", asr.key, _describe_error(payload, resp.status_code), ) return text = "" for choice in (payload.get("output") or {}).get("choices") or []: content = (choice.get("message") or {}).get("content") if isinstance(content, list): for part in content: if isinstance(part, dict) and part.get("text"): text += part["text"] report.add( OK if text.strip() else BAD, "transcribe", asr.key, f"heard: {text[:30]}" if text else "empty transcript", ) async def _check_image_edit_exists( client: httpx.AsyncClient, *, headers: dict, root: str, report: Reporter, ) -> None: """Existence probe: send an incomplete request and read the refusal. A parameter complaint means the model is reachable and billable; a free-tier refusal means the console switch is back on. Costs nothing. """ capability = catalog.capability(catalog.CAP_IMAGE_EDIT) if capability is None: return try: resp = await client.post( f"{root}/{PATH_MULTIMODAL}", headers=headers, timeout=60, json={ "model": capability.vendor_model, "input": {"messages": [{ "role": "user", "content": [{"text": "x"}], }]}, }, ) except httpx.HTTPError as exc: report.add(BAD, "image-edit", capability.key, f"network: {exc}") return payload = _json(resp) code = str(payload.get("code") or "") if code == _FREE_TIER or "free quota" in str(payload.get("message") or ""): report.add( BAD, "image-edit", capability.key, _describe_error(payload, resp.status_code), ) elif code in ("InvalidParameter", ""): report.add(OK, "image-edit", capability.key, "reachable") else: report.add( BAD, "image-edit", capability.key, _describe_error(payload, resp.status_code), ) async def main() -> int: parser = argparse.ArgumentParser( description="Verify the AI Labs catalogue against the live endpoint.", ) parser.add_argument("--api-key", default=None) parser.add_argument("--region", default=None, choices=["cn", "intl"]) parser.add_argument( "--no-generate", action="store_true", help="skip the paid image/video/speech probes", ) args = parser.parse_args() if args.api_key: api_key, region = args.api_key.strip(), (args.region or "intl") else: api_key, region = _resolve_key_from_db() region = args.region or region headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } root = native_root_for_region(region) report = Reporter() print( f"AI Labs catalogue {catalog.CATALOG_VERSION} — region={region}\n", ) async with httpx.AsyncClient(trust_env=False) as client: print("chat models") await _check_chat_models( client, headers=headers, region=region, report=report, ) print("\ntranslation") await _check_translate( client, headers=headers, region=region, report=report, ) print("\nimage editing") await _check_image_edit_exists( client, headers=headers, root=root, report=report, ) if args.no_generate: print("\ngeneration probes skipped (--no-generate)") else: print("\nimage sizes") await _check_image_sizes( client, headers=headers, root=root, report=report, ) print("\nspeech voices") await _check_speech( client, headers=headers, root=root, report=report, ) print("\ntranscription round trip") await _check_transcribe( client, headers=headers, root=root, report=report, ) print("\nvideo resolutions (slow — one clip per resolution)") await _check_video( client, headers=headers, root=root, report=report, ) print(f"\n{report.summary()}") if report.failed: print("\nFailures:") for _verdict, kind, name, detail in report.failed: print(f" {kind} {name}: {detail}") print( "\nA free-tier refusal is a console switch, not a billing " "problem — see the note in app/domains/bailian/media_client.py.", ) return 1 return 0 if __name__ == "__main__": raise SystemExit(asyncio.run(main()))