Movie AI Pipeline

import subprocess

def detect_scenes(video_path: str, threshold: float = 0.4) -> list[float]:
    """Returns list of cut timestamps in seconds."""
    result = subprocess.run([
        "ffmpeg", "-i", video_path,
        "-vf", f"select='gt(scene,{threshold})',showinfo",
        "-vsync", "vfr", "-f", "null", "-"
    ], capture_output=True, text=True)
    timestamps = []
    for line in result.stderr.splitlines():
        if "pts_time" in line:
            for part in line.split():
                if part.startswith("pts_time:"):
                    timestamps.append(float(part.split(':')[1]))
    return timestamps

import cv2, subprocess
from pathlib import Path

def get_fps(video_path: str) -> float:
    cap = cv2.VideoCapture(video_path)
    fps = cap.get(cv2.CAP_PROP_FPS)
    cap.release()
    if abs(fps - 29.97)  < 0.01: return 29.97
    if abs(fps - 23.976) < 0.01: return 23.976
    return fps

def seconds_to_smpte(seconds: float, fps: float) -> str:
    """Convert float seconds to SMPTE HH:MM:SS:FF"""
    total_frames = int(round(seconds * fps))
    ff = total_frames % int(fps)
    ss = (total_frames // int(fps)) % 60
    mm = (total_frames // (int(fps) * 60)) % 60
    hh =  total_frames // (int(fps) * 3600)
    return f'{hh:02d}:{mm:02d}:{ss:02d}:{ff:02d}'

def extract_keyframes(video_path, scene_cuts, output_dir) -> dict:
    fps = get_fps(video_path)
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    cap = cv2.VideoCapture(video_path)
    total_secs = cap.get(cv2.CAP_PROP_FRAME_COUNT) / fps
    cuts = [0.0] + scene_cuts + [total_secs]
    segments = [(cuts[i], cuts[i+1]) for i in range(len(cuts)-1)]
    result = {}
    for idx, (start, end) in enumerate(segments):
        mid = (start + end) / 2
        cap.set(cv2.CAP_PROP_POS_MSEC, mid * 1000)
        ret, frame = cap.read()
        if ret:
            path = f'{output_dir}/scene_{idx:04d}_{mid:.1f}s.jpg'
            cv2.imwrite(path, frame)
            result[idx] = {
                'start':      start,
                'end':        end,
                'tc_start':   seconds_to_smpte(start, fps),
                'tc_end':     seconds_to_smpte(end,   fps),
                'fps':        fps,
                'frame_path': path
            }
    cap.release()
    return result

def extract_audio(video_path: str, output_path: str) -> str:
    subprocess.run([
        "ffmpeg", "-i", video_path,
        "-vn", "-acodec", "pcm_s16le",
        "-ar", "16000", "-ac", "1",
        output_path, "-y"
    ], check=True)
    return output_path

from elevenlabs.client import ElevenLabs
import os
client = ElevenLabs(api_key=os.getenv('ELEVENLABS_API_KEY'))

def transcribe(audio_path: str) -> dict:
    """Returns speaker-diarised transcript + audio events."""
    with open(audio_path, 'rb') as f:
        response = client.speech_to_text.convert(
            file=f,
            model_id='scribe_v2',
            diarize=True,              # speaker_0, speaker_1 ...
            tag_audio_events=True,     # [laughter], [music] ...
            timestamps_granularity='word'
        )
    speakers, events = {}, []
    for word in response.words:
        if word.type == 'audio_event':
            events.append({'time': word.start, 'event': word.text})
        else:
            spk = getattr(word, 'speaker_id', 'speaker_0')
            speakers.setdefault(spk, []).append(
                {'word': word.text, 'start': word.start, 'end': word.end}
            )
    return {
        'full_text':    response.text,
        'speakers':     speakers,   # {'speaker_0': [{word, start, end}, ...]}
        'audio_events': events      # [{'time': 12.3, 'event': '[laughter]'}]
    }

import google.generativeai as genai
import json, os, base64
genai.configure(api_key=os.getenv('GEMINI_API_KEY'))
model = genai.GenerativeModel('gemini-2.5-flash')

VISION_PROMPT = '''Analyse this film frame. Return ONLY valid JSON, no markdown:
{
  "lighting":       {"type": "low-key|high-key|natural|neon|...",
                     "quality": "hard|soft|mixed"},
  "framing":        {"shot_type": "extreme_close|close|medium|wide|extreme_wide",
                     "angle": "low|eye|high|dutch"},
  "color_grading":  {"dominant_palette": ["color1","color2"],
                     "tone": "warm|cold|desaturated|vibrant"},
  "objects":        ["list", "of", "visible", "objects"],
  "camera_movement":"static|pan|tilt|dolly|handheld|unknown",
  "lens_estimate":  "wide_angle|standard|telephoto|unknown",
  "mood":           "primary mood in 2-3 words"
}'''

def analyze_frame(frame_path: str) -> dict:
    with open(frame_path, 'rb') as f:
        img_data = base64.b64encode(f.read()).decode()
    response = model.generate_content([
        {'mime_type': 'image/jpeg', 'data': img_data},
        VISION_PROMPT
    ])
    text = response.text.strip()
    if text.startswith('```'):
        text = text.split('```')[1]
        if text.startswith('json'): text = text[4:]
    return json.loads(text.strip())

import anthropic, json, os
client = anthropic.Anthropic(api_key=os.getenv('ANTHROPIC_API_KEY'))

def analyze_context(vision: dict, transcript: dict, film_title: str) -> dict:
    prompt = f"""You are a film studies expert. Given this scene from "{film_title}",
identify deeper cinematic context. Return ONLY valid JSON:
{{
  "symbolism":                ["list of symbolic elements if any"],
  "cinematographic_references":["director/film influences if detectable"],
  "narrative_function":        "establishing|conflict|climax|resolution|transition",
  "emotional_arc":             "description of emotional progression",
  "thematic_tags":             ["theme1", "theme2"]
}}
Vision: {json.dumps(vision)}
Dialogue: {transcript['full_text'][:500]}
Audio events: {transcript["audio_events"]}"""
    msg = client.messages.create(
        model='claude-sonnet-4-6',
        max_tokens=600,
        messages=[{'role': 'user', 'content': prompt}]
    )
    text = msg.content[0].text.strip()
    if text.startswith('```'):
        text = text.split('```')[1].lstrip('json').strip()
    return json.loads(text)