"""Pilot production package for exactly one controlled V2 video.""" from __future__ import annotations from dataclasses import dataclass import json from typing import Any from autoshorts.compliance.ai_disclosure import AIDisclosureReview from autoshorts.production.resource_gate import ( MediaGenerationGate, RenderApprovalState, ResourceCostTier, evaluate_media_generation_gate, ) from autoshorts.review.workflow_v2 import classify_candidate_v2 from autoshorts.strategy.spine_v2 import ReviewPackageV2, VideoCandidateV2, build_review_package_v2 @dataclass(frozen=True) class WanScenePromptPlan: scene_id: str duration_seconds: float visual_goal: str prompt: str negative_prompt: str camera_motion: str required_overlay_text: tuple[str, ...] renderer_owned_overlay_elements: tuple[str, ...] safe_zone_notes: str transition_to_next_scene: str def to_dict(self) -> dict[str, Any]: return { "scene_id": self.scene_id, "duration_seconds": self.duration_seconds, "visual_goal": self.visual_goal, "prompt": self.prompt, "negative_prompt": self.negative_prompt, "camera_motion": self.camera_motion, "required_overlay_text": list(self.required_overlay_text), "renderer_owned_overlay_elements": list(self.renderer_owned_overlay_elements), "safe_zone_notes": self.safe_zone_notes, "transition_to_next_scene": self.transition_to_next_scene, } @dataclass(frozen=True) class PilotProductionPackage: candidate_id: str version: str script_hash: str approval_state: RenderApprovalState voiceover_text: str german_review_summary: str hook: str first_frame_description: str visible_fail: str anti_example: str mechanism: str fix: str before_after: str takeaway: str save_reason: str share_reason: str follow_reason: str claim_gate_result: dict[str, Any] ai_disclosure_result: dict[str, Any] resource_gate_result: MediaGenerationGate platform_fit: dict[str, str] social_copy: dict[str, str] hashtags: dict[str, tuple[str, ...]] caption_mode_recommendation: str tts_plan: dict[str, str] subtitle_alignment_plan: dict[str, str] shot_plan: tuple[dict[str, Any], ...] render_manifest_draft: dict[str, Any] wan_prompt_plan: tuple[WanScenePromptPlan, ...] styleframe_prompt_draft: dict[str, str] negative_prompts: tuple[str, ...] asset_list: tuple[dict[str, str], ...] estimated_render_cost_time_class: str deterministic_preview: dict[str, Any] score_breakdown: dict[str, Any] approval_commands: dict[str, str] def to_dict(self) -> dict[str, Any]: return { "candidate_id": self.candidate_id, "version": self.version, "script_hash": self.script_hash, "approval_state": self.approval_state.value, "voiceover_text": self.voiceover_text, "german_review_summary": self.german_review_summary, "hook": self.hook, "first_frame_description": self.first_frame_description, "visible_fail": self.visible_fail, "anti_example": self.anti_example, "mechanism": self.mechanism, "fix": self.fix, "before_after": self.before_after, "takeaway": self.takeaway, "save_reason": self.save_reason, "share_reason": self.share_reason, "follow_reason": self.follow_reason, "claim_gate_result": self.claim_gate_result, "ai_disclosure_result": self.ai_disclosure_result, "resource_gate_result": self.resource_gate_result.to_dict(), "platform_fit": dict(self.platform_fit), "social_copy": dict(self.social_copy), "hashtags": {key: list(value) for key, value in self.hashtags.items()}, "caption_mode_recommendation": self.caption_mode_recommendation, "tts_plan": dict(self.tts_plan), "subtitle_alignment_plan": dict(self.subtitle_alignment_plan), "shot_plan": list(self.shot_plan), "render_manifest_draft": dict(self.render_manifest_draft), "wan_prompt_plan": [scene.to_dict() for scene in self.wan_prompt_plan], "styleframe_prompt_draft": dict(self.styleframe_prompt_draft), "negative_prompts": list(self.negative_prompts), "asset_list": list(self.asset_list), "estimated_render_cost_time_class": self.estimated_render_cost_time_class, "deterministic_preview": self.deterministic_preview, "score_breakdown": self.score_breakdown, "approval_commands": dict(self.approval_commands), } def to_json(self) -> str: return json.dumps(self.to_dict(), ensure_ascii=False, separators=(",", ":")) @dataclass(frozen=True) class PilotSelection: main_pilot: VideoCandidateV2 backup: VideoCandidateV2 deferred: VideoCandidateV2 def select_pilot_candidates(candidates: tuple[VideoCandidateV2, ...]) -> PilotSelection: by_series = {candidate.series_id: candidate for candidate in candidates} return PilotSelection( main_pilot=by_series["ai_output_autopsy"], backup=by_series["digital_red_flags"], deferred=by_series["workflow_teardown"], ) def _voiceover_for(candidate: VideoCandidateV2) -> str: return " ".join([ candidate.hook, "Look at the bad output: it sounds polished, but it missed the one thing the viewer needed.", candidate.visible_fail, candidate.mechanism, candidate.fix, candidate.before_after, candidate.takeaway, ]) def _shot_plan(candidate: VideoCandidateV2) -> tuple[dict[str, Any], ...]: return ( {"scene_id": "s01_fail", "duration_seconds": 5, "purpose": "instant fail recognition", "visual": candidate.first_frame, "overlay_owner": "renderer"}, {"scene_id": "s02_bad_output", "duration_seconds": 7, "purpose": "show anti-example", "visual": candidate.anti_example or candidate.plausible_nonsense_example, "overlay_owner": "renderer"}, {"scene_id": "s03_mechanism", "duration_seconds": 9, "purpose": "reveal mechanism", "visual": candidate.mechanism, "overlay_owner": "renderer"}, {"scene_id": "s04_fix", "duration_seconds": 9, "purpose": "show rule/fix", "visual": candidate.fix, "overlay_owner": "renderer"}, {"scene_id": "s05_takeaway", "duration_seconds": 8, "purpose": "memorable close", "visual": candidate.before_after + " " + candidate.takeaway, "overlay_owner": "renderer"}, ) def _wan_prompt_plan(candidate: VideoCandidateV2) -> tuple[WanScenePromptPlan, ...]: negative = "no readable text, no logos, no captions, no numbers, no charts with labels, no brand names, no watermark, no UI text baked into image" return ( WanScenePromptPlan("s01_fail", 5, "Create instant tension between messy transcript and useless summary.", "Vertical 9:16 cinematic abstract office desk, messy paper transcript blocks on one side, clean but suspicious summary card shape on the other, high contrast, modern minimal motion graphics feel, no readable text.", negative, "slow push-in with slight parallax", (), ("hook caption", "red USELESS stamp", "split-screen divider"), "Keep center clean; leave top and bottom subtitle-safe zones empty.", "Cut on red stamp pulse to bad-output closeup."), WanScenePromptPlan("s02_bad_output", 7, "Make the plausible bad AI output feel recognizable without text baked in.", "Close-up of generic glowing document card with blurred lines, confident polished look but hollow center, office meeting notes in background, subtle warning glow, no readable words.", negative, "micro dolly left to right", (), ("bad summary callouts", "missing owner/blocker/decision labels"), "Avoid important detail in bottom 20 percent for captions.", "Callout lines collapse into compression funnel."), WanScenePromptPlan("s03_mechanism", 9, "Reveal equal compression mechanism.", "Abstract stream of many meeting note fragments flowing into a single compressed box, all fragments treated equally, one decision node ignored, clean tech editorial style, no text.", negative, "top-down slight orbit", (), ("decision target node", "equal compression arrows", "mechanism label"), "Leave right third for renderer-owned mechanism labels.", "Decision target node lights up and becomes the fix card."), WanScenePromptPlan("s04_fix", 9, "Show the fix as a structured decision-evidence route.", "Minimal workflow route with four empty nodes representing blockers, owners, risks, next steps, structured path replacing messy notes, no readable labels.", negative, "smooth track along route", (), ("blockers", "owners", "risks", "next steps", "decision evidence"), "Keep nodes away from extreme edges.", "Route resolves into before/after split."), WanScenePromptPlan("s05_takeaway", 8, "Close with before/after contrast and memorable rule.", "Before-after abstract split: cloudy generic summary dissolves into clear decision route with one highlighted target, satisfying clean ending, no embedded text.", negative, "slow pullback, final hold", (), ("Before", "After", "Don't ask for a summary. Ask for decision evidence."), "Final hold must leave center-lower space for takeaway caption.", "End on static frame for manual review."), ) def build_pilot_production_package( candidate: VideoCandidateV2, review_package: ReviewPackageV2, *, approval_state: RenderApprovalState = RenderApprovalState.READY_FOR_REVIEW, ai_disclosure_review: AIDisclosureReview | None = None, ) -> PilotProductionPackage: from autoshorts.claims.fact_gate import validate_claims_for_candidate from autoshorts.compliance.ai_disclosure import AIDisclosureReview as Review, DisclosureValue, SyntheticMediaType, validate_ai_disclosure_review claim_result = validate_claims_for_candidate(candidate) disclosure = ai_disclosure_review or Review.text_only_assistance("Script assistance only; no realistic synthetic person/event/voice generated at this stage.") disclosure_result = validate_ai_disclosure_review(disclosure) gate = evaluate_media_generation_gate( candidate=candidate, review_package=review_package, requested_resource_tier=ResourceCostTier.DETERMINISTIC_PREVIEW, approval_state=approval_state, approved_candidate_id=candidate.candidate_id, approved_version=candidate.version, approved_script_hash=review_package.script_hash, ai_disclosure_review=disclosure, ) voiceover = _voiceover_for(candidate) command_base = f"{candidate.candidate_id} {candidate.version} {review_package.script_hash}" return PilotProductionPackage( candidate_id=candidate.candidate_id, version=candidate.version, script_hash=review_package.script_hash, approval_state=approval_state, voiceover_text=voiceover, german_review_summary="Pilot für AI Output Autopsy: Eine scheinbar gute Meeting-Zusammenfassung verfehlt das Entscheidungsziel. Der Fix ist, nicht nach einer Zusammenfassung zu fragen, sondern nach Decision Evidence: blocker, owners, risks und next steps.", hook=candidate.hook, first_frame_description=candidate.first_frame, visible_fail=candidate.visible_fail, anti_example=candidate.anti_example or candidate.plausible_nonsense_example, mechanism=candidate.mechanism, fix=candidate.fix, before_after=candidate.before_after, takeaway=candidate.takeaway, save_reason=candidate.save_reason, share_reason=candidate.share_reason, follow_reason=candidate.follow_reason, claim_gate_result={"is_valid": claim_result.is_valid, "errors": list(claim_result.errors)}, ai_disclosure_result={"is_valid": disclosure_result.is_valid, "errors": list(disclosure_result.errors), "synthetic_media_type": disclosure.synthetic_media_type.value, "reason": disclosure.reason}, resource_gate_result=gate, platform_fit=review_package.platform_fit, social_copy={ "youtube_title": "Your AI Meeting Summary Is Missing the Decision", "youtube_description": "A polished AI summary can still be useless if it has no decision target. Ask for blockers, owners, risks and next steps instead.", "tiktok_caption": "Your AI summary sounds smart. But did it protect the decision?", "instagram_caption": "The fix is not a better summary. It is a clearer decision target.", "linkedin_caption": candidate.linkedin_caption, }, hashtags={ "youtube_shorts": ("#AIWorkflow", "#Productivity", "#MeetingNotes"), "tiktok": ("#AITips", "#WorkTok", "#ProductivityHack"), "instagram_reels": ("#WorkflowDesign", "#Productivity", "#AIForWork"), "linkedin_manual": ("#AIProductivity", "#WorkflowDesign", "#DecisionMaking"), }, caption_mode_recommendation="word-by-word", tts_plan={"language": "en", "voice_direction": "clear, calm, precise, not hype", "pace": "145-155 wpm", "notes": "Leave micro-pauses after hook and takeaway for subtitle readability."}, subtitle_alignment_plan={"mode": "word-by-word", "method": "forced alignment after TTS render", "highlight_style": "Hormozi/TikTok-style active word pulse", "quality_gate": "No caption segment may drift beyond spoken word timing."}, shot_plan=_shot_plan(candidate), render_manifest_draft={"final_render_enabled": False, "wan_calls_enabled": False, "ai_image_generation_enabled": False, "renderer_owned_text": True, "caption_mode": "word-by-word", "target_duration_seconds": candidate.duration_seconds, "output": "deterministic preview/timeline only"}, wan_prompt_plan=_wan_prompt_plan(candidate), styleframe_prompt_draft={"status": "not_generated", "approval_required": f"APPROVE_STYLEFRAME {command_base}", "prompt": "Single 9:16 styleframe showing messy meeting transcript versus polished but useless summary card; no readable text; renderer owns all captions, labels and facts."}, negative_prompts=("readable text", "logos", "brand names", "captions", "numbers", "watermarks", "UI labels", "embedded facts"), asset_list=( {"asset": "tts_voiceover", "status": "planned", "owner": "deterministic/local until approved"}, {"asset": "subtitle_alignment", "status": "planned", "owner": "renderer"}, {"asset": "overlay_labels", "status": "planned", "owner": "renderer"}, {"asset": "wan_background_motion", "status": "blocked_until_final_render_approval", "owner": "wan_2_2"}, ), estimated_render_cost_time_class="medium", deterministic_preview={"type": "text_timeline", "enabled": True, "frames": list(_shot_plan(candidate))}, score_breakdown=review_package.score.to_dict(), approval_commands={"styleframe": f"APPROVE_STYLEFRAME {command_base}", "final_render": f"APPROVED_FOR_FINAL_RENDER {command_base}", "revise": f"REVISE {candidate.candidate_id} \"Änderungswunsch\"", "reject": f"REJECT {candidate.candidate_id} \"Grund\""}, ) def build_default_pilot_selection_and_package() -> tuple[PilotSelection, PilotProductionPackage]: from autoshorts.strategy.spine_v2 import build_demo_candidates_v2 candidates = build_demo_candidates_v2(limit=3) selection = select_pilot_candidates(candidates) review_package = build_review_package_v2(selection.main_pilot) return selection, build_pilot_production_package(selection.main_pilot, review_package)