| """Explicit real-model unknown-symbol pause + short-first-batch acceptance.""" |
| import argparse |
| from datetime import datetime, timezone |
| import json |
| from pathlib import Path |
| import sys |
| import threading |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| sys.path.insert(0, str(ROOT)) |
| from tts.engine import SpeechEngine |
| from tts.text import SpeechError |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--assets", type=Path, required=True) |
| parser.add_argument("--output", type=Path, required=True) |
| args = parser.parse_args() |
| args.output.mkdir(parents=True, exist_ok=True) |
| engine = SpeechEngine(args.assets, ROOT / "assets.lock.json") |
| from mutagen.mp3 import MP3 |
| report = {"at": datetime.now(timezone.utc).isoformat(), "engine": engine.info, "cases": [], "scope": "Real local CPU encoded MP3 and complete source ranges; not manual audible/production latency acceptance"} |
| cases = (("mixed", "校长读长文。𠮷※★😀αЖ，CPU和API继续。"), ("pause-only", "𠮷※★😀αЖ。"), |
| ("control", "甲\f乙，后文继续。"), ("short-first", "这是一篇长文，校长文章写得很好，我得亲自处理。")) |
| for name, text in cases: |
| file = args.output / (name + ".mp3") |
| published = [] |
| result = engine.synthesize(text, "zf_001", 1, file, threading.Event(), progress=published.append, unsupported_policy="pause") |
| assert result["segments"][0]["start"] == 0 |
| assert result["segments"][-1]["end"] == len(text.encode("utf-16-le")) // 2 |
| assert all(a["end"] == b["start"] for a, b in zip(result["segments"], result["segments"][1:])) |
| assert len(published) == len(result["segments"]) |
| assert abs(MP3(file).info.length - result["audio_seconds"]) < .25 |
| for part in published: |
| clip = file.with_name(f"{file.stem}.{part['index']}.mp3") |
| assert abs(MP3(clip).info.length - part["audio_seconds"]) < .25 |
| if name == "mixed": assert result["unsupported_phoneme_count"] >= 6 and result["batches"] > 0 |
| if name == "pause-only": assert result["batches"] == 0 |
| report["cases"].append({"name": name, "result": result}) |
| print(name, "PASS", "duration", result["audio_seconds"], "first-part", result["first_audio_seconds"], flush=True) |
| try: |
| engine.synthesize(cases[0][1], "zf_001", 1, args.output / "strict.mp3", threading.Event()) |
| except SpeechError as error: |
| assert error.code == "unsupported_text" |
| else: |
| raise AssertionError("Default strict API must remain strict") |
| (args.output / "acceptance.json").write_text(json.dumps(report, ensure_ascii=False, indent=2)) |
|
|
|
|
| if __name__ == "__main__": main() |
|
|
