#!/usr/bin/env python3 """Validate a source-backed nAvid editorial pilot at a declared checkpoint.""" from __future__ import annotations import argparse import json from pathlib import Path import re import sys from typing import Any OFFICIAL_HOSTS = ("github.com/Lum1104/Understand-Anything", "understand-anything.com", "github.com/debpalash/OmniVoice-Studio", "omnivoice-studio") STAGES = ("source", "script", "voice", "render", "final") UNSUPPORTED_CLAIMS = ("instantly understand the entire", "never read code again") def load_json(path: Path, errors: list[str]) -> dict[str, Any]: if not path.is_file(): errors.append(f"missing required JSON artifact: {path.name}") return {} try: return json.loads(path.read_text(encoding="utf-8")) except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc: errors.append(f"invalid JSON artifact {path.name}: {exc}") return {} def require_file(project: Path, relative: str, errors: list[str]) -> Path: path = project / relative if not path.is_file(): errors.append(f"missing required artifact: {relative}") return path def official(url: str) -> bool: return any(host in url for host in OFFICIAL_HOSTS) def validate_source(project: Path, state: dict[str, Any], errors: list[str]) -> None: if state.get("template_selection", {}).get("id") != "repo-update-fullframe": errors.append("approved repo-update-fullframe template is not selected in run-state.json") if state.get("template_selection", {}).get("tier") != "approved": errors.append("selected template is not recorded as approved") record = load_json(project / "capture" / "source-record.json", errors) primary_url = record.get("primary_source", {}).get("url", "") if not official(primary_url): errors.append("primary source is not the official Understand Anything source") claims = record.get("claims", []) if not claims: errors.append("source record has no attributable claims") for claim in claims: if claim.get("role") != "SOURCE PROOF" or not official(claim.get("source_url", "")): errors.append(f"claim lacks official SOURCE PROOF attribution: {claim.get('id', 'unknown')}") assets = record.get("assets", []) if not any(asset.get("role") == "SOURCE PROOF" for asset in assets): errors.append("source record has no SOURCE PROOF visual asset") for asset in assets: if asset.get("kind") == "reference-video" and asset.get("public") is True: errors.append("public output may not contain a reference-video capture") if asset.get("public") is True: path = project / str(asset.get("path", "")) if not path.is_file(): errors.append(f"declared public visual asset is missing: {asset.get('path')}") policy = record.get("reference_video_policy", {}) if policy.get("used_in_public_output") is not False: errors.append("source record must explicitly prohibit reference-video reuse in public output") plan = require_file(project, "capture/visual-evidence-plan.md", errors) if plan.is_file(): text = plan.read_text(encoding="utf-8") for marker in ("SOURCE PROOF", "EXPLAINER", "focused crop", "reference video"): if marker.lower() not in text.lower(): errors.append(f"visual evidence plan missing marker: {marker}") def validate_script(project: Path, errors: list[str]) -> None: script = require_file(project, "SCRIPT.md", errors) publish = require_file(project, "PUBLISH.md", errors) storyboard = require_file(project, "STORYBOARD.md", errors) voiceover = require_file(project, "voiceover.txt", errors) if script.is_file(): text = script.read_text(encoding="utf-8") for marker in ("## Publish Title", "## Source Notes", "## Voiceover", "## Truth Boundary"): if marker not in text: errors.append(f"SCRIPT.md missing marker: {marker}") if any(phrase in text.lower() for phrase in UNSUPPORTED_CLAIMS): errors.append("SCRIPT.md includes an unsupported product promise") if publish.is_file(): text = publish.read_text(encoding="utf-8") for marker in ("## Publish Title", "## Post Caption", "## Hashtags"): if marker not in text: errors.append(f"PUBLISH.md missing marker: {marker}") hashtags = re.findall(r"#[\w-]+", text.partition("## Hashtags")[2], flags=re.UNICODE) if len(hashtags) != 5: errors.append("PUBLISH.md must contain exactly 5 publishable hashtags") if storyboard.is_file(): text = storyboard.read_text(encoding="utf-8") for marker in ("SOURCE PROOF", "EXPLAINER", "CTA"): if marker not in text: errors.append(f"STORYBOARD.md missing declared scene role: {marker}") if voiceover.is_file() and len(voiceover.read_text(encoding="utf-8").strip()) < 80: errors.append("voiceover.txt is too short for a standard pilot narration") def validate_voice(project: Path, state: dict[str, Any], errors: list[str]) -> None: voice = state.get("stages", {}).get("voice", {}) if voice.get("status") != "complete": errors.append("voice checkpoint is not complete") return metadata = voice.get("metadata", {}) try: duration = float(metadata.get("duration_seconds", "")) except (TypeError, ValueError): duration = 0 if duration <= 0: errors.append("voice checkpoint has no measured narration duration") provider = metadata.get("provider", "") profile = metadata.get("profile", "") if provider != "OmniVoice" or profile != "76d11b96": approved = any(event.get("type") == "fallback_approved" and event.get("stage") == "voice" for event in state.get("events", [])) if not approved: errors.append("non-preferred narration provider has no recorded fallback approval") narration = require_file(project, "narration.wav", errors) if narration.is_file() and narration.stat().st_size == 0: errors.append("narration.wav is empty") def validate_render(project: Path, state: dict[str, Any], errors: list[str]) -> None: for stage in ("composition", "render"): if state.get("stages", {}).get(stage, {}).get("status") != "complete": errors.append(f"{stage} checkpoint is not complete") composition = require_file(project, "index.html", errors) ua_render = project / "renders/understand-anything-ai-news-vi.mp4" ov_render = project / "renders/OmniVoice-Studio.mp4" ov_lower_render = project / "renders/omnivoice-studio.mp4" if not (ua_render.is_file() or ov_render.is_file() or ov_lower_render.is_file()): errors.append("missing required artifact: renders/OmniVoice-Studio.mp4") contact_candidates = list((project / "capture").glob("*contact-sheet*")) if (project / "capture").exists() else [] if not contact_candidates: errors.append("render evidence is missing a contact sheet") if composition.is_file(): text = composition.read_text(encoding="utf-8") for marker in ("SOURCE PROOF", "EXPLAINER"): if marker not in text: errors.append(f"composition missing media truth marker: {marker}") def validate_final(project: Path, state: dict[str, Any], errors: list[str]) -> None: qa_path = require_file(project, "capture/qa-report.json", errors) handoff = require_file(project, "HANDOFF.md", errors) if state.get("stages", {}).get("qa", {}).get("status") != "complete": errors.append("QA checkpoint is not complete") if qa_path.is_file(): qa = load_json(qa_path, errors) checks = qa.get("checks", {}) required = ("source_traceability", "crop_legibility", "truth_roles", "vietnamese_rendering", "reference_video_nonreuse", "safe_zones", "timing_sync", "fallback_disclosure") for marker in required: if checks.get(marker) != "pass": errors.append(f"QA report has not passed check: {marker}") if handoff.is_file(): text = handoff.read_text(encoding="utf-8") for marker in ("Publish Title", "Hashtags"): if marker not in text: errors.append(f"HANDOFF.md missing post-render publish metadata: {marker}") hashtags = re.findall(r"#[\w-]+", text.partition("Hashtags")[2], flags=re.UNICODE) if len(hashtags) != 5: errors.append("HANDOFF.md must repeat exactly 5 publishable hashtags") def validate(project: Path, target: str) -> dict[str, Any]: errors: list[str] = [] state = load_json(project / "run-state.json", errors) if state: validate_source(project, state, errors) if STAGES.index(target) >= STAGES.index("script"): validate_script(project, errors) if STAGES.index(target) >= STAGES.index("voice"): validate_voice(project, state, errors) if STAGES.index(target) >= STAGES.index("render"): validate_render(project, state, errors) if target == "final": validate_final(project, state, errors) return { "project": str(project), "stage": target, "checks": "pass" if not errors else "fail", "errors": errors, } def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--project-dir", required=True) parser.add_argument("--stage", choices=STAGES, required=True) args = parser.parse_args() report = validate(Path(args.project_dir).resolve(), args.stage) print(json.dumps(report, ensure_ascii=False, indent=2)) return 0 if report["checks"] == "pass" else 1 if __name__ == "__main__": sys.exit(main())