""" Video Generation Pipeline - Layers 4 v1/v2/v3 (Legacy): 4a (concepts) → 4b (script/production_spec) → 4c (visuals/assets) v4 (Current): Director → Evaluator → (Revise if needed) → Production Ready The v4 architecture uses: - Script Director (Opus): Takes premise + research → produces time-coded multi-track script - Script Evaluator (Opus): Evaluates script against cognitive science criteria → verdict + revision instructions - Max 1 revision iteration if not PRODUCTION_READY """ import json import re import subprocess import time from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Optional, Tuple from prompts import get_prompt, get_prompt_metadata, get_revision_template, get_channel_director_prompt, CHANNEL_SPECIFIC_DIRECTORS # Database operations for opportunities from db_opportunities import update_opportunity_video_gen_by_cluster # Try to import channel config try: from channel_config import get_channel except ImportError: get_channel = None # Output directory OUTPUT_DIR = Path(__file__).parent / "output" # ============================================================================= # OUTPUT ORGANIZATION - Channel folders with premise-named files # ============================================================================= def sanitize_premise_for_filename(premise: str, max_length: int = 50) -> str: """ Convert a premise string to a safe filename component. Examples: "The Tobacco Company That Bought the Women's Movement" → "tobacco-company-bought-womens-movement" "Why You Can't Stop Checking Your Phone" → "why-you-cant-stop-checking-phone" """ # Lowercase name = premise.lower() # Remove common words that don't add meaning stopwords = {'the', 'a', 'an', 'of', 'to', 'in', 'for', 'on', 'with', 'at', 'by', 'from', 'that', 'this', 'is', 'are', 'was', 'were', 'be', 'been', 'being', 'have', 'has', 'had', 'do', 'does', 'did', 'will', 'would', 'could', 'should', 'may', 'might', 'must', 'shall', 'can', 'and', 'but', 'or', 'so', 'yet', 'both', 'either', 'neither', 'not', 'only', 'own', 'same', 'than', 'too', 'very', 'just', 'also', 'now', 'here', 'there', 'when', 'where', 'why', 'how', 'all', 'each', 'every', 'any', 'some', 'no', 'none', 'one', 'two', 'three', 'first', 'last', 'other', 'new', 'old', 'good', 'bad', 'great', 'little', 'big', 'small', 'long', 'short', 'high', 'low', 'right', 'wrong', 'true', 'false', 'yes', 'about', 'after', 'before', 'into', 'through', 'during', 'without', 'between', 'under', 'over', 'again', 'then', 'once', 'out', 'up', 'down', 'off', 'away', 'back', 'even', 'still', 'well', 'much', 'many', 'more', 'most', 'less', 'least', 'never', 'always', 'often', 'sometimes', 'usually', 'really', 'actually', 'probably', 'maybe', 'perhaps', 'it', 'its'} # Replace non-alphanumeric with spaces name = re.sub(r'[^a-z0-9\s]', ' ', name) # Split into words and filter words = [w for w in name.split() if w and w not in stopwords] # If we filtered too aggressively, use first few original words if len(words) < 2: words = [w for w in re.sub(r'[^a-z0-9\s]', ' ', premise.lower()).split() if w][:5] # Join with hyphens and truncate name = '-'.join(words) # Truncate to max length at word boundary if len(name) > max_length: name = name[:max_length].rsplit('-', 1)[0] return name or 'untitled' def get_channel_output_dir(channel_id: str) -> Path: """ Get the output directory for a specific channel, creating it if needed. Returns: Path like: output/one_minute_history/ """ channel_dir = OUTPUT_DIR / channel_id channel_dir.mkdir(parents=True, exist_ok=True) return channel_dir def get_video_output_path(channel_id: str, premise: str, timestamp: datetime = None) -> Path: """ Generate the output path for a video script. Args: channel_id: Channel identifier (e.g., "one_minute_history") premise: The video premise/title timestamp: Optional timestamp (defaults to now) Returns: Path like: output/one_minute_history/2025-02-15_14-25_tobacco-company-bought-womens-movement.json """ if timestamp is None: timestamp = datetime.now() channel_dir = get_channel_output_dir(channel_id) # Format: YYYY-MM-DD_HH-MM_premise-slug.json time_str = timestamp.strftime("%Y-%m-%d_%H-%M") premise_slug = sanitize_premise_for_filename(premise) filename = f"{time_str}_{premise_slug}.json" return channel_dir / filename def save_video_output( opportunity: Dict[str, Any], channel_id: str, timestamp: datetime = None ) -> Path: """ Save a completed video generation result to the channel's output folder. The output file contains: - input: Original opportunity data (premise, research, etc.) - output: Script, evaluation, metadata - pipeline: Iteration history, timing, versions Args: opportunity: The completed opportunity dict with script and evaluation channel_id: Channel identifier timestamp: Optional timestamp for the filename Returns: Path to the saved file """ premise = opportunity.get("premise", "Untitled") output_path = get_video_output_path(channel_id, premise, timestamp) # Extract narration script from the final script script = opportunity.get("script", {}) narration_script = extract_narration_script(script) if script else "" # Structure the output for clarity structured_output = { "meta": { "channel_id": channel_id, "premise": premise, "generated_at": datetime.now().isoformat(), "production_ready": opportunity.get("production_ready", False), "iteration_count": opportunity.get("iteration_count", 0), "final_verdict": opportunity.get("evaluation", {}).get("verdict", "UNKNOWN"), "final_score": opportunity.get("evaluation", {}).get("composite_score", 0), }, "narration_script": narration_script, "input": { "premise": premise, "first_frame": opportunity.get("first_frame"), "trigger_map": opportunity.get("trigger_map"), "opening_hook": opportunity.get("opening_hook"), "core_reveal": opportunity.get("core_reveal"), "depth_check": opportunity.get("depth_check"), "emotional_payoff": opportunity.get("emotional_payoff"), "structure": opportunity.get("structure"), "target_audience": opportunity.get("target_audience"), "weighted_score": opportunity.get("weighted_score"), "verdict": opportunity.get("verdict"), "research_word_count": opportunity.get("research_word_count"), "research_source_count": opportunity.get("research_source_count"), "research_report_content": opportunity.get("research_report_content"), }, "output": { "script": opportunity.get("script"), "evaluation": opportunity.get("evaluation"), }, "pipeline": { "director_metadata": opportunity.get("director_metadata"), "evaluator_metadata": opportunity.get("evaluator_metadata"), "all_iterations": opportunity.get("all_iterations"), } } # Save to file with open(output_path, "w") as f: json.dump(structured_output, f, indent=2, default=str) print(f" [Saved] {output_path.relative_to(OUTPUT_DIR)}") return output_path def list_channel_outputs(channel_id: str) -> List[Path]: """ List all video output files for a channel, sorted by timestamp (newest first). Returns: List of Path objects to JSON files """ channel_dir = get_channel_output_dir(channel_id) files = sorted(channel_dir.glob("*.json"), reverse=True) return files def extract_narration_script(script: Dict[str, Any]) -> str: """ Extract all voiceover text from a script into a single narration string. Pulls voiceover from every segment in every beat, in order. Separates segments with blank lines, beats with --- dividers. Preserves all [beat], [pause], [thinking], [drops voice] markers. Args: script: The script dict containing beats with segments Returns: Single string ready for voice actor or TTS to read top to bottom """ beats = script.get("beats", []) if not beats: return "" beat_texts = [] for beat in beats: segments = beat.get("segments", []) segment_texts = [] for segment in segments: voiceover = segment.get("voiceover", "") if voiceover and voiceover.strip(): segment_texts.append(voiceover.strip()) if segment_texts: # Join segments within a beat with double newlines beat_text = "\n\n".join(segment_texts) beat_texts.append(beat_text) # Join beats with --- dividers narration = "\n\n---\n\n".join(beat_texts) return narration # ============================================================================= # OPPORTUNITIES FILE MANAGEMENT (legacy, for batch processing) # ============================================================================= def get_opportunities_path(channel_id: str = None) -> Path: """ Get the path for opportunities JSON file. Args: channel_id: If specified, returns channel-specific path. If None, returns the generic path (backward compatibility). Returns: Path to the opportunities JSON file """ if channel_id: return OUTPUT_DIR / f"analysis_opportunities_{channel_id}.json" return OUTPUT_DIR / "analysis_opportunities.json" def get_latest_opportunities_file(channel_id: str = None) -> Path: """ Find the latest analysis_opportunities file for the given channel. Args: channel_id: If specified, looks for channel-specific file. If None, looks for any opportunities file (backward compatibility). Returns: Path to the latest opportunities JSON file """ if channel_id: # Look for channel-specific file channel_file = get_opportunities_path(channel_id) if channel_file.exists(): return channel_file return None # Backward compatibility: look for timestamped files or base file pattern = "analysis_opportunities_*.json" timestamped_files = sorted(OUTPUT_DIR.glob(pattern), reverse=True) if timestamped_files: return timestamped_files[0] # Fall back to base file base_file = OUTPUT_DIR / "analysis_opportunities.json" if base_file.exists(): return base_file return None def save_opportunities_timestamped(opportunities_data: Dict[str, Any], channel_id: str = None) -> Path: """ Save opportunities to channel-specific file (and optionally a timestamped backup). Args: opportunities_data: The full opportunities data dict channel_id: If specified, saves to channel-specific file. Returns: Path to the saved file """ # Determine the primary output path output_path = get_opportunities_path(channel_id) # Save to the channel-specific (or generic) file with open(output_path, "w") as f: json.dump(opportunities_data, f, indent=2, default=str) # Also save a timestamped backup for history timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") if channel_id: timestamped_path = OUTPUT_DIR / f"analysis_opportunities_{channel_id}_{timestamp}.json" else: timestamped_path = OUTPUT_DIR / f"analysis_opportunities_{timestamp}.json" with open(timestamped_path, "w") as f: json.dump(opportunities_data, f, indent=2, default=str) return output_path def call_claude(prompt: str, model: str = "opus", timeout: int = 300) -> Tuple[Dict[str, Any], float]: """ Call Claude CLI and parse JSON response. Args: prompt: The full prompt to send model: Model name ("opus" or "sonnet") timeout: Timeout in seconds Returns: Tuple of (parsed_json_dict, duration_seconds) Raises: RuntimeError: If Claude CLI fails ValueError: If response is not valid JSON """ start_time = time.time() result = subprocess.run( ["claude", "--print", "--model", model, "-p", "-"], input=prompt, capture_output=True, text=True, timeout=timeout ) duration = time.time() - start_time if result.returncode != 0: raise RuntimeError(f"Claude CLI failed (exit {result.returncode}): {result.stderr[:200]}") response_text = result.stdout.strip() # Extract JSON from response - try multiple strategies parsed = None # Strategy 1: Look for JSON in markdown code block code_block_match = re.search(r'```(?:json)?\s*(\{.*?\})\s*```', response_text, re.DOTALL) if code_block_match: try: parsed = json.loads(code_block_match.group(1)) except json.JSONDecodeError: pass # Strategy 2: Find outermost JSON object (greedy match from first { to last }) if not parsed: json_match = re.search(r'\{.*\}', response_text, re.DOTALL) if json_match: try: parsed = json.loads(json_match.group()) except json.JSONDecodeError as e: # Debug: show where parsing failed error_pos = e.pos if hasattr(e, 'pos') else 'unknown' json_text = json_match.group() print(f" [DEBUG] JSON parse error at position {error_pos}: {e}") print(f" [DEBUG] JSON length: {len(json_text)} chars") if hasattr(e, 'pos') and e.pos: context_start = max(0, e.pos - 50) context_end = min(len(json_text), e.pos + 50) print(f" [DEBUG] Context around error: ...{json_text[context_start:context_end]}...") # Strategy 3: Find JSON that starts at beginning of a line if not parsed: line_json_match = re.search(r'^\s*(\{.*\})\s*$', response_text, re.DOTALL | re.MULTILINE) if line_json_match: try: parsed = json.loads(line_json_match.group(1)) except json.JSONDecodeError: pass if not parsed: raise ValueError(f"No JSON found in response. First 500 chars: {response_text[:500]}") return parsed, duration # ============================================================================= # V4 ARCHITECTURE: SCRIPT DIRECTOR + EVALUATOR # ============================================================================= def layer4_director( opportunity: Dict[str, Any], channel_id: str = None, target_duration: int = 90, revision_instructions: str = None ) -> Dict[str, Any]: """ Layer 4 Director: Generate time-coded multi-track script from validated premise + research. Uses channel-specific prompts when available (v4 architecture). Each channel has its own voice, beat structure, pacing profile, and visual language. Args: opportunity: Opportunity dict with Layer 3 analysis + Layer 3.5 research channel_id: Channel ID for metadata lookup and prompt selection target_duration: Target video duration in seconds (60-120) revision_instructions: If provided, includes revision section in prompt Returns: Dict with script object and execution_metadata """ # Load channel context channel_name = "General" channel_description = "" competitive_gap = "" if channel_id and get_channel: channel = get_channel(channel_id) if channel: channel_name = channel.name channel_description = channel.description competitive_gap = channel.to_dict().get("competitive_gap", "") # Extract data from opportunity for template # Handle nested structures gracefully trigger_map = opportunity.get("trigger_map", {}) if isinstance(trigger_map, dict): trigger_map_str = json.dumps(trigger_map, indent=2) else: trigger_map_str = str(trigger_map) structure = opportunity.get("structure", "") if isinstance(structure, list): structure_str = "\n".join([f"- {s}" for s in structure]) elif isinstance(structure, dict): structure_str = json.dumps(structure, indent=2) else: structure_str = str(structure) target_audience = opportunity.get("target_audience", "") if isinstance(target_audience, dict): target_audience_str = json.dumps(target_audience, indent=2) else: target_audience_str = str(target_audience) # Check if we have a channel-specific prompt use_channel_specific = channel_id in CHANNEL_SPECIFIC_DIRECTORS if use_channel_specific: # Use channel-specific prompt (v4 split architecture) # Note: channel_id passed as positional arg, not in kwargs prompt = get_channel_director_prompt( channel_id, channel_name=channel_name, channel_description=channel_description, competitive_gap=competitive_gap, premise=opportunity.get("premise", opportunity.get("suggested_title", "Unknown")), first_frame=opportunity.get("first_frame", ""), trigger_map=trigger_map_str, opening_hook=opportunity.get("opening_hook", ""), core_reveal=opportunity.get("core_reveal", ""), depth_check=opportunity.get("depth_check", ""), emotional_payoff=opportunity.get("emotional_payoff", ""), structure=structure_str, target_audience=target_audience_str, weighted_score=opportunity.get("weighted_score", 0), verdict=opportunity.get("verdict", ""), target_duration=target_duration, research_report_content=opportunity.get("research_report_content", "No research available"), revision_instructions=revision_instructions ) metadata = get_prompt_metadata("layer4_director", version="v4", channel=channel_id) else: # Fall back to unified prompt prompt_template = get_prompt("layer4_director", version="v4") metadata = get_prompt_metadata("layer4_director", version="v4") # Build revision section if needed revision_section = "" if revision_instructions: revision_template = get_revision_template("v4") revision_section = revision_template.format(revision_instructions=revision_instructions) # Fill in the template prompt = prompt_template.format( channel_name=channel_name, channel_id=channel_id or "general", channel_description=channel_description, competitive_gap=competitive_gap, premise=opportunity.get("premise", opportunity.get("suggested_title", "Unknown")), first_frame=opportunity.get("first_frame", ""), trigger_map=trigger_map_str, opening_hook=opportunity.get("opening_hook", ""), core_reveal=opportunity.get("core_reveal", ""), depth_check=opportunity.get("depth_check", ""), emotional_payoff=opportunity.get("emotional_payoff", ""), structure=structure_str, target_audience=target_audience_str, weighted_score=opportunity.get("weighted_score", 0), verdict=opportunity.get("verdict", ""), target_duration=target_duration, research_report_content=opportunity.get("research_report_content", "No research available"), revision_section=revision_section ) # Call Claude Opus response, duration = call_claude(prompt, model="opus", timeout=600) return { "script": response, "execution_metadata": { "layer": "4_director", "model": metadata.get("model", "opus"), "prompt_version": "v4", "channel_specific": use_channel_specific, "channel": channel_id, "is_revision": revision_instructions is not None, "generated_at": datetime.now(timezone.utc).isoformat(), "duration_seconds": round(duration, 2) } } def layer4_evaluator( script: Dict[str, Any] ) -> Dict[str, Any]: """ Layer 4 Evaluator: Evaluate script against cognitive science criteria. Args: script: The script dict from layer4_director Returns: Dict with evaluation scores, verdict, and revision_instructions """ # Get prompt template prompt_template = get_prompt("layer4_evaluator", version="v4") metadata = get_prompt_metadata("layer4_evaluator", version="v4") # Serialize script to JSON for the prompt script_json = json.dumps(script, indent=2) # Fill in the template base_prompt = prompt_template.format(script_json=script_json) # Add JSON enforcement wrapper json_enforcement_prefix = """CRITICAL INSTRUCTION: You MUST respond with ONLY a valid JSON object. - Do NOT include any text before the JSON - Do NOT include any text after the JSON - Do NOT wrap the JSON in markdown code blocks - Start your response with the opening brace { - End your response with the closing brace } """ json_enforcement_suffix = """ REMINDER: Your response must be ONLY the JSON object. Start with { and end with }. No other text.""" prompt = json_enforcement_prefix + base_prompt + json_enforcement_suffix # Call Claude Opus response, duration = call_claude(prompt, model="opus", timeout=600) # Add execution metadata response["execution_metadata"] = { "layer": "4_evaluator", "model": metadata.get("model", "opus"), "prompt_version": "v4", "generated_at": datetime.now(timezone.utc).isoformat(), "duration_seconds": round(duration, 2) } return response # ============================================================================= # V4 VISUAL DECIDER - Intelligent visual type selection for each segment # ============================================================================= def layer4_visual_decider( script: Dict[str, Any], opportunity: Dict[str, Any], use_parallel: bool = False ) -> Dict[str, Any]: """ Layer 4 Visual Decider: Process each segment to determine optimal visual type. Takes a script from layer4_director and enhances each segment's visual field with structured visual specifications (IMAGE/VIDEO/DATA_VISUALIZATION). Args: script: The script dict from layer4_director opportunity: Full opportunity dict (for research access) use_parallel: Whether to process segments in parallel (faster but more API calls) Returns: Enhanced script with structured visual specifications """ from prompts.prompt_v4.layer4_visual_decider import format_prompt, METADATA start_time = time.time() premise = script.get("metadata", {}).get("premise", opportunity.get("premise", "")) research_excerpt = opportunity.get("research_report_content", "")[:2000] # First 2000 chars beats = script.get("beats", []) total_segments = sum(len(beat.get("segments", [])) for beat in beats) processed = 0 print(f" [4-VisualDecider] Processing {total_segments} segments...") # Flatten segments for processing all_segments = [] for beat in beats: for segment in beat.get("segments", []): all_segments.append({ "beat": beat, "segment": segment, "beat_name": beat.get("beat_name", "Unknown") }) # Process each segment for i, item in enumerate(all_segments): segment = item["segment"] beat_name = item["beat_name"] # Build segment ID beat_num = item["beat"].get("beat_number", 1) seg_idx = item["beat"].get("segments", []).index(segment) + 1 segment_id = f"b{beat_num}_s{seg_idx}" # Get context from adjacent segments prev_voiceover = all_segments[i-1]["segment"].get("voiceover", "") if i > 0 else "" next_voiceover = all_segments[i+1]["segment"].get("voiceover", "") if i < len(all_segments)-1 else "" # Format prompt prompt = format_prompt( segment_id=segment_id, time_start=segment.get("time_start", 0), time_end=segment.get("time_end", 5), beat_name=beat_name, voiceover=segment.get("voiceover", ""), previous_voiceover=prev_voiceover, next_voiceover=next_voiceover, research_excerpt=research_excerpt, premise=premise ) # Call Claude for visual decision try: result = call_claude_json(prompt, model="sonnet") if result and "decision" in result: # Store original visual as backup segment["visual_original"] = segment.get("visual", "") # Store the full result segment["visual_structured"] = result segment["visual_thinking"] = result.get("thinking", "") # Update main visual field for compatibility with downstream decision = result.get("decision", "IMAGE") if decision == "IMAGE": img = result.get("image", {}) segment["visual"] = img.get("search_query", segment["visual_original"]) segment["visual_era"] = img.get("era") elif decision == "VIDEO": vid = result.get("video", {}) segment["visual"] = vid.get("search_query", segment["visual_original"]) elif decision == "DATA_VISUALIZATION": data_viz = result.get("data_visualization", {}) segment["visual"] = f"DATA_VISUALIZATION: {data_viz.get('title', 'Chart')}" segment["visual_data"] = data_viz # Check if data was wanted but not available if result.get("if_data_not_available"): segment["visual_data_gap"] = result["if_data_not_available"] processed += 1 except Exception as e: print(f" ! Segment {segment_id} failed: {e}") # Keep original visual on failure segment["visual_structured"] = { "decision": "IMAGE", "image": {"search_query": segment.get("visual", "")[:50], "era": None} } duration = time.time() - start_time print(f" → Processed {processed}/{total_segments} segments in {duration:.1f}s") # Add visual decider metadata to script script["visual_decider_metadata"] = { "processed_segments": processed, "total_segments": total_segments, "duration_seconds": round(duration, 2), "model": "sonnet", "prompt_version": "v4" } return script def call_claude_json(prompt: str, model: str = "sonnet") -> Dict[str, Any]: """ Call Claude and parse JSON response. Uses the claude CLI for consistency with other calls. """ # Use subprocess to call claude CLI with --print flag for non-interactive mode try: result = subprocess.run( ["claude", "--print", prompt, "--model", model, "--output-format", "json"], capture_output=True, text=True, timeout=120 ) if result.returncode != 0: raise Exception(f"Claude CLI error: {result.stderr}") # Parse JSON from response response_text = result.stdout.strip() # The output-format json wraps the response in a structure with 'result' field try: wrapper = json.loads(response_text) # The wrapper has a 'result' field with the actual response if isinstance(wrapper, dict) and "result" in wrapper: content = wrapper["result"] else: content = response_text except json.JSONDecodeError: content = response_text # Remove markdown code fences if present content = str(content) content = re.sub(r'^```json\s*', '', content) content = re.sub(r'^```\s*', '', content) content = re.sub(r'\s*```$', '', content) content = content.strip() # Try to extract JSON from the content # Handle case where response might have text before/after JSON json_match = re.search(r'\{[\s\S]*\}', content) if json_match: return json.loads(json_match.group()) return json.loads(content) except subprocess.TimeoutExpired: raise Exception("Claude CLI timeout") except json.JSONDecodeError as e: raise Exception(f"JSON parse error: {e}") def run_layer4_v4_pipeline( opportunity: Dict[str, Any], channel_id: str = None, target_duration: int = 90, max_iterations: int = 2 ) -> Dict[str, Any]: """ Run the full v4 pipeline: Director → Evaluator → (Revise if needed) → Final Args: opportunity: Opportunity dict with Layer 3 analysis + Layer 3.5 research channel_id: Channel ID for metadata lookup target_duration: Target video duration in seconds (60-120) max_iterations: Maximum iterations (1 = no revision, 2 = one revision pass) Returns: Updated opportunity dict with: - script: The final script - evaluation: The final evaluation - director_metadata: Execution metadata from director - evaluator_metadata: Execution metadata from evaluator - iteration_count: Number of iterations used - production_ready: Boolean indicating if script passed evaluation """ iteration = 0 revision_instructions = None script = None evaluation = None all_iterations = [] while iteration < max_iterations: iteration += 1 print(f" [4-Director] Iteration {iteration}/{max_iterations}...") # Generate script try: director_result = layer4_director( opportunity=opportunity, channel_id=channel_id, target_duration=target_duration, revision_instructions=revision_instructions ) script = director_result["script"] director_metadata = director_result["execution_metadata"] # Log script stats meta = script.get("metadata", {}) beat_count = meta.get("beat_count", len(script.get("beats", []))) word_count = meta.get("word_count", "?") actual_duration = meta.get("actual_duration", "?") print(f" → {beat_count} beats, {word_count} words, ~{actual_duration}s") except Exception as e: print(f" [4-Director ERROR] {e}") opportunity["layer4_director_error"] = str(e) return opportunity # Evaluate script print(f" [4-Evaluator] Evaluating script...") try: evaluation = layer4_evaluator(script) evaluator_metadata = evaluation.get("execution_metadata", {}) verdict = evaluation.get("verdict", "UNKNOWN") composite_score = evaluation.get("composite_score", 0) print(f" → Verdict: {verdict} (score: {composite_score:.2f})") except Exception as e: print(f" [4-Evaluator ERROR] {e}") opportunity["layer4_evaluator_error"] = str(e) # Still save the script even if evaluation failed opportunity["script"] = script opportunity["director_metadata"] = director_metadata return opportunity # Store iteration data all_iterations.append({ "iteration": iteration, "script": script, "evaluation": evaluation, "director_metadata": director_metadata, "evaluator_metadata": evaluator_metadata }) # Check if production ready if verdict == "PRODUCTION_READY": print(f" → PRODUCTION READY - no revision needed") break # Check if we should revise if iteration < max_iterations: revision_instructions = evaluation.get("revision_instructions") if revision_instructions: critical_failures = evaluation.get("critical_failures", []) failure_count = len(critical_failures) print(f" → {verdict} - {failure_count} critical failures, revising...") else: print(f" → {verdict} - no revision instructions provided, stopping") break else: print(f" → {verdict} - max iterations reached") # Store final results in opportunity final_iteration = all_iterations[-1] if all_iterations else {} # Run Visual Decider on final script to get structured visual specs final_script = final_iteration.get("script") if final_script: print(f" [4-VisualDecider] Enhancing visuals with intelligent selection...") try: final_script = layer4_visual_decider(final_script, opportunity) # Update the iteration with enhanced script if all_iterations: all_iterations[-1]["script"] = final_script except Exception as e: print(f" [4-VisualDecider ERROR] {e} - continuing with original visuals") opportunity["script"] = final_script opportunity["evaluation"] = final_iteration.get("evaluation") opportunity["director_metadata"] = final_iteration.get("director_metadata") opportunity["evaluator_metadata"] = final_iteration.get("evaluator_metadata") opportunity["iteration_count"] = iteration opportunity["all_iterations"] = all_iterations opportunity["production_ready"] = evaluation.get("verdict") == "PRODUCTION_READY" if evaluation else False # Auto-save to channel output folder if channel_id: save_video_output(opportunity, channel_id) return opportunity # ============================================================================= # LEGACY V1/V2/V3 ARCHITECTURE (kept for backward compatibility) # ============================================================================= def layer4a_generate_concepts( opportunity: Dict[str, Any], channel_id: str = None, prompt_version: str = "v1" ) -> Dict[str, Any]: """ Layer 4a: Generate 3-4 ranked video concepts from an opportunity. LEGACY - use v4 architecture for new implementations. """ # Load channel context channel_name = "General" channel_description = "" competitive_gap = "" if channel_id and get_channel: channel = get_channel(channel_id) if channel: channel_name = channel.name channel_description = channel.description competitive_gap = channel.to_dict().get("competitive_gap", "") # Get prompt template prompt_template = get_prompt("layer4a_concept", version=prompt_version) metadata = get_prompt_metadata("layer4a_concept", version=prompt_version) # Fill in the template prompt = prompt_template.format( channel_name=channel_name, channel_description=channel_description, competitive_gap=competitive_gap, suggested_title=opportunity.get("suggested_title", "Unknown"), the_question=opportunity.get("the_question", ""), missing_angle=opportunity.get("missing_angle", ""), click_trigger=opportunity.get("target_audience", {}).get("click_trigger", ""), weighted_score=opportunity.get("weighted_score", 0), research_report_content=opportunity.get("research_report_content", "No research available") ) # Call Claude (Sonnet for 4a) response, duration = call_claude(prompt, model="sonnet") # Extract concepts and ensure they're sorted by virality_score concepts = response.get("video_concepts", []) concepts.sort(key=lambda x: x.get("virality_score", 0), reverse=True) # Assign ranks based on sort order for i, concept in enumerate(concepts): concept["rank"] = i + 1 return { "video_concepts": concepts, "execution_metadata": { "layer": "4a", "model": metadata.get("model", "sonnet"), "prompt_version": prompt_version, "generated_at": datetime.now(timezone.utc).isoformat(), "duration_seconds": round(duration, 2) } } def layer4b_generate_script( opportunity: Dict[str, Any], concept: Dict[str, Any], channel_id: str = None, prompt_version: str = "v1" ) -> Dict[str, Any]: """ Layer 4b: Generate a production-ready script from the #1 concept. LEGACY - use v4 architecture for new implementations. """ # Load channel context channel_name = "General" channel_description = "" competitive_gap = "" if channel_id and get_channel: channel = get_channel(channel_id) if channel: channel_name = channel.name channel_description = channel.description competitive_gap = channel.to_dict().get("competitive_gap", "") # Get prompt template prompt_template = get_prompt("layer4b_script", version=prompt_version) metadata = get_prompt_metadata("layer4b_script", version=prompt_version) # Format data_points as string data_points = concept.get("data_points", []) if isinstance(data_points, list): data_points_str = "\n".join(f"- {dp}" for dp in data_points) else: data_points_str = str(data_points) # Fill in the template prompt = prompt_template.format( channel_name=channel_name, channel_description=channel_description, competitive_gap=competitive_gap, concept_title=concept.get("title", ""), concept_hook=concept.get("hook", ""), concept_core_revelation=concept.get("core_revelation", ""), concept_emotional_arc=concept.get("emotional_arc", ""), concept_data_points=data_points_str, concept_share_trigger=concept.get("share_trigger", ""), research_report_content=opportunity.get("research_report_content", "No research available") ) # Call Claude (Opus for 4b) response, duration = call_claude(prompt, model="opus") script = response.get("video_script", {}) return { "video_script": script, "execution_metadata": { "layer": "4b", "model": metadata.get("model", "opus"), "prompt_version": prompt_version, "generated_at": datetime.now(timezone.utc).isoformat(), "duration_seconds": round(duration, 2) } } def layer4c_visual_direction( opportunity: Dict[str, Any], script: Dict[str, Any], prompt_version: str = "v1" ) -> Dict[str, Any]: """ Layer 4c: Generate shot-by-shot visual direction from script. LEGACY - use v4 architecture for new implementations. """ # Get prompt template prompt_template = get_prompt("layer4c_visual", version=prompt_version) metadata = get_prompt_metadata("layer4c_visual", version=prompt_version) # Fill in the template prompt = prompt_template.format( full_script=script.get("full_script", ""), research_report_content=opportunity.get("research_report_content", "No research available") ) # Call Claude (Opus for 4c) response, duration = call_claude(prompt, model="opus") visual = response.get("visual_direction", {}) return { "visual_direction": visual, "execution_metadata": { "layer": "4c", "model": metadata.get("model", "opus"), "prompt_version": prompt_version, "generated_at": datetime.now(timezone.utc).isoformat(), "duration_seconds": round(duration, 2) } } def layer4b_production_spec( opportunity: Dict[str, Any], concept: Dict[str, Any], channel_id: str = None, prompt_version: str = "v3" ) -> Dict[str, Any]: """ Layer 4b (v3): Generate unified production spec - script + visual direction integrated. LEGACY - use v4 architecture for new implementations. """ # Load channel context channel_name = "General" channel_description = "" competitive_gap = "" if channel_id and get_channel: channel = get_channel(channel_id) if channel: channel_name = channel.name channel_description = channel.description competitive_gap = channel.to_dict().get("competitive_gap", "") # Get prompt template prompt_template = get_prompt("layer4b_production_spec", version=prompt_version) metadata = get_prompt_metadata("layer4b_production_spec", version=prompt_version) # Format data_points as string data_points = concept.get("data_points", []) if isinstance(data_points, list): data_points_str = "\n".join(f"- {dp}" for dp in data_points) else: data_points_str = str(data_points) # Fill in the template prompt = prompt_template.format( channel_name=channel_name, channel_description=channel_description, competitive_gap=competitive_gap, concept_title=concept.get("title", ""), concept_hook=concept.get("hook", ""), concept_core_revelation=concept.get("core_revelation", ""), concept_emotional_arc=concept.get("emotional_arc", ""), concept_data_points=data_points_str, concept_share_trigger=concept.get("share_trigger", ""), research_report_content=opportunity.get("research_report_content", "No research available") ) # Call Claude (Opus for production spec) response, duration = call_claude(prompt, model="opus", timeout=600) production_spec = response.get("production_spec", {}) return { "production_spec": production_spec, "execution_metadata": { "layer": "4b", "model": metadata.get("model", "opus"), "prompt_version": prompt_version, "generated_at": datetime.now(timezone.utc).isoformat(), "duration_seconds": round(duration, 2) } } def layer4c_asset_specs( opportunity: Dict[str, Any], production_spec: Dict[str, Any], prompt_version: str = "v3" ) -> Dict[str, Any]: """ Layer 4c (v3): Generate detailed asset specs for graphics, infographics, images. LEGACY - use v4 architecture for new implementations. """ # Get prompt template prompt_template = get_prompt("layer4c_asset_specs", version=prompt_version) metadata = get_prompt_metadata("layer4c_asset_specs", version=prompt_version) # Extract color palette from production spec color_palette = production_spec.get("metadata", {}).get("color_palette", {}) color_palette_str = json.dumps(color_palette, indent=2) # Fill in the template prompt = prompt_template.format( production_spec=json.dumps(production_spec, indent=2), research_report_content=opportunity.get("research_report_content", "No research available"), color_palette=color_palette_str ) # Call Claude (Opus for asset specs) response, duration = call_claude(prompt, model="opus", timeout=600) asset_specs = response.get("asset_specs", {}) return { "asset_specs": asset_specs, "execution_metadata": { "layer": "4c", "model": metadata.get("model", "opus"), "prompt_version": prompt_version, "generated_at": datetime.now(timezone.utc).isoformat(), "duration_seconds": round(duration, 2) } } def run_video_generation_chain_legacy( opportunity: Dict[str, Any], channel_id: str = None, prompt_version: str = "v1" ) -> Dict[str, Any]: """ Run the legacy 4a → 4b → 4c chain for a single opportunity. LEGACY - use v4 architecture for new implementations. """ # Clear any previous video generation fields fields_to_clear = [ "video_concepts", "layer4a_metadata", "layer4a_error", "video_script", "production_spec", "layer4b_metadata", "layer4b_error", "visual_direction", "asset_specs", "layer4c_metadata", "layer4c_error" ] for field in fields_to_clear: opportunity.pop(field, None) # Layer 4a: Generate concepts print(" [4a] Generating concepts (Sonnet)...") try: result_4a = layer4a_generate_concepts(opportunity, channel_id, prompt_version) opportunity["video_concepts"] = result_4a["video_concepts"] opportunity["layer4a_metadata"] = result_4a["execution_metadata"] num_concepts = len(result_4a["video_concepts"]) top_score = result_4a["video_concepts"][0].get("virality_score", "?") if num_concepts > 0 else "?" print(f" → {num_concepts} concepts generated, top virality score: {top_score}") except Exception as e: print(f" [4a ERROR] {e}") opportunity["layer4a_error"] = str(e) return opportunity # Get #1 concept concepts = opportunity.get("video_concepts", []) if not concepts: opportunity["layer4b_error"] = "No concepts generated by 4a" return opportunity top_concept = concepts[0] # Check if using v3+ (unified production spec) or v1/v2 (separate script + visual) use_v3_architecture = prompt_version.startswith("v3") if use_v3_architecture: # V3 ARCHITECTURE: 4b = Production Spec print(" [4b] Generating production spec (Opus)...") try: result_4b = layer4b_production_spec(opportunity, top_concept, channel_id, prompt_version) opportunity["production_spec"] = result_4b["production_spec"] opportunity["layer4b_metadata"] = result_4b["execution_metadata"] spec = result_4b["production_spec"] meta = spec.get("metadata", {}) shot_count = meta.get("shot_count", len(spec.get("shots", []))) runtime = meta.get("total_runtime_seconds", "?") print(f" → {shot_count} shots, ~{runtime}s runtime") except Exception as e: print(f" [4b ERROR] {e}") opportunity["layer4b_error"] = str(e) return opportunity production_spec = opportunity.get("production_spec", {}) # V3: 4c = Asset Generation Specs print(" [4c] Generating asset specs (Opus)...") try: result_4c = layer4c_asset_specs(opportunity, production_spec, prompt_version) opportunity["asset_specs"] = result_4c["asset_specs"] opportunity["layer4c_metadata"] = result_4c["execution_metadata"] assets = result_4c["asset_specs"] total_assets = assets.get("total_assets_needed", "?") print(f" → {total_assets} assets specified") except Exception as e: print(f" [4c ERROR] {e}") opportunity["layer4c_error"] = str(e) else: # V1/V2 ARCHITECTURE: Separate script and visual direction print(" [4b] Generating script (Opus)...") try: result_4b = layer4b_generate_script(opportunity, top_concept, channel_id, prompt_version) opportunity["video_script"] = result_4b["video_script"] opportunity["layer4b_metadata"] = result_4b["execution_metadata"] word_count = result_4b["video_script"].get("word_count", "?") duration = result_4b["video_script"].get("estimated_duration_seconds", "?") print(f" → {word_count} words, ~{duration}s duration") except Exception as e: print(f" [4b ERROR] {e}") opportunity["layer4b_error"] = str(e) return opportunity script = opportunity.get("video_script", {}) # V1/V2: 4c = Visual Direction print(" [4c] Generating visual direction (Opus)...") try: result_4c = layer4c_visual_direction(opportunity, script, prompt_version) opportunity["visual_direction"] = result_4c["visual_direction"] opportunity["layer4c_metadata"] = result_4c["execution_metadata"] num_shots = len(result_4c["visual_direction"].get("shots", [])) summary = result_4c["visual_direction"].get("summary", {}) infographics = summary.get("infographics_needed", "?") print(f" → {num_shots} shots, {infographics} infographics needed") except Exception as e: print(f" [4c ERROR] {e}") opportunity["layer4c_error"] = str(e) return opportunity # ============================================================================= # MAIN ORCHESTRATION # ============================================================================= def run_video_generation_chain( opportunity: Dict[str, Any], channel_id: str = None, prompt_version: str = "v4", target_duration: int = 90 ) -> Dict[str, Any]: """ Run the video generation chain for a single opportunity. Automatically selects the appropriate architecture based on prompt_version: - v4: Director → Evaluator → (Revise if needed) - v1/v2/v3: Legacy 4a → 4b → 4c chain Args: opportunity: Must have research_completed=True (or use skip_research) channel_id: Channel for metadata lookup prompt_version: Prompt version to use ("v4" for new architecture) target_duration: Target video duration in seconds (for v4) Returns: Updated opportunity dict with video generation fields added """ if prompt_version.startswith("v4"): # V4 Architecture: Director → Evaluator → (Revise) return run_layer4_v4_pipeline( opportunity=opportunity, channel_id=channel_id, target_duration=target_duration, max_iterations=2 # Initial + 1 revision ) else: # Legacy Architecture: 4a → 4b → 4c return run_video_generation_chain_legacy( opportunity=opportunity, channel_id=channel_id, prompt_version=prompt_version ) def run_video_generation_for_opportunities( opportunities_data: Dict[str, Any], channel_id: str = None, specific_id: int = None, prompt_version: str = "v4", skip_research: bool = False, target_duration: int = 90 ) -> Dict[str, Any]: """ Run video generation chain for opportunities. Args: opportunities_data: Full opportunities JSON data channel_id: Channel ID for context specific_id: If set, only process this opportunity index prompt_version: Prompt version to use ("v4" recommended) skip_research: If True, process even without research target_duration: Target video duration in seconds Returns: Updated opportunities_data with video fields added """ print("\n" + "=" * 60) print(f"LAYER 4: VIDEO GENERATION (v{prompt_version})") print("=" * 60) if prompt_version.startswith("v4"): print("Architecture: Director → Evaluator → (Revise if needed)") else: print("Architecture: Legacy 4a → 4b → 4c") opportunities = opportunities_data.get("opportunities", []) # Filter to eligible opportunities if specific_id is not None: # Single opportunity mode if specific_id < 0 or specific_id >= len(opportunities): print(f"[ERROR] Invalid opportunity ID: {specific_id} (valid: 0-{len(opportunities)-1})") return opportunities_data to_process = [(specific_id, opportunities[specific_id])] else: # All researched opportunities (or all if skip_research) to_process = [ (i, opp) for i, opp in enumerate(opportunities) if opp.get("research_completed", False) or skip_research ] if not to_process: print("[WARN] No opportunities eligible for video generation") print(" (Need research_completed=true or use --skip-research)") return opportunities_data print(f"[4.0] Processing {len(to_process)} opportunity(ies)") print(f" Prompt version: {prompt_version}") print(f" Target duration: {target_duration}s") if channel_id: print(f" Channel: {channel_id}") # Get analysis_run_id for DB updates analysis_run_id = opportunities_data.get("analysis_run_id") for idx, opp in to_process: title = opp.get("suggested_title", opp.get("premise", "Unknown"))[:50] print(f"\n[{idx}] {title}...") # Run video generation chain updated_opp = run_video_generation_chain( opportunity=opp, channel_id=channel_id, prompt_version=prompt_version, target_duration=target_duration ) opportunities[idx] = updated_opp # Update database if analysis_run_id and updated_opp.get("cluster_id") is not None: # Determine success status based on architecture if prompt_version.startswith("v4"): has_output = "script" in updated_opp and updated_opp["script"] is not None is_production_ready = updated_opp.get("production_ready", False) video_status = "completed" if has_output else "failed" else: has_output = "production_spec" in updated_opp or "video_script" in updated_opp has_error = "layer4b_error" in updated_opp video_status = "completed" if (has_output and not has_error) else "failed" try: # For v4, store script and evaluation if prompt_version.startswith("v4"): update_opportunity_video_gen_by_cluster( analysis_run_id=analysis_run_id, cluster_id=updated_opp["cluster_id"], video_gen_status=video_status, video_concepts=None, # v4 doesn't use concepts production_spec=updated_opp.get("script"), # Store script as production_spec asset_specs=updated_opp.get("evaluation") # Store evaluation as asset_specs ) else: update_opportunity_video_gen_by_cluster( analysis_run_id=analysis_run_id, cluster_id=updated_opp["cluster_id"], video_gen_status=video_status, video_concepts=updated_opp.get("video_concepts"), production_spec=updated_opp.get("production_spec"), asset_specs=updated_opp.get("asset_specs") ) except Exception as e: print(f" [WARN] DB update for video gen failed: {e}") # Update the data structure opportunities_data["opportunities"] = opportunities opportunities_data["video_generation_timestamp"] = datetime.now(timezone.utc).isoformat() opportunities_data["video_generation_version"] = prompt_version # Summary print("\n" + "-" * 60) print("VIDEO GENERATION SUMMARY") print("-" * 60) # Count success based on architecture if prompt_version.startswith("v4"): success_count = sum( 1 for i, _ in to_process if opportunities[i].get("script") is not None ) production_ready_count = sum( 1 for i, _ in to_process if opportunities[i].get("production_ready", False) ) print(f" Processed: {len(to_process)}") print(f" Scripts generated: {success_count}") print(f" Production ready: {production_ready_count}") print(f" Need revision: {success_count - production_ready_count}") else: def is_successful(opp): has_output = "production_spec" in opp or "video_script" in opp has_error = "layer4b_error" in opp return has_output and not has_error success_count = sum(1 for i, _ in to_process if is_successful(opportunities[i])) print(f" Processed: {len(to_process)}") print(f" Successful: {success_count}") print(f" Failed: {len(to_process) - success_count}") return opportunities_data if __name__ == "__main__": # Quick test: run on first opportunity in existing file import sys output_path = Path(__file__).parent / "output" / "analysis_opportunities.json" if not output_path.exists(): print(f"No opportunities file found at {output_path}") sys.exit(1) with open(output_path, "r") as f: data = json.load(f) # Run on first opportunity with v4 architecture result = run_video_generation_for_opportunities( data, specific_id=0, prompt_version="v4", target_duration=90 ) # Save with open(output_path, "w") as f: json.dump(result, f, indent=2) print(f"\nSaved to {output_path}")