#!/usr/bin/env python3 from __future__ import annotations import base64 import datetime import hashlib import json import os import subprocess import sys import tempfile import urllib.parse import urllib.request from pathlib import Path ROOT = Path(__file__).resolve().parents[1] MANIFEST_PATH = ROOT / "script/project-manifest.json" ENDPOINT = "http://192.168.40.32:7861/voice/ngoc-huyen-clone" def sha256(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for chunk in iter(lambda: handle.read(8 * 1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def resolve_active(manifest: dict, key: str) -> Path: value = (manifest.get("active_paths") or {}).get(key) if not isinstance(value, str) or not value: raise RuntimeError(f"active_paths.{key} is missing") path = Path(value) if not path.is_absolute(): path = ROOT / path try: path.resolve().relative_to(ROOT.resolve()) except ValueError as exc: raise RuntimeError(f"active path escapes project: {key}") from exc return path def atomic_write(path: Path, data: bytes) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: with os.fdopen(fd, "wb") as handle: handle.write(data) handle.flush() os.fsync(handle.fileno()) os.replace(name, path) finally: if os.path.exists(name): os.unlink(name) def atomic_json(path: Path, value: dict) -> None: atomic_write(path, (json.dumps(value, ensure_ascii=False, indent=2) + "\n").encode("utf-8")) def make_sample(text: str) -> str: paragraphs = [value.strip() for value in text.split("\n\n") if value.strip()] chosen: list[str] = [] count = 0 for paragraph in paragraphs: chosen.append(paragraph) count += len(paragraph.split()) if count >= 600: break sample = "\n\n".join(chosen) + "\n" words = len(sample.split()) if not 500 <= words <= 800: raise RuntimeError(f"sample word count outside 500-800: {words}") return sample def find_audio(value: object) -> bytes | None: if isinstance(value, dict): for key in ("audio_base64", "base64", "audio", "data"): item = value.get(key) if isinstance(item, str): raw = item.split(",", 1)[-1] if item.startswith("data:") else item try: decoded = base64.b64decode(raw, validate=True) except Exception: decoded = b"" if decoded.startswith(b"RIFF") and decoded[8:12] == b"WAVE": return decoded for item in value.values(): found = find_audio(item) if found: return found elif isinstance(value, list): for item in value: found = find_audio(item) if found: return found return None def main() -> int: manifest = json.loads(MANIFEST_PATH.read_text(encoding="utf-8")) candidate = resolve_active(manifest, "candidate") validation_path = resolve_active(manifest, "candidate_receipt") producer_close_path = resolve_active(manifest, "producer_close_receipt") receipt_path = resolve_active(manifest, "voice_test_receipt") run_dir = receipt_path.parent sample_path = run_dir / "candidate.txt" wav_path = run_dir / "voice-test.wav" run_dir.mkdir(parents=True, exist_ok=True) if receipt_path.exists() or sample_path.exists() or wav_path.exists(): raise RuntimeError("voice-test run output already exists and requires audit/invalidation") candidate_hash = sha256(candidate) close = json.loads(producer_close_path.read_text(encoding="utf-8")) validation = json.loads(validation_path.read_text(encoding="utf-8")) if close.get("status") != "completed" or close.get("writer_closed") is not True: raise RuntimeError("candidate writer is not closed") if close.get("candidate_sha256") != candidate_hash or close.get("bytes", close.get("byte_count")) != candidate.stat().st_size: raise RuntimeError("candidate differs from producer-close marker") if validation.get("verified") is not True or validation.get("candidate_sha256") != candidate_hash: raise RuntimeError("candidate validation is absent, failed, or stale") sample_text = make_sample(candidate.read_text(encoding="utf-8")) atomic_write(sample_path, sample_text.encode("utf-8")) sample_words = len(sample_text.split()) payload = urllib.parse.urlencode({ "text": sample_text, "style": "doc_truyen", "speed": "1.0", "denoise": "true", }).encode("utf-8") request = urllib.request.Request( ENDPOINT, data=payload, headers={"Content-Type": "application/x-www-form-urlencoded"}, method="POST", ) with urllib.request.urlopen(request, timeout=600) as response: body = response.read() http_status = response.status content_type = response.headers.get("Content-Type", "") audio = body if body.startswith(b"RIFF") and body[8:12] == b"WAVE" else None response_shape = "wav" if audio is None: response_shape = "json" value = json.loads(body.decode("utf-8")) audio = find_audio(value) if audio is None: raise RuntimeError("TTS response contains no WAV audio") atomic_write(wav_path, audio) probe = json.loads(subprocess.check_output([ "ffprobe", "-v", "error", "-show_entries", "format=duration,format_name:stream=codec_name,codec_type,sample_rate,channels", "-of", "json", str(wav_path), ], text=True)) duration = float(probe["format"]["duration"]) audio_streams = [value for value in probe.get("streams", []) if value.get("codec_type") == "audio"] verified = ( http_status == 200 and duration > 0 and len(audio_streams) == 1 and audio_streams[0].get("codec_name") in {"pcm_s16le", "pcm_s24le", "pcm_f32le"} and int(audio_streams[0].get("channels", 0)) >= 1 ) receipt = { "status": "completed" if verified else "failed", "verified": verified, "candidate_path": str(candidate), "candidate_sha256": candidate_hash, "producer_close_receipt": str(producer_close_path), "producer_close_sha256": sha256(producer_close_path), "candidate_validation_receipt": str(validation_path), "candidate_validation_sha256": sha256(validation_path), "sample_path": str(sample_path), "sample_sha256": sha256(sample_path), "sample_words": sample_words, "endpoint": ENDPOINT, "voice": "Ngoc-Huyen-Clone", "style": "doc_truyen", "speed": 1.0, "denoise": True, "http_status": http_status, "response_content_type": content_type, "response_shape": response_shape, "wav_path": str(wav_path), "wav_sha256": sha256(wav_path), "duration_seconds": duration, "wpm": sample_words / duration * 60, "probe": probe, "verified_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), "hash_method": "sha256 streaming 8 MiB", } atomic_json(receipt_path, receipt) print(json.dumps({ "status": receipt["status"], "verified": verified, "sample_words": sample_words, "duration_seconds": duration, "wpm": receipt["wpm"], "receipt": str(receipt_path), }, ensure_ascii=False)) return 0 if verified else 1 if __name__ == "__main__": try: sys.exit(main()) except Exception as exc: failure = { "status": "failed", "verified": False, "error": f"{type(exc).__name__}: {exc}", "failed_at": datetime.datetime.now(datetime.timezone.utc).isoformat(), } try: manifest = json.loads(MANIFEST_PATH.read_text(encoding="utf-8")) atomic_json(resolve_active(manifest, "voice_test_receipt"), failure) except Exception: pass print(json.dumps(failure, ensure_ascii=False), file=sys.stderr) sys.exit(1)