#!/usr/bin/env python3 """ Layer 4: Video Structure Generation Takes the full Layer 3 analysis output and generates detailed, virality-optimized video structures for each opportunity. This is a standalone enrichment step. Usage: python3 generate_video_structures.py """ import json import subprocess import sys from pathlib import Path from datetime import datetime OUTPUT_DIR = Path(__file__).parent / "output" ANALYSIS_PATH = OUTPUT_DIR / "analysis_opportunities.json" VIRALITY_PROMPT = """You are a YouTube virality strategist who has studied thousands of successful explainer videos. You understand retention curves, pattern interrupts, payoff timing, and what makes viewers share content. You are given the full analysis of a video opportunity. Your ONLY job is to create the optimal video structure - the exact blueprint for how this video should flow from first frame to last. ## THE OPPORTUNITY **Title:** {title} **Theme:** {theme} **The Winning Angle:** {missing_angle} **Target Audience - The Moment:** {the_moment} **What They Know:** {what_they_know} **What They Think Is Wrong:** {what_is_wrong} **Click Trigger:** {click_trigger} **The Core Confusion:** Primary Question: {primary_question} Secondary Questions: {secondary_questions} **Existing Content & Why It Fails:** {existing_content} **Opening Hook (from initial analysis):** {opening_hook} **Critical Success Factor:** {critical_success_factor} **Verdict:** {verdict} **Reasoning:** {verdict_reasoning} --- ## YOUR TASK Create a detailed video structure optimized for: 1. **Retention** - Viewers must stay through the whole video 2. **Satisfaction** - The payoff must match or exceed the promise 3. **Shareability** - Include moments that make people want to send this to someone ## STRUCTURE REQUIREMENTS Create exactly 5-7 sections. For EACH section, provide: 1. **Section Title** - What this part is called (for your reference, not shown to viewer) 2. **Duration** - Exact timing (e.g., "0:00-0:45" or "45 seconds") 3. **Purpose** - What this section accomplishes for retention/engagement 4. **Content** - Exactly what happens in this section, what you say, what you show 5. **The Hook to Next** - How this section creates anticipation for what's coming ## STRUCTURAL PRINCIPLES TO FOLLOW - **First 30 seconds**: Must reframe the viewer's understanding. Don't just state the topic - create a gap between what they thought and what's true. - **2-3 minute mark**: This is where most viewers drop off. Place a "holy shit" moment here - a counterintuitive fact, a surprising connection, or a demonstration. - **Pattern interrupts**: Every 60-90 seconds, change something - pacing, visuals, tone, or introduce a new thread. - **The payoff must land**: The main insight should feel earned, not given away too early. - **End with implications**: Don't just conclude - show why this matters beyond the immediate topic. ## RESPOND IN THIS EXACT JSON FORMAT {{ "video_structure": {{ "total_duration": "X:XX - X:XX minutes", "retention_strategy": "One sentence on the core retention approach", "sections": [ {{ "section_number": 1, "title": "Section Name", "timestamp": "0:00-0:45", "duration_seconds": 45, "purpose": "What this accomplishes", "content": "Detailed description of what happens", "visuals": "What the viewer sees", "hook_to_next": "How this creates anticipation" }} ], "key_retention_moments": [ {{ "timestamp": "2:15", "description": "The moment and why it keeps viewers watching" }} ], "share_triggers": [ "Specific moment or line that makes viewers want to share" ] }} }} Generate the structure now. Be specific. This is a production blueprint, not a vague outline.""" def extract_text(val, default=""): """Safely extract text from various formats.""" if val is None: return default if isinstance(val, str): return val if isinstance(val, dict): # Try common keys for key in ["text", "value", "content", "description", "primary", "main"]: if key in val: return extract_text(val[key], default) # Format dict as text parts = [] for k, v in val.items(): parts.append(f"{k.replace('_', ' ').title()}: {extract_text(v)}") return " | ".join(parts) if isinstance(val, list): return "; ".join(extract_text(item) for item in val[:5]) return str(val) def generate_structure_for_opportunity(opp: dict, index: int, total: int) -> dict: """Generate detailed video structure for one opportunity.""" title = opp.get("suggested_title", opp.get("topic", "Unknown")) print(f"\n[{index+1}/{total}] {title[:60]}...", flush=True) # Extract fields for prompt theme = extract_text(opp.get("theme", "")) missing_angle = extract_text(opp.get("missing_angle", opp.get("angle", ""))) # Audience audience = opp.get("target_audience", {}) if isinstance(audience, str): audience = {"the_moment": audience} the_moment = extract_text(audience.get("the_moment", audience.get("moment", ""))) what_they_know = extract_text(audience.get("what_they_know", "")) what_is_wrong = extract_text(audience.get("what_they_think_is_wrong", audience.get("what_is_wrong", ""))) click_trigger = extract_text(audience.get("click_trigger", "")) # Confusion confusion = opp.get("confusion_analysis", {}) if isinstance(confusion, str): confusion = {"primary_question": confusion} primary_question = extract_text(confusion.get("primary_question", "")) secondary_qs = confusion.get("secondary_questions", []) if isinstance(secondary_qs, list): secondary_questions = "; ".join(extract_text(q) for q in secondary_qs[:3]) else: secondary_questions = extract_text(secondary_qs) # Other fields existing_content = extract_text(opp.get("existing_content", "")) opening_hook = extract_text(opp.get("opening_hook", "")) critical_success_factor = extract_text(opp.get("critical_success_factor", "")) verdict = opp.get("verdict", "") verdict_reasoning = extract_text(opp.get("verdict_reasoning", "")) # Build prompt prompt = VIRALITY_PROMPT.format( title=title, theme=theme, missing_angle=missing_angle, the_moment=the_moment, what_they_know=what_they_know, what_is_wrong=what_is_wrong, click_trigger=click_trigger, primary_question=primary_question, secondary_questions=secondary_questions, existing_content=existing_content[:1000], # Truncate if too long opening_hook=opening_hook, critical_success_factor=critical_success_factor, verdict=verdict, verdict_reasoning=verdict_reasoning[:500] ) try: # Call Claude using stdin with explicit flush import tempfile import os # Write prompt to temp file prompt_file = f"/tmp/layer4_prompt_{index}.txt" with open(prompt_file, 'w') as f: f.write(prompt) # Call Claude reading from file result = subprocess.run( f"cat '{prompt_file}' | claude --print --model sonnet -p -", shell=True, capture_output=True, text=True, timeout=120 ) # Clean up os.unlink(prompt_file) if result.returncode != 0: print(f" ERROR: Claude returned non-zero exit code", flush=True) return None response = result.stdout.strip() # Extract JSON from response json_start = response.find("{") json_end = response.rfind("}") + 1 if json_start == -1 or json_end == 0: print(f" ERROR: No JSON found in response", flush=True) return None json_str = response[json_start:json_end] structure_data = json.loads(json_str) # Extract the video_structure if "video_structure" in structure_data: structure = structure_data["video_structure"] else: structure = structure_data section_count = len(structure.get("sections", [])) duration = structure.get("total_duration", "unknown") print(f" ✓ Generated {section_count} sections ({duration})", flush=True) return structure except subprocess.TimeoutExpired: print(f" ERROR: Timeout", flush=True) return None except json.JSONDecodeError as e: print(f" ERROR: JSON parse failed: {e}", flush=True) return None except Exception as e: print(f" ERROR: {e}", flush=True) return None def save_data(data, success_count): """Save data incrementally.""" data["layer4_timestamp"] = datetime.now().isoformat() data["layer4_success_count"] = success_count with open(ANALYSIS_PATH, "w") as f: json.dump(data, f, indent=2) def main(): print("=" * 60, flush=True) print("LAYER 4: VIDEO STRUCTURE GENERATION", flush=True) print("=" * 60, flush=True) # Load existing analysis if not ANALYSIS_PATH.exists(): print(f"ERROR: {ANALYSIS_PATH} not found", flush=True) print("Run the analysis pipeline first: python3 analyze.py", flush=True) sys.exit(1) with open(ANALYSIS_PATH, "r") as f: data = json.load(f) opportunities = data.get("opportunities", []) print(f"\nLoaded {len(opportunities)} opportunities", flush=True) # Count already processed already_done = sum(1 for o in opportunities if "video_structure" in o and not o.get("video_structure", {}).get("legacy")) if already_done > 0: print(f"Resuming: {already_done} already have video_structure", flush=True) # Process each opportunity success_count = already_done for i, opp in enumerate(opportunities): # Skip if already has non-legacy video_structure if "video_structure" in opp and not opp.get("video_structure", {}).get("legacy"): print(f"\n[{i+1}/{len(opportunities)}] SKIP (already done)", flush=True) continue structure = generate_structure_for_opportunity(opp, i, len(opportunities)) if structure: opp["video_structure"] = structure success_count += 1 # Save after each success save_data(data, success_count) print(f" (saved)", flush=True) else: # Keep old structure if new one fails if "structure" in opp and "video_structure" not in opp: opp["video_structure"] = {"sections": opp["structure"], "legacy": True} # Final save save_data(data, success_count) print("\n" + "=" * 60, flush=True) print(f"COMPLETE: {success_count}/{len(opportunities)} structures generated", flush=True) print(f"Updated: {ANALYSIS_PATH}", flush=True) print("=" * 60, flush=True) print("\nNext: Run 'python3 generate_dashboard_data.py' to update the dashboard", flush=True) if __name__ == "__main__": main()