build_examples.py

"""Build the precomputed example cache.

Inputs:
  example_decomps.json   — phrase decompositions authored by a LARGE model
                           (so the examples are reliably correct, unlike the
                           small on-device model).
Pipeline:
  decomposition -> translate.build_layers() (deterministic 7 layers)
               -> Qwen3-TTS audio per layer, fetched from the deployed Modal
                  endpoint, downloaded and saved under static/example_audio/
Outputs:
  examples_cache.py      — full results (layers/links + per-layer `audio` URL)
  static/example_audio/  — committed .wav files (instant, no model at runtime)

Run:  python build_examples.py
"""
import hashlib
import json
import time
import urllib.request

import config
import examples
import translate

REMOTE = config.TTS_REMOTE_URL.rstrip("/")
AUDIO_DIR = config.STATIC_DIR / "example_audio"


def fetch_tts(text: str, lang: str) -> bytes:
    """Fetch one utterance from the deployed Qwen3-TTS endpoint (retry for the
    L4 cold-start blip; reject fallback beeps)."""
    payload = json.dumps({"text": text, "lang": lang}).encode("utf-8")
    last = None
    for attempt in range(4):
        try:
            req = urllib.request.Request(
                REMOTE + "/api/tts", data=payload,
                headers={"Content-Type": "application/json"},
            )
            with urllib.request.urlopen(req, timeout=300) as r:
                info = json.load(r)
            with urllib.request.urlopen(REMOTE + info["url"], timeout=120) as r:
                data = r.read()
            if len(data) < 25000:
                raise RuntimeError("remote returned fallback audio (beep)")
            return data
        except Exception as e:
            last = e
            print(f"    retry {attempt+1} ({str(e)[:60]})")
            time.sleep(5)
    raise last


def main():
    decomps = json.load(open("example_decomps.json", encoding="utf-8"))
    AUDIO_DIR.mkdir(parents=True, exist_ok=True)
    cache = {}
    seen_audio = set()

    for text, src, tgt in examples.EXAMPLES:
        k = f"{src}|{tgt}|{text}"
        if k not in decomps:
            print("!! missing decomposition:", k)
            continue
        d = decomps[k]
        decomp = {"final": d["final"], "units": d["units"], "source_text": text}
        result = translate.build_layers(decomp, src, tgt)

        for layer in result["layers"]:
            idx = layer["index"]
            lang = src if idx == 0 else tgt            # matches frontend langForLayer
            ltext = layer["text"]
            h = hashlib.sha1(f"{lang}|{ltext}".encode("utf-8")).hexdigest()[:16]
            fpath = AUDIO_DIR / f"{h}.wav"
            if h not in seen_audio and not fpath.exists():
                print(f"  {src}->{tgt} L{idx} [{lang}]: {ltext[:34]}")
                fpath.write_bytes(fetch_tts(ltext, lang))
            seen_audio.add(h)
            layer["audio"] = f"/static/example_audio/{h}.wav"

        cache[k] = result
        print("done:", k)

    blob = json.dumps(cache, ensure_ascii=False)
    with open("examples_cache.py", "w", encoding="utf-8") as f:
        f.write('"""Auto-generated by build_examples.py — do not edit by hand.\n')
        f.write("Precomputed example results (layers + per-layer audio) so the\n")
        f.write('Surprise-me button needs no LLM and no TTS at runtime."""\n')
        f.write("import json\n\nCACHE = json.loads(%r)\n" % blob)
    print(f"\nwrote examples_cache.py: {len(cache)} examples, "
          f"{len(seen_audio)} audio clips in {AUDIO_DIR}")


if __name__ == "__main__":
    main()