scripts/record_demo_fixtures.py

Rama main · réplica de lectura del repositorio original.

#!/usr/bin/env python3
"""Capture the demo's canned API responses from a running STT server.

Development tool, never shipped. Hand-writing fixtures for two dozen endpoints
guesses their shapes wrong and drifts silently as the API moves; recording them from
a real server gets the shapes right and can be re-run when something changes.

Everything captured passes through stt.demo_redact first, because a real server's
responses carry the operator's home directory, the machine's hostname, the paired
machine's address and any SMB credentials. Read the residual-flag report it prints
before committing the result.

Usage:
    .venv/bin/python3 scripts/record_demo_fixtures.py --base-url http://127.0.0.1:8080
    .venv/bin/python3 scripts/record_demo_fixtures.py --dry-run     # print, don't write
"""

from __future__ import annotations

import argparse
import json
import os
import socket
import sys
import urllib.error
import urllib.request

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

from stt import demo_redact  # noqa: E402

# GET endpoints whose real responses the demo serves back. Only the families the demo
# intercepts — anything the demo lets through to the real server has no fixture.
ENDPOINTS = (
    "/api/models/list",
    "/api/models/manager",
    "/api/models/local",
    "/api/models/nllb-list",
    "/api/models/nllb-status",
    "/api/models/madlad-list",
    "/api/models/gguf-list",
    "/api/models/faster-whisper/list",
    "/api/models/tts-list",
    "/api/models/panns-status",
    "/api/models/silero-vad-status",
    "/api/models/download-status",
    "/api/tts/status",
    "/api/tts/voices",
    "/api/tts/models",
    "/api/translation/settings",
    "/api/translation/status",
    "/api/audio-devices",
    "/api/file-mover/status",
    "/api/tunnel/status",
    "/api/translate/llm-prompt",
)

HEADER = '''"""Canned API responses the demo serves in place of the real backends.

GENERATED by scripts/record_demo_fixtures.py — do not edit by hand; re-record instead.
Captured from a real server so the shapes are right, then passed through
stt.demo_redact to remove the recording machine's identity.
"""

from __future__ import annotations

from typing import Any, Dict

'''


def fetch(base_url: str, path: str, timeout: float = 15.0):
    """The endpoint's JSON, or None if it is not one this server serves."""
    try:
        with urllib.request.urlopen(base_url.rstrip("/") + path, timeout=timeout) as response:
            body = response.read().decode("utf-8")
    except urllib.error.HTTPError as exc:
        print(f"  {path}: HTTP {exc.code} — skipped")
        return None
    except (urllib.error.URLError, OSError, TimeoutError) as exc:
        print(f"  {path}: unreachable ({exc}) — skipped")
        return None
    try:
        return json.loads(body)
    except ValueError:
        print(f"  {path}: not JSON — skipped")
        return None


def render(fixtures) -> str:
    body = json.dumps(fixtures, indent=4, ensure_ascii=False, sort_keys=True)
    return (HEADER
            + "RESPONSES: Dict[str, Any] = "
            + body.replace(": true", ": True").replace(": false", ": False")
                  .replace(": null", ": None")
            + "\n")


def main() -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--base-url", default="http://127.0.0.1:8080",
                        help="a running STT server to record from")
    parser.add_argument("--out", default=os.path.join("stt", "demo_fixtures.py"))
    parser.add_argument("--hostname", default=socket.gethostname(),
                        help="the recording machine's hostname, replaced wherever it appears")
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()

    print(f"Recording from {args.base_url} (hostname {args.hostname!r})")
    fixtures = {}
    for path in ENDPOINTS:
        payload = fetch(args.base_url, path)
        if payload is None:
            continue
        fixtures[path] = demo_redact.redact(payload, hostname=args.hostname)
        print(f"  {path}: captured")

    if not fixtures:
        print("Nothing captured — is the server running and reachable?")
        return 1

    flags = []
    for path, payload in fixtures.items():
        flags.extend(f"{path} {flag}" for flag in
                     demo_redact.residual_flags(payload, hostname=args.hostname))

    print(f"\n{len(fixtures)} endpoints captured.")
    if flags:
        print(f"\n!! {len(flags)} residual flags — review before committing:")
        for flag in flags:
            print(f"   {flag}")
    else:
        print("No residual flags.")

    rendered = render(fixtures)
    if args.dry_run:
        print(f"\n--dry-run: would write {len(rendered)} bytes to {args.out}")
        return 0
    with open(args.out, "w", encoding="utf-8") as handle:
        handle.write(rendered)
    print(f"\nWrote {args.out} ({len(rendered)} bytes)")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())