#!/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] SCRIPT = PROJECT / "script" STORY = PROJECT / "story" CHAPTERS = STORY / "chapters" REPORT = STORY / "promotion-report.json" WORD_RE = re.compile(r"[\wÀ-ỹĐđ]+", re.UNICODE) HEADING_RE = re.compile(r"(?im)^\s*(?:chương|phần|cảnh|hồi)\s+(?:\d+|[ivxlcdm]+)\s*[:.\-–—]?\s*$") PRODUCTION_RE = re.compile(r"(?i)\b(?:scene_id|target_words|production note|ghi chú sản xuất|end state|reveal_id)\b") CTA_RE = re.compile(r"(?i)(?:đăng ký|subscribe)\s+kênh|hãy\s+(?:like|chia sẻ)") def load(name): return json.loads((SCRIPT / name).read_text(encoding="utf-8")) def sha(path): return hashlib.sha256(path.read_bytes()).hexdigest() def words(text): return len(WORD_RE.findall(text)) def main(): outline = load("outline.json") registry = load("identity-registry.json") ledger = load("reveal-ledger.json") brief = load("story-brief.json") plan = load("production-plan.json") manifest = load("project-manifest.json") issues = [] authority = { "outline_sha256": sha(SCRIPT / "outline.json"), "identity_registry_sha256": sha(SCRIPT / "identity-registry.json"), "reveal_ledger_sha256": sha(SCRIPT / "reveal-ledger.json"), "story_brief_sha256": sha(SCRIPT / "story-brief.json"), } if authority != plan.get("authority_hashes"): issues.append("current authority bytes differ from production-plan lock") if authority != manifest.get("authority_hashes"): issues.append("current authority bytes differ from project-manifest lock") expected = int(plan["expected_chapters"]) paths = [CHAPTERS / f"chapter-{index:02d}.txt" for index in range(1, expected + 1)] missing = [str(path.relative_to(PROJECT)) for path in paths if not path.exists()] if missing: issues.append("missing chapters: " + ", ".join(missing)) rows, chapter_bytes = [], [] for index, path in enumerate(paths, 1): if not path.exists(): continue raw = path.read_bytes() try: text = raw.decode("utf-8") except UnicodeDecodeError: issues.append(f"chapter {index:02d} is not UTF-8") continue clean = text.rstrip() chapter_bytes.append(clean.encode("utf-8")) if not clean: issues.append(f"chapter {index:02d} empty") if HEADING_RE.search(clean): issues.append(f"chapter {index:02d} contains spoken heading") if PRODUCTION_RE.search(clean) or CTA_RE.search(clean): issues.append(f"chapter {index:02d} contains production/CTA marker") if "\t" in clean or any(line.endswith(" ") for line in text.splitlines()): issues.append(f"chapter {index:02d} has whitespace hygiene issue") rows.append({"chapter": index, "path": str(path.relative_to(PROJECT)), "words": words(clean), "bytes": len(raw), "sha256": hashlib.sha256(raw).hexdigest()}) joined = b"\n\n".join(chapter_bytes) + b"\n" if len(chapter_bytes) == expected else b"" spoken_path = STORY / "spoken-narration.txt" canon_path = STORY / "story-canon.txt" actual = spoken_path.read_bytes() if spoken_path.exists() else b"" canon_bytes = canon_path.read_bytes() if canon_path.exists() else b"" exact_join = bool(joined) and actual == joined and canon_bytes == actual if not exact_join: issues.append("spoken narration/story canon are not exact sequential join") try: actual_text = actual.decode("utf-8") except UnicodeDecodeError: actual_text = "" issues.append("spoken narration is not UTF-8") total_words = words(actual_text) low, high = plan["accepted_range"] if not low <= total_words <= high: issues.append(f"word count {total_words} outside {low}-{high}") draft_path = STORY / "draft-report.json" draft = json.loads(draft_path.read_text(encoding="utf-8")) if draft_path.exists() else {} if draft.get("verified") is not True or draft.get("total_words") != total_words: issues.append("draft report is missing, unverified, or word count mismatched") spoken_row = draft.get("spoken_narration") or {} if spoken_row.get("exact_sequential_join") is not True or spoken_row.get("sha256") != hashlib.sha256(actual).hexdigest(): issues.append("draft report exact-join/hash mismatch") anchors = draft.get("beat_anchors") or draft.get("beats") or [] if len(anchors) < len(outline.get("beats", [])): issues.append("draft report does not cover every locked beat") allowed_names = [] for entity in registry.get("entities", []): allowed_names.extend(entity.get("aliases", [])) identity_ok = all(entity.get("full_name") in actual_text for entity in registry.get("entities", []) if entity.get("id") in {"char_001", "char_002"}) if not identity_ok: issues.append("central identity names missing") if HEADING_RE.search(actual_text) or PRODUCTION_RE.search(actual_text) or CTA_RE.search(actual_text): issues.append("spoken narration hygiene failed") reveal_ok = all(int(row.get("reveal_beat", 0)) > max(row.get("seed_beats", [0])) for row in ledger.get("reveals", [])) if not reveal_ok: issues.append("reveal ledger order invalid") report = { "version": 1, "status": "passed" if not issues else "failed", "verified": not issues, "canonical_title": outline["canonical_title"], "expected_chapters": expected, "chapters_present": len(rows), "chapter_rows": rows, "spoken_narration": str(spoken_path.relative_to(PROJECT)), "exact_sequential_join": exact_join, "total_words": total_words, "target_range": [low, high], "tts_words_per_minute": 231, "estimated_minutes_at_baseline": round(total_words / 231, 3), "spoken_sha256": hashlib.sha256(actual).hexdigest() if actual else None, "authority_hashes_current": authority, "brief_state": brief.get("state"), "identity_verified": identity_ok, "reveal_ledger_verified": reveal_ok, "draft_report_sha256": sha(draft_path) if draft_path.exists() else None, "issues": issues, "checked_at": datetime.now(timezone.utc).isoformat(), } REPORT.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") print(json.dumps({"verified": report["verified"], "words": total_words, "issues": issues}, ensure_ascii=False)) return 0 if report["verified"] else 1 if __name__ == "__main__": sys.exit(main())