#!/usr/bin/env python3 import hashlib import json import re import sys from datetime import datetime, timezone from pathlib import Path PROJECT = Path(__file__).resolve().parents[1] WORD_RE = re.compile(r"[\wÀ-ỹĐđ]+", re.UNICODE) TARGETS = {"cmrq10vod000hj7cbt8zvuj6a", "cmrpr0j9u000bj7cboqonlx4b"} def load(rel): path = PROJECT / rel if not path.exists(): raise RuntimeError(f"missing {rel}") return json.loads(path.read_text()) def digest(path): hasher = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(8 * 1024 * 1024), b""): hasher.update(block) return hasher.hexdigest() def visual(row): return isinstance(row.get("visual_qa"), dict) and row["visual_qa"].get("verified") is True def main(): manifest = load("script/project-manifest.json") draft = load("story/draft-report.json") promotion = load("story/promotion-report.json") originality = load("story/originality-report.json") promotion_receipt = load("story/promotion-receipt.json") tts = load("log/tts-verification.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") media = load("log/postiz-media.json") schedule = load("log/postiz-schedule.json") chapters = [(PROJECT / "story/chapters" / f"{index:02d}.txt").read_text().strip() for index in range(1, 13)] expected = "\n\n".join(chapters) + "\n" spoken_path = PROJECT / "story/spoken-narration.txt" spoken = spoken_path.read_text() canon = digest(spoken_path) upload = PROJECT / "output/final-upload.mp4" upload_hash = digest(upload) if upload.exists() else None posts = schedule.get("posts") or {} timestamps = set() schedule_ok = schedule.get("verified") is True and schedule.get("status") == "completed" and set(posts) == TARGETS if schedule_ok: for integration in TARGETS: rows = posts.get(integration) or [] schedule_ok = schedule_ok and len(rows) == 1 and rows[0].get("state") == "QUEUE" if rows: timestamps.add(rows[0].get("publishDate")) schedule_ok = schedule_ok and len(timestamps) == 1 and None not in timestamps exact_flag = draft.get("exact_sequential_join") if exact_flag is None: exact_flag = (draft.get("spoken_narration") or {}).get("exact_sequential_join") chain_rows = [promotion, tts, layout, intro, footage, final, production, transcode, media, schedule] draft_words = draft.get("total_words", draft.get("total")) checks = { "draft_verified": draft.get("verified", True) is True and draft_words == len(WORD_RE.findall(spoken)), "exact_join": exact_flag is True and spoken == expected, "canon_chain": promotion.get("spoken_sha256") == canon and originality.get("source_sha256") == canon and promotion_receipt.get("promoted_canon_sha256") == canon and all(row.get("source_canon_sha256", canon) == canon for row in chain_rows), "story_gates": promotion.get("verified") is True and originality.get("verified") is True, "tts": tts.get("verified") is True and tts.get("segments_completed") == tts.get("segments_total") and tts.get("duration_seconds", 0) > 0, "layout": layout.get("verified") is True and visual(layout), "intro": intro.get("verified") is True and visual(intro), "footage": footage.get("verified") is True and footage.get("status") == "completed" and visual(footage), "final": final.get("verified") is True and final.get("status") == "completed" and visual(final), "production_gate": production.get("verified") is True and production.get("status") == "passed", "upload_copy": transcode.get("verified") is True and transcode.get("status") == "completed" and visual(transcode) and upload.exists() and upload.stat().st_size < 1_000_000_000 and upload_hash == transcode.get("output_sha256"), "media": media.get("verified") is True and ((media.get("media") or {}).get("video") or {}).get("sha256") == upload_hash, "schedule_exact_two": set(posts) == TARGETS, "schedule_queue_same_timestamp": schedule_ok, } verified = all(checks.values()) report = { "version": 1, "verified": verified, "project_id": PROJECT.name, "title": manifest.get("story_title"), "canon_sha256": canon, "word_count": len(WORD_RE.findall(spoken)), "tts_duration_seconds": tts.get("duration_seconds"), "final": {"job_id": final.get("job_id"), "duration_seconds": (final.get("server_verification") or {}).get("duration"), "bytes": int(((final.get("probe") or {}).get("format") or {}).get("size", 0)), "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") 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" for step in ("story", "tts", "images", "layout", "intro", "footage", "final_render", "metadata", "transcode", "upload", "schedule"): manifest.setdefault("steps", {})[step] = "completed" manifest["updated_at"] = report["checked_at"] (PROJECT / "script/project-manifest.json").write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n") 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: raise SystemExit(main()) except Exception as exc: print(f"Completion blocked: {exc}", file=sys.stderr) raise SystemExit(1)