Upload 13 files

__init__.py

@@ -0,0 +1 @@
+# StoryLens - Video Ad Narrative Structure Analyzer

app.py

@@ -0,0 +1,333 @@
+import streamlit as st
+import os
+from PIL import Image
+
+from config import INDUSTRIES, CAMPAIGN_GOALS, CATEGORY_COLORS, MAX_VIDEO_LENGTH_SECONDS
+from video_loader import VideoLoader
+from frame_extractor import FrameExtractor
+from audio_extractor import AudioExtractor
+from vision_analyzer import VisionAnalyzer
+from segment_synchronizer import SegmentSynchronizer
+from narrative_classifier import NarrativeClassifier
+from report_generator import ReportGenerator
+
+# Page config
+st.set_page_config(
+    page_title="StoryLens - Ad Narrative Analyzer",
+    page_icon="🎬",
+    layout="wide"
+)
+
+# Initialize session state
+if 'analysis_result' not in st.session_state:
+    st.session_state.analysis_result = None
+if 'transcript' not in st.session_state:
+    st.session_state.transcript = None
+
+# Sidebar
+with st.sidebar:
+    st.header("Configuration")
+
+    # API Settings
+    with st.expander("API Settings", expanded=True):
+        st.subheader("MiniMax (Vision & LLM)")
+        api_key = st.text_input(
+            "MiniMax API Key",
+            type="password",
+            value=os.getenv("MINIMAX_API_KEY", ""),
+            help="Get your API key from MiniMax platform"
+        )
+        group_id = st.text_input(
+            "MiniMax Group ID",
+            value=os.getenv("MINIMAX_GROUP_ID", "")
+        )
+
+        if api_key and group_id:
+            st.session_state.api_key = api_key
+            st.session_state.group_id = group_id
+            st.success("MiniMax configured")
+
+        st.divider()
+
+        st.subheader("OpenAI (Whisper)")
+        openai_key = st.text_input(
+            "OpenAI API Key",
+            type="password",
+            value=os.getenv("OPENAI_API_KEY", ""),
+            help="For audio transcription (Whisper)"
+        )
+
+        if openai_key:
+            st.session_state.openai_key = openai_key
+            st.success("OpenAI configured")
+
+    st.divider()
+
+    # Campaign Settings
+    st.subheader("Campaign Settings")
+
+    industry = st.selectbox("Industry", INDUSTRIES)
+    campaign_goal = st.selectbox("Campaign Goal", CAMPAIGN_GOALS)
+
+# Main content
+st.title("StoryLens")
+st.markdown("*Diagnose your video ad's narrative structure in 60 seconds*")
+
+# Video Input
+st.header("Video Input")
+
+col1, col2 = st.columns(2)
+
+with col1:
+    st.subheader("Upload File")
+    uploaded_file = st.file_uploader(
+        "Choose video file",
+        type=["mp4", "mov", "avi", "webm"],
+        help="Max 120 seconds"
+    )
+
+with col2:
+    st.subheader("YouTube URL")
+    youtube_url = st.text_input(
+        "Paste URL",
+        placeholder="https://youtube.com/watch?v=..."
+    )
+
+# Analyze button
+video_source = uploaded_file or youtube_url
+minimax_ready = hasattr(st.session_state, 'api_key') and st.session_state.api_key
+openai_ready = hasattr(st.session_state, 'openai_key') and st.session_state.openai_key
+api_ready = minimax_ready and openai_ready
+
+if video_source and api_ready:
+    if st.button("Analyze", type="primary", use_container_width=True):
+
+        # Progress container
+        progress_container = st.container()
+
+        with progress_container:
+            progress_bar = st.progress(0)
+            status_text = st.empty()
+
+            try:
+                # Initialize components
+                api_key = st.session_state.api_key
+                group_id = st.session_state.group_id
+                openai_key = st.session_state.openai_key
+
+                video_loader = VideoLoader()
+                frame_extractor = FrameExtractor()
+                audio_extractor = AudioExtractor(openai_api_key=openai_key)
+                vision_analyzer = VisionAnalyzer(api_key, group_id)
+                synchronizer = SegmentSynchronizer()
+                classifier = NarrativeClassifier(api_key, group_id)
+                report_generator = ReportGenerator()
+
+                # Step 1: Load video
+                status_text.text("Loading video...")
+                progress_bar.progress(10)
+
+                if uploaded_file:
+                    video_path = video_loader.load_local(uploaded_file)
+                else:
+                    video_path = video_loader.load_youtube(youtube_url)
+
+                if not video_path:
+                    st.error("Failed to load video")
+                    st.stop()
+
+                # Check duration
+                duration = video_loader.get_video_duration(video_path)
+                if duration > MAX_VIDEO_LENGTH_SECONDS:
+                    st.error(f"Video too long ({duration:.0f}s). Max allowed: {MAX_VIDEO_LENGTH_SECONDS}s")
+                    st.stop()
+
+                # Step 2: Extract frames
+                status_text.text("Extracting frames...")
+                progress_bar.progress(20)
+
+                frames = frame_extractor.extract_frames(video_path)
+
+                # Step 3: Extract & transcribe audio
+                status_text.text("Transcribing audio...")
+                progress_bar.progress(35)
+
+                audio_path = audio_extractor.extract_audio(video_path)
+                transcript = audio_extractor.transcribe(audio_path)
+
+                # Step 4: Analyze frames visually
+                status_text.text("Analyzing frames...")
+                progress_bar.progress(50)
+
+                frame_descriptions = vision_analyzer.describe_frames_batch(frames)
+
+                # Step 5: Synchronize
+                status_text.text("Synchronizing segments...")
+                progress_bar.progress(70)
+
+                segments = synchronizer.synchronize(frame_descriptions, transcript)
+
+                # Step 6: Classify narrative
+                status_text.text("Classifying narrative structure...")
+                progress_bar.progress(85)
+
+                analysis = classifier.classify(segments)
+
+                # Step 7: Generate report
+                status_text.text("Generating report...")
+                progress_bar.progress(95)
+
+                report = report_generator.generate(analysis, industry, campaign_goal)
+
+                progress_bar.progress(100)
+                status_text.text("Analysis complete!")
+
+                # Store result
+                st.session_state.analysis_result = report
+                st.session_state.transcript = transcript
+
+            except Exception as e:
+                st.error(f"Analysis failed: {str(e)}")
+                import traceback
+                st.code(traceback.format_exc())
+
+elif not api_ready:
+    missing = []
+    if not minimax_ready:
+        missing.append("MiniMax API Key + Group ID")
+    if not openai_ready:
+        missing.append("OpenAI API Key")
+    st.warning(f"Please configure API settings in the sidebar: {', '.join(missing)}")
+elif not video_source:
+    st.info("Upload a video file or paste a YouTube URL to begin")
+
+# Display results
+if st.session_state.analysis_result:
+    result = st.session_state.analysis_result
+
+    st.divider()
+
+    # Summary metrics
+    st.header("Analysis Results")
+
+    col1, col2, col3, col4 = st.columns(4)
+
+    with col1:
+        story_status = "YES" if result['summary']['has_story'] else "NO"
+        st.metric("Story Detected", story_status)
+
+    with col2:
+        st.metric("Detected Arc", result['summary']['detected_arc'])
+
+    with col3:
+        st.metric("Optimal Arc", result['summary']['optimal_arc_for_goal'])
+
+    with col4:
+        st.metric("Potential Uplift", result['summary']['potential_uplift'])
+
+    # Story explanation
+    if result['summary']['story_explanation']:
+        st.info(f"**Story Analysis:** {result['summary']['story_explanation']}")
+
+    st.divider()
+
+    # Timeline visualization
+    st.subheader("Narrative Timeline")
+
+    for seg in result['segments']:
+        col1, col2, col3, col4 = st.columns([1, 1, 2, 3])
+
+        with col1:
+            # Frame thumbnail
+            if seg.get('frame_path') and os.path.exists(seg['frame_path']):
+                img = Image.open(seg['frame_path'])
+                st.image(img, width=120)
+            else:
+                st.write("[Frame]")
+
+        with col2:
+            st.caption(f"**{seg['start']:.1f}s - {seg['end']:.1f}s**")
+
+            # Role badge with color
+            category = seg.get('role_category', 'OTHER')
+            color = CATEGORY_COLORS.get(category, '#9E9E9E')
+            role = seg.get('functional_role', 'Unknown')
+
+            st.markdown(
+                f'<span style="background-color: {color}; color: white; '
+                f'padding: 4px 8px; border-radius: 4px; font-size: 12px;">'
+                f'{role}</span>',
+                unsafe_allow_html=True
+            )
+
+        with col3:
+            visual_text = seg.get('visual', 'N/A')
+            st.write(f"**Visual:** {visual_text}")
+
+        with col4:
+            if seg.get('speech'):
+                st.write(f"**Speech:** \"{seg['speech']}\"")
+            if seg.get('reasoning'):
+                st.caption(f"*{seg['reasoning']}*")
+
+        st.divider()
+
+    # Detected sequence
+    if result.get('detected_sequence'):
+        st.subheader("Story Arc Flow")
+        arc_flow = " -> ".join(result['detected_sequence'])
+        st.markdown(f"**{arc_flow}**")
+
+    # Missing elements
+    if result.get('missing_elements'):
+        st.subheader("Missing Elements")
+        for element in result['missing_elements']:
+            st.warning(f"- {element}")
+
+    st.divider()
+
+    # Recommendations
+    st.subheader("Recommendations")
+
+    for rec in result.get('recommendations', []):
+        priority = rec.get('priority', 'LOW')
+        icon = "[HIGH]" if priority == "HIGH" else "[MEDIUM]" if priority == "MEDIUM" else "[LOW]"
+
+        with st.expander(f"{icon} {rec['action']}", expanded=(priority == "HIGH")):
+            col1, col2 = st.columns(2)
+            with col1:
+                st.metric("Expected Impact", rec.get('expected_impact', 'N/A'))
+            with col2:
+                st.metric("Priority", priority)
+            st.write(f"**Reasoning:** {rec.get('reasoning', '')}")
+
+    # Benchmark info
+    with st.expander("Benchmark Details"):
+        benchmark = result.get('benchmark', {})
+        st.write(f"**Best Arc for {campaign_goal}:** {benchmark.get('best_arc', 'N/A')}")
+        st.write(f"**Average Uplift:** +{benchmark.get('uplift_percent', '?')}%")
+        st.write(f"**Recommendation:** {benchmark.get('recommendation', 'N/A')}")
+
+    # Full Transcript
+    if hasattr(st.session_state, 'transcript') and st.session_state.transcript:
+        st.divider()
+        st.subheader("Full Transcript")
+
+        transcript = st.session_state.transcript
+
+        # Display with timestamps
+        for seg in transcript:
+            start = seg.get('start', 0)
+            end = seg.get('end', 0)
+            text = seg.get('text', '')
+
+            if text:
+                if start > 0 or end > 0:
+                    st.markdown(f"**[{start:.1f}s - {end:.1f}s]** {text}")
+                else:
+                    st.markdown(text)
+
+        # Also show as plain text block
+        with st.expander("Plain Text"):
+            full_text = " ".join([seg.get('text', '') for seg in transcript if seg.get('text')])
+            st.text_area("Full transcript", full_text, height=150, disabled=True)

audio_extractor.py

@@ -0,0 +1,78 @@
+import os
+from typing import List, Dict
+
+from moviepy.editor import VideoFileClip
+from openai import OpenAI
+
+
+class AudioExtractor:
+    def __init__(self, openai_api_key: str = None, **kwargs):
+        self.openai_api_key = openai_api_key
+        self.client = None
+        if openai_api_key:
+            self.client = OpenAI(api_key=openai_api_key)
+
+    def extract_audio(self, video_path: str, output_path: str = None) -> str:
+        """
+        Extract audio track from video.
+
+        Returns:
+            Path to extracted MP3 file (better for Whisper API)
+        """
+        if output_path is None:
+            output_path = video_path.rsplit('.', 1)[0] + '.mp3'
+
+        video = VideoFileClip(video_path)
+        video.audio.write_audiofile(output_path, codec='mp3', verbose=False, logger=None)
+        video.close()
+
+        return output_path
+
+    def transcribe(self, audio_path: str) -> List[Dict]:
+        """
+        Transcribe audio with timestamps using OpenAI Whisper API.
+
+        Returns:
+            List of segments: [
+                {"start": 0.0, "end": 3.2, "text": "Tired of everyday exhaustion?"},
+                {"start": 3.2, "end": 7.1, "text": "Meet the new SuperVit..."},
+                ...
+            ]
+        """
+        if not self.client:
+            print("OpenAI API key not configured")
+            return []
+
+        try:
+            with open(audio_path, "rb") as audio_file:
+                # Use whisper-1 model with verbose_json for timestamps
+                response = self.client.audio.transcriptions.create(
+                    model="whisper-1",
+                    file=audio_file,
+                    response_format="verbose_json",
+                    timestamp_granularities=["segment"]
+                )
+
+            segments = []
+
+            # Extract segments with timestamps
+            if hasattr(response, 'segments') and response.segments:
+                for segment in response.segments:
+                    segments.append({
+                        "start": segment.get('start', 0) if isinstance(segment, dict) else getattr(segment, 'start', 0),
+                        "end": segment.get('end', 0) if isinstance(segment, dict) else getattr(segment, 'end', 0),
+                        "text": (segment.get('text', '') if isinstance(segment, dict) else getattr(segment, 'text', '')).strip()
+                    })
+            elif hasattr(response, 'text') and response.text:
+                # Fallback if no segments
+                segments.append({
+                    "start": 0.0,
+                    "end": 0.0,
+                    "text": response.text.strip()
+                })
+
+            return segments
+
+        except Exception as e:
+            print(f"Transcription error: {e}")
+            return []

benchmarks.py

@@ -0,0 +1,151 @@
+from typing import Dict
+
+BENCHMARKS = {
+    "Apparel & Accessories": {
+        "retention": {
+            "best_arc": "Hook-Feature-Benefit-Action",
+            "best_arc_short": "HFBA",
+            "uplift_percent": 5.8,
+            "recommendation": "Start with strong hook, quickly show product features and benefits"
+        },
+        "ctr": {
+            "best_arc": "AIDA",
+            "best_arc_short": "AIDA",
+            "uplift_percent": 8.9,
+            "recommendation": "Build desire through aspirational content before call-to-action"
+        },
+        "cvr": {
+            "best_arc": "Social-Proof-Action",
+            "best_arc_short": "SPA",
+            "uplift_percent": 4.6,
+            "recommendation": "Lead with testimonials and reviews to build trust"
+        }
+    },
+    "Beauty": {
+        "retention": {
+            "best_arc": "Hook-Problem-Demo-Solution",
+            "best_arc_short": "HPDS",
+            "uplift_percent": 4.9,
+            "recommendation": "Hook attention, show problem, demonstrate product solving it"
+        },
+        "ctr": {
+            "best_arc": "Hook-Feature-Benefit-Action",
+            "best_arc_short": "HFBA",
+            "uplift_percent": 2.8,
+            "recommendation": "Focus on product features and benefits after initial hook"
+        },
+        "cvr": {
+            "best_arc": "Social-Proof-Action",
+            "best_arc_short": "SPA",
+            "uplift_percent": 3.7,
+            "recommendation": "Beauty buyers respond well to testimonials and reviews"
+        }
+    },
+    "Food": {
+        "retention": {
+            "best_arc": "Problem-Agitate-Solution",
+            "best_arc_short": "PAS",
+            "uplift_percent": 6.3,
+            "recommendation": "Amplify the problem/need before showing solution"
+        },
+        "ctr": {
+            "best_arc": "AIDA",
+            "best_arc_short": "AIDA",
+            "uplift_percent": 4.8,
+            "recommendation": "Build appetite and desire psychologically"
+        },
+        "cvr": {
+            "best_arc": "Problem-Agitate-Solution",
+            "best_arc_short": "PAS",
+            "uplift_percent": 8.5,
+            "recommendation": "Strong problem-solution narrative drives food conversions"
+        }
+    },
+    "Beverages": {
+        "retention": {
+            "best_arc": "Hook-Problem-Solution",
+            "best_arc_short": "HPS",
+            "uplift_percent": 4.1,
+            "recommendation": "Quick hook into problem-solution flow"
+        },
+        "ctr": {
+            "best_arc": "Feature-Benefit-Action",
+            "best_arc_short": "FBA",
+            "uplift_percent": 3.9,
+            "recommendation": "Direct product focus works for beverages"
+        },
+        "cvr": {
+            "best_arc": "Feature-Benefit-Action",
+            "best_arc_short": "FBA",
+            "uplift_percent": 5.1,
+            "recommendation": "Detailed feature explanation drives beverage conversions"
+        }
+    },
+    "Other": {
+        "retention": {
+            "best_arc": "Hook-Feature-Benefit-Action",
+            "best_arc_short": "HFBA",
+            "uplift_percent": 5.0,
+            "recommendation": "General best practice: hook + features + benefits"
+        },
+        "ctr": {
+            "best_arc": "AIDA",
+            "best_arc_short": "AIDA",
+            "uplift_percent": 5.0,
+            "recommendation": "Classic AIDA funnel works across categories"
+        },
+        "cvr": {
+            "best_arc": "Social-Proof-Action",
+            "best_arc_short": "SPA",
+            "uplift_percent": 4.0,
+            "recommendation": "Social proof generally effective for conversions"
+        }
+    }
+}
+
+MISSING_ELEMENT_IMPACT = {
+    "Hook": {
+        "impact": "+5-8% retention in first 2 seconds",
+        "suggestion": "Add attention-grabbing opening (question, surprising visual, bold statement)"
+    },
+    "Problem Setup": {
+        "impact": "+4-6% retention",
+        "suggestion": "Establish relatable pain point before showing product"
+    },
+    "Social Proof": {
+        "impact": "+3-5% CVR",
+        "suggestion": "Add testimonial, review, or crowd validation before CTA"
+    },
+    "Urgency Trigger": {
+        "impact": "+2-4% CVR",
+        "suggestion": "Add time-limited element (limited time offer, countdown)"
+    },
+    "Call-to-Action": {
+        "impact": "Critical for conversions",
+        "suggestion": "Add clear CTA (Shop Now, Learn More, Get Started)"
+    },
+    "Outcome": {
+        "impact": "+3-5% retention and CVR",
+        "suggestion": "Show transformation or result after using product"
+    }
+}
+
+GOAL_MAPPING = {
+    "Retention (Dwell Rate)": "retention",
+    "Click-Through (CTR)": "ctr",
+    "Conversions (CVR)": "cvr"
+}
+
+
+def get_benchmark(industry: str, goal: str) -> Dict:
+    """Get benchmark data for industry and goal."""
+    goal_key = GOAL_MAPPING.get(goal, "retention")
+    return BENCHMARKS.get(industry, BENCHMARKS["Other"]).get(goal_key, {})
+
+
+def get_missing_element_recommendation(element: str) -> Dict:
+    """Get recommendation for missing element."""
+    return MISSING_ELEMENT_IMPACT.get(element, {
+        "impact": "May improve ad performance",
+        "suggestion": f"Consider adding {element} to strengthen narrative"
+    })

config.py

@@ -0,0 +1,32 @@
+import os
+
+# API Configuration - MiniMax
+MINIMAX_API_KEY = os.getenv("MINIMAX_API_KEY", "")
+MINIMAX_GROUP_ID = os.getenv("MINIMAX_GROUP_ID", "")
+MINIMAX_BASE_URL = "https://api.minimaxi.chat/v1"
+
+# API Configuration - OpenAI (for Whisper transcription)
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY", "")
+
+# Models
+MINIMAX_MODEL_VISION = "MiniMax-Text-01"
+MINIMAX_MODEL_LLM = "MiniMax-Text-01"
+
+# Processing Settings
+MAX_VIDEO_LENGTH_SECONDS = 120
+FRAME_INTERVAL_SECONDS = 2
+SUPPORTED_VIDEO_FORMATS = [".mp4", ".avi", ".mov", ".webm"]
+
+# UI Options
+INDUSTRIES = ["Apparel & Accessories", "Beauty", "Food", "Beverages", "Other"]
+CAMPAIGN_GOALS = ["Retention (Dwell Rate)", "Click-Through (CTR)", "Conversions (CVR)"]
+
+# Role category colors for UI
+CATEGORY_COLORS = {
+    "OPENING": "#4CAF50",      # Green
+    "PROBLEM": "#FF5722",       # Deep Orange
+    "PRODUCT": "#2196F3",       # Blue
+    "PERSUASIVE": "#9C27B0",    # Purple
+    "CLOSURE": "#FFC107",       # Amber
+    "OTHER": "#9E9E9E"          # Grey
+}

frame_extractor.py

@@ -0,0 +1,51 @@
+import subprocess
+import os
+import tempfile
+from typing import List, Dict
+
+
+class FrameExtractor:
+    def __init__(self, output_dir: str = None):
+        self.output_dir = output_dir or tempfile.mkdtemp()
+        os.makedirs(self.output_dir, exist_ok=True)
+
+    def extract_frames(self, video_path: str, interval_seconds: float = 2.0) -> List[Dict]:
+        """
+        Extract frames at regular intervals using FFmpeg.
+
+        Args:
+            video_path: Path to video file
+            interval_seconds: Extract one frame every N seconds
+
+        Returns:
+            List of dicts with timestamp and frame path:
+            [
+                {"timestamp": 0.0, "path": "/tmp/frame_001.jpg"},
+                {"timestamp": 2.0, "path": "/tmp/frame_002.jpg"},
+                ...
+            ]
+        """
+        fps = 1 / interval_seconds
+        output_pattern = os.path.join(self.output_dir, "frame_%03d.jpg")
+
+        cmd = [
+            'ffmpeg', '-i', video_path,
+            '-vf', f'fps={fps}',
+            '-q:v', '2',  # High quality
+            output_pattern,
+            '-y'  # Overwrite
+        ]
+
+        subprocess.run(cmd, capture_output=True, check=True)
+
+        # Build result list with timestamps
+        frames = []
+        frame_files = sorted([f for f in os.listdir(self.output_dir) if f.startswith('frame_')])
+
+        for i, frame_file in enumerate(frame_files):
+            frames.append({
+                "timestamp": i * interval_seconds,
+                "path": os.path.join(self.output_dir, frame_file)
+            })
+
+        return frames

narrative_classifier.py

@@ -0,0 +1,166 @@
+import requests
+import json
+from typing import List, Dict
+
+from taxonomy import STORY_ARCS, STORY_DEFINITION, get_taxonomy_formatted
+
+
+class NarrativeClassifier:
+    def __init__(self, api_key: str, group_id: str):
+        self.api_key = api_key
+        self.group_id = group_id
+        self.base_url = "https://api.minimaxi.chat/v1"
+
+    def _build_prompt(self, segments: List[Dict]) -> str:
+        """Build classification prompt."""
+
+        # Format segments
+        segments_text = ""
+        for seg in segments:
+            segments_text += f"\n[{seg['start']:.1f}s - {seg['end']:.1f}s]"
+            segments_text += f"\nVisual: {seg['visual']}"
+            if seg['speech']:
+                segments_text += f"\nSpeech: \"{seg['speech']}\""
+            segments_text += "\n"
+
+        # Format story arcs
+        arcs_text = ""
+        for arc_name, arc_info in STORY_ARCS.items():
+            arcs_text += f"\n- {arc_name}: {' -> '.join(arc_info['sequence'])}"
+
+        prompt = f"""You are an expert in advertising narrative structure analysis.
+
+Analyze this video advertisement segment by segment.
+
+## SEGMENTS TO ANALYZE:
+{segments_text}
+
+## FUNCTIONAL ROLE TAXONOMY:
+{get_taxonomy_formatted()}
+
+## KNOWN STORY ARCS:
+{arcs_text}
+
+## STORY DEFINITION:
+{STORY_DEFINITION}
+
+## YOUR TASK:
+
+1. For each segment, determine the PRIMARY functional role from the taxonomy
+2. Determine if this ad contains a STORY (YES/NO)
+3. Identify which STORY ARC best matches (or "Custom" if none match)
+4. List any MISSING elements that could strengthen the ad
+
+## RESPONSE FORMAT (use exactly this JSON structure):
+
+```json
+{{
+    "segments": [
+        {{
+            "timestamp": "0.0-2.0s",
+            "functional_role": "Hook",
+            "role_category": "OPENING",
+            "reasoning": "Opens with provocative question to grab attention"
+        }}
+    ],
+    "has_story": true,
+    "story_explanation": "Brief explanation of why story is present/absent",
+    "story_arc": "Problem-Solution-Outcome",
+    "detected_sequence": ["Hook", "Problem Setup", "Solution Reveal", "Call-to-Action"],
+    "missing_elements": ["Social Proof", "Outcome"]
+}}
+```
+
+Respond ONLY with valid JSON, no other text."""
+
+        return prompt
+
+    def classify(self, segments: List[Dict]) -> Dict:
+        """
+        Classify each segment and detect overall story arc.
+
+        Returns:
+            {
+                "segments": [...],
+                "has_story": True/False,
+                "story_arc": "...",
+                "detected_sequence": [...],
+                "missing_elements": [...],
+                "raw_response": "..."
+            }
+        """
+        url = f"{self.base_url}/text/chatcompletion_v2"
+
+        headers = {
+            'Authorization': f'Bearer {self.api_key}',
+            'Content-Type': 'application/json'
+        }
+
+        prompt = self._build_prompt(segments)
+
+        payload = {
+            "model": "MiniMax-Text-01",
+            "messages": [
+                {"role": "user", "content": prompt}
+            ],
+            "temperature": 0.3  # Lower temperature for more consistent classification
+        }
+
+        response = requests.post(url, headers=headers, json=payload)
+
+        if response.status_code != 200:
+            print(f"Classification API error: {response.text}")
+            return self._fallback_result(segments)
+
+        result = response.json()
+        raw_response = result['choices'][0]['message']['content']
+
+        # Parse JSON from response
+        try:
+            # Extract JSON from response (may be wrapped in markdown code block)
+            json_str = raw_response
+            if "```json" in json_str:
+                json_str = json_str.split("```json")[1].split("```")[0]
+            elif "```" in json_str:
+                json_str = json_str.split("```")[1].split("```")[0]
+
+            parsed = json.loads(json_str.strip())
+
+            # Merge with original segment data
+            for i, seg_analysis in enumerate(parsed.get('segments', [])):
+                if i < len(segments):
+                    segments[i]['functional_role'] = seg_analysis.get('functional_role', 'Unknown')
+                    segments[i]['role_category'] = seg_analysis.get('role_category', 'OTHER')
+                    segments[i]['reasoning'] = seg_analysis.get('reasoning', '')
+
+            return {
+                "segments": segments,
+                "has_story": parsed.get('has_story', False),
+                "story_explanation": parsed.get('story_explanation', ''),
+                "story_arc": parsed.get('story_arc', 'Unknown'),
+                "detected_sequence": parsed.get('detected_sequence', []),
+                "missing_elements": parsed.get('missing_elements', []),
+                "raw_response": raw_response
+            }
+
+        except json.JSONDecodeError as e:
+            print(f"JSON parse error: {e}")
+            print(f"Raw response: {raw_response}")
+            return self._fallback_result(segments, raw_response)
+
+    def _fallback_result(self, segments: List[Dict], raw_response: str = "") -> Dict:
+        """Return fallback result when parsing fails."""
+        for seg in segments:
+            seg['functional_role'] = 'Unknown'
+            seg['role_category'] = 'OTHER'
+            seg['reasoning'] = 'Classification failed'
+
+        return {
+            "segments": segments,
+            "has_story": False,
+            "story_explanation": "Unable to determine",
+            "story_arc": "Unknown",
+            "detected_sequence": [],
+            "missing_elements": [],
+            "raw_response": raw_response
+        }

report_generator.py

@@ -0,0 +1,96 @@
+from typing import Dict
+
+from benchmarks import get_benchmark, get_missing_element_recommendation, GOAL_MAPPING
+
+
+class ReportGenerator:
+    def generate(
+        self,
+        analysis: Dict,
+        industry: str,
+        campaign_goal: str
+    ) -> Dict:
+        """
+        Generate actionable report with recommendations.
+
+        Args:
+            analysis: Output from NarrativeClassifier
+            industry: Selected industry
+            campaign_goal: Selected campaign goal
+
+        Returns:
+            Complete report with summary, segments, and recommendations
+        """
+        benchmark = get_benchmark(industry, campaign_goal)
+        goal_key = GOAL_MAPPING.get(campaign_goal, "retention")
+
+        # Build recommendations
+        recommendations = []
+
+        # 1. Check if current arc matches optimal
+        current_arc = analysis.get('story_arc', 'Unknown')
+        optimal_arc = benchmark.get('best_arc', 'Unknown')
+
+        arc_matches = self._arcs_match(current_arc, optimal_arc)
+
+        if not arc_matches and optimal_arc != 'Unknown':
+            recommendations.append({
+                "priority": "HIGH",
+                "type": "arc_mismatch",
+                "action": f"Consider restructuring to {optimal_arc} arc",
+                "expected_impact": f"+{benchmark.get('uplift_percent', '?')}% {goal_key}",
+                "reasoning": benchmark.get('recommendation', '')
+            })
+
+        # 2. Check missing elements
+        missing = analysis.get('missing_elements', [])
+        for element in missing:
+            rec = get_missing_element_recommendation(element)
+            recommendations.append({
+                "priority": "MEDIUM" if element in ['Hook', 'Call-to-Action'] else "LOW",
+                "type": "missing_element",
+                "action": f"Add {element}",
+                "expected_impact": rec.get('impact', ''),
+                "reasoning": rec.get('suggestion', '')
+            })
+
+        # 3. Story presence recommendation
+        if not analysis.get('has_story', False):
+            recommendations.append({
+                "priority": "MEDIUM",
+                "type": "no_story",
+                "action": "Consider adding narrative elements",
+                "expected_impact": "+5-10% retention",
+                "reasoning": "Ads with stories show 5-10% better retention than feature-focused ads"
+            })
+
+        # Sort by priority
+        priority_order = {"HIGH": 0, "MEDIUM": 1, "LOW": 2}
+        recommendations.sort(key=lambda x: priority_order.get(x['priority'], 3))
+
+        return {
+            "summary": {
+                "has_story": analysis.get('has_story', False),
+                "story_explanation": analysis.get('story_explanation', ''),
+                "detected_arc": current_arc,
+                "optimal_arc_for_goal": optimal_arc,
+                "arc_matches_optimal": arc_matches,
+                "potential_uplift": f"+{benchmark.get('uplift_percent', '?')}%"
+            },
+            "segments": analysis.get('segments', []),
+            "detected_sequence": analysis.get('detected_sequence', []),
+            "missing_elements": missing,
+            "recommendations": recommendations,
+            "benchmark": benchmark
+        }
+
+    def _arcs_match(self, current: str, optimal: str) -> bool:
+        """Check if arcs match (fuzzy matching)."""
+        if current == optimal:
+            return True
+
+        # Normalize
+        current_norm = current.lower().replace('-', '').replace(' ', '')
+        optimal_norm = optimal.lower().replace('-', '').replace(' ', '')
+
+        return current_norm == optimal_norm

requirements.txt

@@ -0,0 +1,21 @@
+# StoryLens - Requirements
+
+# Web framework
+streamlit>=1.28.0
+
+# Video processing
+yt-dlp>=2023.10.13
+moviepy>=1.0.3
+ffmpeg-python>=0.2.0
+
+# Image processing
+Pillow>=10.0.0
+
+# HTTP requests
+requests>=2.31.0
+
+# Environment variables
+python-dotenv>=1.0.0
+
+# Speech-to-Text (OpenAI Whisper API)
+openai>=1.0.0

segment_synchronizer.py

@@ -0,0 +1,54 @@
+from typing import List, Dict
+
+
+class SegmentSynchronizer:
+    def synchronize(
+        self,
+        frames: List[Dict],      # [{"timestamp": 0.0, "path": "...", "description": "..."}]
+        transcript: List[Dict]   # [{"start": 0.0, "end": 3.2, "text": "..."}]
+    ) -> List[Dict]:
+        """
+        Create unified segments with visual + speech.
+
+        Returns:
+            List of synchronized segments:
+            [
+                {
+                    "start": 0.0,
+                    "end": 2.0,
+                    "frame_path": "/tmp/frame_001.jpg",
+                    "visual": "Woman looking frustrated in kitchen",
+                    "speech": "Tired of everyday exhaustion?"
+                },
+                ...
+            ]
+        """
+        segments = []
+
+        for i, frame in enumerate(frames):
+            timestamp = frame['timestamp']
+
+            # Calculate segment end (next frame timestamp or +interval)
+            if i < len(frames) - 1:
+                end_time = frames[i + 1]['timestamp']
+            else:
+                end_time = timestamp + 2.0  # Default interval
+
+            # Find overlapping speech
+            speech_text = ""
+            for t in transcript:
+                # Check if speech segment overlaps with this frame's time window
+                if t['end'] > timestamp and t['start'] < end_time:
+                    speech_text += " " + t['text']
+
+            speech_text = speech_text.strip()
+
+            segments.append({
+                "start": timestamp,
+                "end": end_time,
+                "frame_path": frame['path'],
+                "visual": frame['description'],
+                "speech": speech_text if speech_text else None
+            })
+
+        return segments

taxonomy.py

@@ -0,0 +1,111 @@
+FUNCTIONAL_ROLES = {
+    "OPENING": {
+        "Hook": "Grabs viewers' attention or interest; appears in first few seconds",
+        "Establish Context": "Sets up the status quo—who, where, or when—before story progression"
+    },
+    "PROBLEM": {
+        "Problem Setup": "Presents a problem, need, or pain point to resolve for the first time",
+        "Problem Agitation": "Amplifies the problem to make it relatable or severe"
+    },
+    "PRODUCT": {
+        "Feature Explanation": "Explains product features and why it delivers benefits; goes beyond just showing",
+        "Product Highlight": "Presents key product attributes or benefits (surface-level showcasing)",
+        "Demonstration": "Shows the product being used or tested to accomplish a task",
+        "Comparison": "Contrasts product with competitors or previous states",
+        "Social Proof": "Shows reviews or testimonials from other people",
+        "Solution Reveal": "Presents product as solution to a problem"
+    },
+    "PERSUASIVE": {
+        "Emotional Appeal": "Uses emotions to connect with and engage the audience",
+        "Humor": "Uses comedic elements to entertain and make message relatable",
+        "Aspirational Vision": "Depicts an ideal lifestyle or future enabled by the product",
+        "Promotion": "Communicates offer mechanics: discount, bundle, code, pricing terms",
+        "Urgency Trigger": "Adds time pressure to accelerate action",
+        "Scarcity Trigger": "Highlights limited availability to create FOMO"
+    },
+    "CLOSURE": {
+        "Call-to-Action": "Cues to act; drives immediate action",
+        "Outcome": "Shows post-intervention payoff or transformation",
+        "Branding Moment": "Displays brand identity (logo, tagline, slogans)",
+        "Insight/Philosophy": "Expresses brand philosophy; leads viewers to discover something new"
+    },
+    "OTHER": {
+        "Visual Filler": "Provides transitional pacing without narrative contribution"
+    }
+}
+
+STORY_ARCS = {
+    "Problem-Solution-Outcome": {
+        "sequence": ["Problem Setup", "Solution Reveal", "Outcome"],
+        "description": "Introduces a problem, offers a solution, and shows the outcome"
+    },
+    "Hook-Feature-Benefit-Action": {
+        "sequence": ["Hook", "Feature Explanation", "Product Highlight", "Call-to-Action"],
+        "abbreviation": "HFBA",
+        "description": "Grabs attention, explains features, highlights benefits, drives action"
+    },
+    "AIDA": {
+        "sequence": ["Hook", "Feature Explanation", "Aspirational Vision", "Call-to-Action"],
+        "description": "Attention-Interest-Desire-Action classic marketing funnel"
+    },
+    "Social-Proof-Action": {
+        "sequence": ["Social Proof", "Call-to-Action"],
+        "abbreviation": "SPA",
+        "description": "Shows testimonials/reviews then drives action"
+    },
+    "Problem-Agitate-Solution": {
+        "sequence": ["Problem Setup", "Problem Agitation", "Solution Reveal"],
+        "abbreviation": "PAS",
+        "description": "Presents problem, amplifies pain, offers solution"
+    },
+    "Before-After-Bridge": {
+        "sequence": ["Establish Context", "Outcome", "Solution Reveal"],
+        "abbreviation": "BAB",
+        "description": "Shows current situation, desired outcome, product as bridge"
+    },
+    "Hook-Problem-Solution": {
+        "sequence": ["Hook", "Problem Setup", "Solution Reveal"],
+        "abbreviation": "HPS",
+        "description": "Grabs attention, presents problem, offers solution"
+    },
+    "Feature-Benefit-Action": {
+        "sequence": ["Feature Explanation", "Product Highlight", "Call-to-Action"],
+        "abbreviation": "FBA",
+        "description": "Direct product-focused approach"
+    }
+}
+
+STORY_DEFINITION = """
+A story is an account of an event or a sequence of connected events
+that leads to a transition from an initial state to a later stage or outcome.
+
+Signals of STORY PRESENT:
+- Dialogues between characters
+- Sharing of personal experiences
+- Inclusion of challenges/conflicts/problem solutions
+- Character transformation or journey
+
+Signals of STORY ABSENT:
+- Announcer/narrator voiceover only
+- Promotional language dominance
+- Heavy focus on product features without context
+- Visual mashups without narrative connection
+"""
+
+
+def get_taxonomy_formatted() -> str:
+    """Return taxonomy as formatted string for prompts."""
+    lines = []
+    for category, roles in FUNCTIONAL_ROLES.items():
+        lines.append(f"\n**{category}**")
+        for role, description in roles.items():
+            lines.append(f"- {role}: {description}")
+    return "\n".join(lines)
+
+
+def get_role_category(role_name: str) -> str:
+    """Get category for a role name."""
+    for category, roles in FUNCTIONAL_ROLES.items():
+        if role_name in roles:
+            return category
+    return "OTHER"

video_loader.py

@@ -0,0 +1,79 @@
+import tempfile
+import os
+import subprocess
+import json
+import shutil
+from typing import Optional
+
+import yt_dlp
+
+
+class VideoLoader:
+    def __init__(self):
+        self.temp_dir = tempfile.mkdtemp()
+
+    def load_youtube(self, url: str) -> Optional[str]:
+        """
+        Download YouTube video.
+
+        Args:
+            url: YouTube URL
+
+        Returns:
+            Path to downloaded video file, or None if failed
+        """
+        output_path = os.path.join(self.temp_dir, "video.mp4")
+
+        ydl_opts = {
+            'format': 'bestvideo[ext=mp4]+bestaudio[ext=m4a]/bestvideo+bestaudio/best',
+            'outtmpl': output_path,
+            'merge_output_format': 'mp4',
+            'quiet': True,
+            'no_warnings': True,
+            'postprocessors': [{
+                'key': 'FFmpegVideoConvertor',
+                'preferedformat': 'mp4',
+            }],
+        }
+
+        try:
+            with yt_dlp.YoutubeDL(ydl_opts) as ydl:
+                ydl.download([url])
+            return output_path
+        except Exception as e:
+            print(f"Error downloading YouTube video: {e}")
+            return None
+
+    def load_local(self, uploaded_file) -> Optional[str]:
+        """
+        Save uploaded file to temp directory.
+
+        Args:
+            uploaded_file: Streamlit UploadedFile object
+
+        Returns:
+            Path to saved file
+        """
+        output_path = os.path.join(self.temp_dir, uploaded_file.name)
+
+        with open(output_path, "wb") as f:
+            f.write(uploaded_file.read())
+
+        return output_path
+
+    def get_video_duration(self, video_path: str) -> float:
+        """Get video duration in seconds using ffprobe."""
+        cmd = [
+            'ffprobe', '-v', 'quiet', '-print_format', 'json',
+            '-show_format', video_path
+        ]
+
+        result = subprocess.run(cmd, capture_output=True, text=True)
+        data = json.loads(result.stdout)
+
+        return float(data['format']['duration'])
+
+    def cleanup(self):
+        """Remove temp files."""
+        if os.path.exists(self.temp_dir):
+            shutil.rmtree(self.temp_dir)

vision_analyzer.py

@@ -0,0 +1,91 @@
+import base64
+import requests
+from typing import List, Dict
+
+
+class VisionAnalyzer:
+    def __init__(self, api_key: str, group_id: str):
+        self.api_key = api_key
+        self.group_id = group_id
+        self.base_url = "https://api.minimaxi.chat/v1"
+
+        self.prompt = """Describe this video frame in one concise sentence. Focus on:
+- Who/what is shown (people, products, text overlays)
+- Setting/environment
+- Actions or emotions displayed
+- Any visible brand elements or text
+
+Be factual and specific. Do not interpret or add assumptions."""
+
+    def _encode_image(self, image_path: str) -> str:
+        """Encode image to base64."""
+        with open(image_path, "rb") as f:
+            return base64.b64encode(f.read()).decode('utf-8')
+
+    def describe_frame(self, image_path: str, timestamp: float) -> str:
+        """
+        Generate description of a single frame.
+
+        Returns:
+            Description string, e.g., "Woman looking frustrated in messy kitchen"
+        """
+        url = f"{self.base_url}/text/chatcompletion_v2"
+
+        headers = {
+            'Authorization': f'Bearer {self.api_key}',
+            'Content-Type': 'application/json'
+        }
+
+        image_data = self._encode_image(image_path)
+
+        payload = {
+            "model": "MiniMax-Text-01",
+            "messages": [
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text", "text": self.prompt},
+                        {
+                            "type": "image_url",
+                            "image_url": {"url": f"data:image/jpeg;base64,{image_data}"}
+                        }
+                    ]
+                }
+            ]
+        }
+
+        response = requests.post(url, headers=headers, json=payload)
+
+        if response.status_code != 200:
+            print(f"Vision API error: {response.text}")
+            return f"[Frame at {timestamp}s - description unavailable]"
+
+        result = response.json()
+
+        try:
+            return result['choices'][0]['message']['content']
+        except (KeyError, IndexError):
+            return f"[Frame at {timestamp}s - description unavailable]"
+
+    def describe_frames_batch(self, frames: List[Dict]) -> List[Dict]:
+        """
+        Describe all frames.
+
+        Args:
+            frames: [{"timestamp": 0.0, "path": "/tmp/frame_001.jpg"}, ...]
+
+        Returns:
+            [{"timestamp": 0.0, "path": "...", "description": "Woman looking..."}, ...]
+        """
+        results = []
+
+        for frame in frames:
+            description = self.describe_frame(frame['path'], frame['timestamp'])
+
+            results.append({
+                "timestamp": frame['timestamp'],
+                "path": frame['path'],
+                "description": description
+            })
+
+        return results