#!/usr/bin/env python3 import hashlib import json import re import subprocess import sys from datetime import datetime, timezone from pathlib import Path PROJECT = Path(__file__).resolve().parents[1] WORD_RE = re.compile(r"[^\W_]+(?:['’-][^\W_]+)*", re.UNICODE) def load(rel): path = PROJECT / rel if not path.is_file(): raise RuntimeError(f"missing {rel}") return json.loads(path.read_text(encoding="utf-8")) def sha(path): digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(8 * 1024 * 1024), b""): digest.update(block) return digest.hexdigest() def report_words(draft): return draft.get("word_count", draft.get("spoken_word_count", draft.get("aggregate_word_count"))) def main(): gate = subprocess.run([sys.executable, str(PROJECT / "script/validate-workflow-copy.py"), "full"], cwd=PROJECT) if gate.returncode: raise RuntimeError("workflow-copy authority gate failed") manifest = load("script/project-manifest.json") brief = load("script/story-brief.json") draft = load("story/draft-report.json") promotion = load("story/promotion-report.json") originality = load("story/originality-report.json") pronunciation = load("log/tts-pronunciation.json") intro_pronunciation = load("log/intro-tts-pronunciation.json") tts = load("log/tts-verification.json") intro_voice = load("log/intro-voice.json") images = load("log/image-generation.json") layout = load("image/layout-manifest.json") intro = load("log/intro-render.json") footage = load("log/footage-job.json") final = load("log/final-render.json") production = load("log/production-verification.json") transcode = load("log/upload-transcode.json") metadata = load("log/metadata-verification.json") media = load("log/postiz-media.json") schedule = load("log/postiz-schedule.json") chapters = [(PROJECT / f"story/chapters/chapter-{index:02d}.txt").read_text(encoding="utf-8").rstrip() for index in range(1, 11)] expected = "\n\n\n".join(chapters) + "\n" spoken_path = PROJECT / "story/spoken-narration.txt" canon_path = PROJECT / "story/story-canon.txt" pronunciation_path = PROJECT / "story/tts-pronunciation.txt" intro_source_path = PROJECT / "audio/intro-voice-source.txt" intro_projection_path = PROJECT / "audio/intro-voice-pronunciation.txt" intro_voice_path = PROJECT / "audio/intro-voice.wav" lexicon_path = PROJECT / "script/pronunciation-lexicon.json" story_audio = PROJECT / "audio/story-full.wav" spoken = spoken_path.read_text(encoding="utf-8") canon = sha(spoken_path) words = len(WORD_RE.findall(spoken)) upload = PROJECT / "output/final-upload.mp4" upload_hash = sha(upload) if upload.is_file() else None posts = schedule.get("posts") or {} timestamps = set() expected_integrations = {"cmrq10vod000hj7cbt8zvuj6a", "cmrpr0j9u000bj7cboqonlx4b"} schedule_ok = schedule.get("verified") is True and schedule.get("status") == "completed" and set(posts) == expected_integrations if schedule_ok: for integration in expected_integrations: rows = posts.get(integration) or [] schedule_ok = schedule_ok and len(rows) == 1 and rows[0].get("state") == "QUEUE" and bool(rows[0].get("id")) if rows: timestamps.add(rows[0].get("publishDate")) schedule_ok = schedule_ok and len(timestamps) == 1 and None not in timestamps audio_row = next((row for row in intro_pronunciation.get("lexicon", []) if row.get("source") == "audio"), {}) family_ok = ( brief.get("primary_story_family") == "Cưới trước yêu sau / hôn ước" and brief.get("secondary_story_family") == "Không có" and promotion.get("primary_story_family") == brief.get("primary_story_family") and promotion.get("secondary_story_family") == brief.get("secondary_story_family") and promotion.get("family_adherence_verified") is True ) chain = [promotion, pronunciation, intro_pronunciation, tts, images, layout, intro, footage, final, production, transcode, metadata, media, schedule] checks = { "story": draft.get("verified") is True and report_words(draft) == words and int(brief["length"]["min_words"]) <= words <= int(brief["length"]["max_words"]), "exact_join": spoken == expected and canon_path.read_bytes() == spoken_path.read_bytes(), "story_family": family_ok, "originality_inside_family": originality.get("verified") is True and originality.get("primary_story_family_preserved") is True, "canon_chain": promotion.get("spoken_sha256") == canon and originality.get("source_canon_sha256") == canon and all(row.get("source_canon_sha256", canon) == canon for row in chain), "narration_pronunciation": pronunciation.get("verified") is True and pronunciation.get("kind") == "narration" and pronunciation.get("reverse_verified") is True and pronunciation.get("semantic_content_changed") is False and pronunciation.get("non_lexicon_changes") == 0 and pronunciation.get("source_sha256") == canon and pronunciation_path.is_file() and lexicon_path.is_file() and pronunciation.get("projection_sha256") == sha(pronunciation_path) and pronunciation.get("lexicon_sha256") == sha(lexicon_path), "intro_pronunciation": intro_pronunciation.get("verified") is True and intro_pronunciation.get("kind") == "intro" and intro_pronunciation.get("reverse_verified") is True and intro_pronunciation.get("semantic_content_changed") is False and intro_pronunciation.get("non_lexicon_changes") == 0 and audio_row.get("source_count") == audio_row.get("replacement_count") and audio_row.get("replacement_count", 0) >= 1 and intro_source_path.is_file() and intro_projection_path.is_file() and lexicon_path.is_file() and intro_pronunciation.get("source_sha256") == sha(intro_source_path) and intro_pronunciation.get("projection_sha256") == sha(intro_projection_path) and intro_pronunciation.get("lexicon_sha256") == sha(lexicon_path), "narration_tts": tts.get("verified") is True and tts.get("voice") == "ngoc-huyen-vbee" and tts.get("pronunciation_verified") is True and tts.get("source_sha256") == pronunciation.get("projection_sha256") and tts.get("chunk_reconstruction_sha256") == pronunciation.get("projection_sha256") and tts.get("pronunciation_receipt_sha256") == sha(PROJECT / "log/tts-pronunciation.json") and tts.get("pronunciation_lexicon_sha256") == sha(lexicon_path) and tts.get("segments_completed") == tts.get("segments_total") and story_audio.is_file() and tts.get("output_sha256") == sha(story_audio), "intro_tts": intro_voice.get("verified") is True and intro_voice.get("voice") == "ngoc-huyen-vbee" and intro_voice.get("source_sha256") == intro_pronunciation.get("source_sha256") and intro_voice.get("projection_sha256") == intro_pronunciation.get("projection_sha256") and intro_voice.get("pronunciation_receipt_sha256") == sha(PROJECT / "log/intro-tts-pronunciation.json") and intro_voice.get("pronunciation_lexicon_sha256") == sha(lexicon_path) and intro_voice_path.is_file() and intro_voice.get("output_sha256") == sha(intro_voice_path), "provider_images": images.get("verified") is True and images.get("provider_rendered_text_only") is True, "layout_artwork_qa": layout.get("verified") is True and layout.get("status") == "completed" and layout.get("builder_location") == "agent_local" and layout.get("remote_layout_build_forbidden") is True and layout.get("visual_qa", {}).get("verified") is True and images.get("provider_rendered_text_only") is True, "intro_gpu": intro.get("verified") is True and intro.get("visual_qa", {}).get("verified") is True and intro.get("encoder_required") == "h264_nvenc" and intro.get("intro_pronunciation_sha256") == intro_pronunciation.get("projection_sha256") and intro.get("intro_pronunciation_receipt_sha256") == sha(PROJECT / "log/intro-tts-pronunciation.json") and intro.get("intro_voice_receipt_sha256") == sha(PROJECT / "log/intro-voice.json"), "footage_technical": footage.get("verified") is True and footage.get("status") == "completed" and footage.get("visual_qa", {}).get("required") is False and footage.get("server_verification", {}).get("mirror_applied") is True and footage.get("server_verification", {}).get("source_audio_discarded") is True, "final_technical": final.get("verified") is True and final.get("status") == "completed" and final.get("visual_qa", {}).get("required") is False, "production": production.get("verified") is True and production.get("status") == "passed", "upload_copy": transcode.get("verified") is True and transcode.get("status") == "completed" and transcode.get("visual_qa", {}).get("required") is False and upload.is_file() and upload.stat().st_size < 1_000_000_000 and upload_hash == transcode.get("output_sha256"), "metadata": metadata.get("verified") is True and 2 <= metadata.get("title_length", 0) <= 100 and metadata.get("sha256") == sha(PROJECT / metadata.get("path", "output/info.txt")), "media": media.get("verified") is True and ((media.get("media") or {}).get("video") or {}).get("sha256") == upload_hash, "schedule": schedule_ok, } verified = all(checks.values()) report = { "version": 1, "verified": verified, "project_id": PROJECT.name, "title": brief.get("canonical_title"), "canon_sha256": canon, "word_count": words, "tts_words_per_minute": 231, "predicted_tts_minutes": round(words / 231, 3), "tts_duration_seconds": tts.get("duration_seconds"), "final": {"job_id": final.get("job_id"), "sha256": final.get("output_sha256")}, "upload_copy": {"job_id": transcode.get("job_id"), "bytes": transcode.get("output_bytes"), "sha256": transcode.get("output_sha256")}, "schedule": {"slot_local": schedule.get("slot_local"), "slot_utc": schedule.get("slot_utc"), "posts": posts}, "checks": checks, "checks_passed": sum(checks.values()), "checks_total": len(checks), "checked_at": datetime.now(timezone.utc).isoformat(), } output = PROJECT / "log/completion-report.json" output.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") if not verified: print(json.dumps({"verified": False, "failed": [key for key, value in checks.items() if not value]}, ensure_ascii=False)) return 1 manifest["status"] = "completed" manifest["steps"] = {key: "completed" for key in ("capability_preflight", "story_brief", "architecture", "voice", "story", "pronunciation", "tts", "artwork", "intro", "footage", "final", "transcode", "publish", "completion")} manifest["updated_at"] = report["checked_at"] (PROJECT / "script/project-manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"verified": True, "checks_passed": report["checks_passed"], "checks_total": report["checks_total"], "manifest_status": "completed"}, ensure_ascii=False)) return 0 if __name__ == "__main__": try: sys.exit(main()) except Exception as exc: print(f"Completion blocked: {exc}", file=sys.stderr) sys.exit(1)