Video Overlays

# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info

# Virtual environments
.venv

3.13

[project]
name = "video-overlays"
version = "0.1.0"
description = "Detect content mentions and generate overlay specifications"
requires-python = ">=3.13"
dependencies = [
    "playwright>=1.49.0",
    "requests>=2.32.0",
    "beautifulsoup4>=4.13.0",
]

"""Phase 1: Detect content mentions in transcript for overlay generation.

Scans the transcript for references to blog posts, talks, tools, people,
and key terms. Uses word-level timestamps to map each mention to source time.

Input: <stem>_analysis.json + <stem>_edl.json
Output: <stem>_mentions.json
"""

import json
import os
import re
import sys
from pathlib import Path


# ── Mention type patterns ────────────────────────────────────────────────────

# Phrases that indicate a blog post reference
BLOG_PATTERNS = [
    r"(?:i |we |I )?wrote (?:a |an )?(?:blog )?post (?:about|on|called|titled)",
    r"(?:i |we |I )?(?:published|released) (?:a |an )?(?:blog )?(?:post|article)",
    r"(?:my |our )?blog post (?:about|on|called|titled)",
    r"(?:there'?s|there is) (?:a |an )?(?:blog )?post (?:about|on)",
    r"(?:check out|read|see) (?:my |our |the )?(?:blog )?post",
    r"(?:in |on )?(?:my |our |the )?blog",
]

# Phrases that indicate a talk/presentation reference
TALK_PATTERNS = [
    r"(?:i |we )?(?:gave|did|presented) (?:a |an )?(?:talk|presentation|keynote|session)",
    r"(?:my |our )?(?:talk|presentation|keynote) (?:at|about|on|called|titled|from)",
    r"(?:there'?s|there is) (?:a |an )?(?:talk|presentation) (?:about|on|from)",
    r"(?:check out|watch|see) (?:my |our |the )?(?:talk|presentation)",
    r"(?:at |from )?(?:a |the )?conference",
]

# Phrases that indicate a tool/library/product reference
TOOL_PATTERNS = [
    r"(?:using|use|used|try|check out|install|import) (\w+(?:\.\w+)?(?:[-/]\w+)?)",
    r"(?:a |the )?(?:tool|library|framework|package|sdk|api) called (\w+(?:[-./]\w+)?)",
    r"(?:built (?:with|on|using)|powered by) (\w+(?:[-./]\w+)?)",
]

# URL patterns (spoken or in transcript)
URL_PATTERN = r"(?:https?://)?(?:www\.)?([a-zA-Z0-9][-a-zA-Z0-9]*(?:\.[a-zA-Z]{2,})+)(?:/[^\s,.)]*)*"

# Key term indicators
KEY_TERM_PATTERNS = [
    r"(?:this is (?:called|known as|what we call)) [\"']?(\w[\w\s]{1,40})[\"']?",
    r"(?:the concept of|the idea of|the term) [\"']?(\w[\w\s]{1,40})[\"']?",
    r"(?:what (?:i |we )?(?:call|mean by|refer to as)) [\"']?(\w[\w\s]{1,40})[\"']?",
]

# Person reference patterns
PERSON_PATTERNS = [
    r"(\w+ \w+) (?:wrote|created|built|invented|designed|proposed|introduced)",
    r"(?:by|from|according to) (\w+ \w+)",
    r"(\w+ \w+)(?:'s| 's) (?:work|paper|book|talk|blog|library|framework|tool|project)",
]


# ── Mention detection ────────────────────────────────────────────────────────

def _find_word_window(words, char_start, char_end, text):
    """Map character offsets in the full transcript text to word indices and timestamps."""
    # Build character offset map for each word
    pos = 0
    word_char_starts = []
    for w in words:
        # Find this word in text starting from pos
        idx = text.find(w["word"], pos)
        if idx == -1:
            idx = pos
        word_char_starts.append(idx)
        pos = idx + len(w["word"])

    # Find first word that overlaps char_start
    first_word = 0
    for i, wcs in enumerate(word_char_starts):
        if wcs + len(words[i]["word"]) > char_start:
            first_word = i
            break

    # Find last word that overlaps char_end
    last_word = len(words) - 1
    for i, wcs in enumerate(word_char_starts):
        if wcs >= char_end:
            last_word = max(0, i - 1)
            break

    return first_word, last_word


def _context_around(words, first_idx, last_idx, context_words=5):
    """Get text context around a mention."""
    start = max(0, first_idx - context_words)
    end = min(len(words), last_idx + context_words + 1)
    return " ".join(w["word"] for w in words[start:end])


def detect_mentions(analysis, edl):
    """Detect content mentions in the transcript.

    Returns a list of mention dicts with type, text, timing, and context.
    """
    words = analysis["transcript"]["words"]
    text = analysis["transcript"]["text"]
    source_file = analysis["source_file"]

    if not words or not text:
        return []

    # Build set of kept time ranges from EDL for filtering
    kept_ranges = []
    for seg in edl.get("segments", []):
        if seg.get("action") == "keep":
            kept_ranges.append((seg["start"], seg["end"]))

    mentions = []
    mention_id = 0

    def _add_mention(mention_type, label, match_start, match_end, url=None, extra_meta=None):
        nonlocal mention_id
        first_idx, last_idx = _find_word_window(words, match_start, match_end, text)

        source_start = words[first_idx]["start"]
        source_end = words[last_idx]["end"]

        # Check if this mention falls within a kept segment
        in_kept = any(
            ks <= source_start <= ke or ks <= source_end <= ke
            for ks, ke in kept_ranges
        )

        if not in_kept:
            return  # Skip mentions in cut segments

        mention_id += 1
        mention = {
            "id": f"mention_{mention_id:03d}",
            "type": mention_type,
            "label": label.strip(),
            "matched_text": text[match_start:match_end].strip(),
            "context": _context_around(words, first_idx, last_idx),
            "timing": {
                "source_start": round(source_start, 3),
                "source_end": round(source_end, 3),
            },
        }
        if url:
            mention["url"] = url
        if extra_meta:
            mention["metadata"] = extra_meta

        mentions.append(mention)

    # Detect blog post mentions
    for pattern in BLOG_PATTERNS:
        for m in re.finditer(pattern, text, re.IGNORECASE):
            # Try to extract the topic from text after the match
            after = text[m.end():m.end() + 100].strip()
            topic = after.split(".")[0].split(",")[0].strip()[:80]
            label = topic if topic else m.group(0)
            _add_mention("blog-card", label, m.start(), m.end())

    # Detect talk mentions
    for pattern in TALK_PATTERNS:
        for m in re.finditer(pattern, text, re.IGNORECASE):
            after = text[m.end():m.end() + 100].strip()
            topic = after.split(".")[0].split(",")[0].strip()[:80]
            label = topic if topic else m.group(0)
            _add_mention("talk-thumbnail", label, m.start(), m.end())

    # Detect URL mentions
    for m in re.finditer(URL_PATTERN, text):
        url = m.group(0)
        if not url.startswith("http"):
            url = "https://" + url
        domain = m.group(1)
        _add_mention("url-callout", domain, m.start(), m.end(), url=url)

    # Detect key term definitions
    for pattern in KEY_TERM_PATTERNS:
        for m in re.finditer(pattern, text, re.IGNORECASE):
            term = m.group(1).strip() if m.lastindex else m.group(0)
            _add_mention("key-term", term, m.start(), m.end())

    # Detect tool/library mentions (only well-known or explicitly named)
    for pattern in TOOL_PATTERNS:
        for m in re.finditer(pattern, text, re.IGNORECASE):
            tool_name = m.group(1) if m.lastindex else m.group(0)
            # Filter out very common/generic words
            if tool_name.lower() in {"it", "this", "that", "them", "something", "the", "a", "an"}:
                continue
            if len(tool_name) < 2:
                continue
            _add_mention("tool", tool_name, m.start(), m.end())

    # Deduplicate: if same label appears multiple times, keep the first occurrence
    seen_labels = set()
    deduped = []
    for mention in mentions:
        key = (mention["type"], mention["label"].lower())
        if key not in seen_labels:
            seen_labels.add(key)
            deduped.append(mention)

    # Sort by source_start time
    deduped.sort(key=lambda m: m["timing"]["source_start"])

    return deduped


# ── CLI ──────────────────────────────────────────────────────────────────────

def _find_project_root(start_path):
    """Walk up from start_path to find the project root."""
    current = os.path.dirname(start_path) if os.path.isfile(start_path) else start_path
    while current != os.path.dirname(current):
        if os.path.isdir(os.path.join(current, ".claude")) or os.path.isdir(os.path.join(current, ".git")):
            return current
        current = os.path.dirname(current)
    return os.path.dirname(start_path) if os.path.isfile(start_path) else start_path


if __name__ == "__main__":
    if len(sys.argv) < 3:
        print("Usage: uv run detect_mentions.py <analysis_json> <edl_json> [--output-dir <dir>]")
        sys.exit(1)

    analysis_path = sys.argv[1]
    edl_path = sys.argv[2]
    output_dir = None

    args = sys.argv[3:]
    i = 0
    while i < len(args):
        if args[i] == "--output-dir" and i + 1 < len(args):
            output_dir = args[i + 1]
            i += 2
        else:
            i += 1

    with open(analysis_path) as f:
        analysis = json.load(f)
    with open(edl_path) as f:
        edl = json.load(f)

    stem = Path(analysis["source_file"]).stem

    if output_dir is None:
        output_dir = os.path.dirname(os.path.abspath(analysis_path))
    os.makedirs(output_dir, exist_ok=True)

    print(f"Detecting mentions in: {stem}")
    mentions = detect_mentions(analysis, edl)
    print(f"  Found {len(mentions)} mentions:")

    for m in mentions:
        t = m["timing"]
        print(f"  [{t['source_start']:.1f}s] {m['type']}: {m['label']}")

    output = {
        "source_file": analysis["source_file"],
        "stem": stem,
        "mentions": mentions,
    }

    output_path = os.path.join(output_dir, f"{stem}_mentions.json")
    with open(output_path, "w") as f:
        json.dump(output, f, indent=2)
    print(f"\nMentions saved to {output_path}")
    print("Review these mentions and confirm which ones to overlay before proceeding.")

"""Phase 2: Gather visual assets for confirmed mentions.

For each mention in the reviewed mentions file, fetch appropriate visual assets:
- Blog posts: screenshot via Playwright or OG image
- Talks: YouTube/conference thumbnail
- Tools: logo or homepage screenshot
- URLs: OG image or screenshot

Input: <stem>_mentions.json (reviewed/confirmed by user)
Output: screenshots/images in claude-edits/overlays/assets/
"""

import json
import os
import re
import sys
from pathlib import Path
from urllib.parse import urlparse

import requests
from bs4 import BeautifulSoup


def fetch_og_metadata(url):
    """Fetch Open Graph metadata from a URL."""
    try:
        resp = requests.get(url, timeout=10, headers={
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
        })
        resp.raise_for_status()
        soup = BeautifulSoup(resp.text, "html.parser")

        og = {}
        for tag in soup.find_all("meta"):
            prop = tag.get("property", "") or tag.get("name", "")
            if prop.startswith("og:"):
                og[prop[3:]] = tag.get("content", "")
            elif prop == "twitter:image":
                og.setdefault("image", tag.get("content", ""))

        # Fallback to title tag
        if "title" not in og:
            title_tag = soup.find("title")
            if title_tag:
                og["title"] = title_tag.text.strip()

        return og
    except Exception as e:
        print(f"  Warning: could not fetch OG data from {url}: {e}")
        return {}


def download_image(url, output_path):
    """Download an image from a URL."""
    try:
        resp = requests.get(url, timeout=15, headers={
            "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
        })
        resp.raise_for_status()
        content_type = resp.headers.get("content-type", "")
        if "image" not in content_type and "octet-stream" not in content_type:
            print(f"  Warning: {url} returned content-type {content_type}, skipping")
            return False
        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        with open(output_path, "wb") as f:
            f.write(resp.content)
        return True
    except Exception as e:
        print(f"  Warning: could not download {url}: {e}")
        return False


def screenshot_url(url, output_path, width=1280, height=800):
    """Take a screenshot of a URL using Playwright."""
    try:
        from playwright.sync_api import sync_playwright
    except ImportError:
        print("  Warning: playwright not available, skipping screenshot")
        return False

    try:
        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        with sync_playwright() as p:
            browser = p.chromium.launch(headless=True)
            page = browser.new_page(viewport={"width": width, "height": height})
            page.goto(url, wait_until="networkidle", timeout=15000)
            page.screenshot(path=output_path, type="png")
            browser.close()
        return True
    except Exception as e:
        print(f"  Warning: screenshot failed for {url}: {e}")
        return False


def fetch_youtube_thumbnail(url):
    """Extract YouTube video ID and return thumbnail URL."""
    patterns = [
        r"(?:youtube\.com/watch\?v=|youtu\.be/|youtube\.com/embed/)([a-zA-Z0-9_-]{11})",
    ]
    for pattern in patterns:
        m = re.search(pattern, url)
        if m:
            video_id = m.group(1)
            # maxresdefault is highest quality, falls back to hqdefault
            return f"https://img.youtube.com/vi/{video_id}/maxresdefault.jpg"
    return None


def gather_asset_for_mention(mention, assets_dir):
    """Gather the visual asset for a single mention.

    Returns updated mention dict with asset_path and metadata.
    """
    mention_id = mention["id"]
    mention_type = mention["type"]
    url = mention.get("url")
    asset_path = os.path.join(assets_dir, f"{mention_id}.png")

    metadata = mention.get("metadata", {})

    if mention_type in ("blog-card", "url-callout", "tool", "generic-image"):
        if url:
            # Try OG metadata first
            og = fetch_og_metadata(url)
            metadata.update({
                "og_title": og.get("title", mention.get("label", "")),
                "og_description": og.get("description", ""),
                "og_image": og.get("image", ""),
                "domain": urlparse(url).netloc,
            })

            # Try OG image first (lighter weight)
            if og.get("image"):
                if download_image(og["image"], asset_path):
                    mention["asset_path"] = asset_path
                    mention["metadata"] = metadata
                    return mention

            # Fall back to screenshot
            if screenshot_url(url, asset_path):
                mention["asset_path"] = asset_path
                mention["metadata"] = metadata
                return mention

    elif mention_type == "talk-thumbnail":
        if url:
            # Check for YouTube URL
            thumb_url = fetch_youtube_thumbnail(url)
            if thumb_url:
                jpg_path = os.path.join(assets_dir, f"{mention_id}.jpg")
                if download_image(thumb_url, jpg_path):
                    mention["asset_path"] = jpg_path
                    mention["metadata"] = metadata
                    return mention

            # Try OG image
            og = fetch_og_metadata(url)
            metadata.update({
                "og_title": og.get("title", mention.get("label", "")),
                "og_description": og.get("description", ""),
                "domain": urlparse(url).netloc,
            })
            if og.get("image"):
                if download_image(og["image"], asset_path):
                    mention["asset_path"] = asset_path
                    mention["metadata"] = metadata
                    return mention

    elif mention_type == "key-term":
        # No asset needed for key terms — rendered as text
        mention["metadata"] = metadata
        return mention

    # If we get here, no asset was fetched
    mention["metadata"] = metadata
    return mention


def gather_assets(mentions_data, output_dir):
    """Gather assets for all confirmed mentions.

    Args:
        mentions_data: Parsed mentions JSON
        output_dir: Base output directory (claude-edits/)

    Returns:
        Updated mentions_data with asset_path and metadata filled in
    """
    assets_dir = os.path.join(output_dir, "overlays", "assets")
    os.makedirs(assets_dir, exist_ok=True)

    mentions = mentions_data.get("mentions", [])
    print(f"Gathering assets for {len(mentions)} mentions...")

    for i, mention in enumerate(mentions):
        print(f"\n  [{i + 1}/{len(mentions)}] {mention['type']}: {mention['label']}")
        mention = gather_asset_for_mention(mention, assets_dir)
        mentions[i] = mention

        if mention.get("asset_path"):
            print(f"    Asset: {mention['asset_path']}")
        else:
            print(f"    No asset (will render as text-only overlay)")

    mentions_data["mentions"] = mentions
    return mentions_data


# ── CLI ──────────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    if len(sys.argv) < 2:
        print("Usage: uv run gather_assets.py <mentions_json> [--output-dir <dir>]")
        sys.exit(1)

    mentions_path = sys.argv[1]
    output_dir = None

    args = sys.argv[2:]
    i = 0
    while i < len(args):
        if args[i] == "--output-dir" and i + 1 < len(args):
            output_dir = args[i + 1]
            i += 2
        else:
            i += 1

    with open(mentions_path) as f:
        mentions_data = json.load(f)

    if output_dir is None:
        output_dir = os.path.dirname(os.path.abspath(mentions_path))

    mentions_data = gather_assets(mentions_data, output_dir)

    # Save updated mentions with asset paths
    with open(mentions_path, "w") as f:
        json.dump(mentions_data, f, indent=2)
    print(f"\nUpdated mentions saved to {mentions_path}")

"""Phase 3: Generate overlay specification from reviewed mentions.

Maps mention timing to display parameters and produces the overlay spec
that video-remotion reads to render overlay graphics.

Input: <stem>_mentions.json (with assets gathered) + <stem>_edl.json
Output: <stem>_overlays.json
"""

import json
import os
import sys
from pathlib import Path


# Default display durations by type (seconds)
DEFAULT_DURATIONS = {
    "blog-card": 6.0,
    "talk-thumbnail": 5.0,
    "url-callout": 4.0,
    "key-term": 4.0,
    "tool": 5.0,
    "generic-image": 5.0,
}

# Default positions by type
DEFAULT_POSITIONS = {
    "blog-card": "lower-right",
    "talk-thumbnail": "lower-right",
    "url-callout": "lower-third",
    "key-term": "lower-third",
    "tool": "lower-right",
    "generic-image": "lower-right",
}

# Minimum gap between overlays (seconds) — avoid visual clutter
MIN_OVERLAY_GAP = 2.0


def compute_display_timing(mention, edl_segments):
    """Compute display timing for an overlay.

    The overlay should appear shortly after the mention starts and stay
    on screen for the default duration, but not extend past the end of
    the segment it falls in.
    """
    mention_type = mention["type"]
    source_start = mention["timing"]["source_start"]
    source_end = mention["timing"]["source_end"]

    # Display starts 0.5s after the mention begins (let the viewer hear the reference first)
    display_start = source_start + 0.5

    # Default duration for this type
    duration = DEFAULT_DURATIONS.get(mention_type, 5.0)

    # Find the EDL segment this mention falls in
    containing_seg = None
    for seg in edl_segments:
        if seg.get("action") == "keep" and seg["start"] <= source_start <= seg["end"]:
            containing_seg = seg
            break

    # Don't extend past the segment boundary
    if containing_seg:
        max_end = containing_seg["end"]
        display_end = min(display_start + duration, max_end - 0.3)
    else:
        display_end = display_start + duration

    # Ensure minimum display time
    actual_duration = max(display_end - display_start, 2.0)

    return {
        "source_start": round(display_start, 3),
        "source_end": round(display_start + actual_duration, 3),
        "display_duration": round(actual_duration, 3),
    }


def resolve_overlaps(overlays):
    """Resolve overlapping overlays by shifting later ones or shortening earlier ones."""
    if len(overlays) < 2:
        return overlays

    overlays.sort(key=lambda o: o["timing"]["source_start"])

    for i in range(1, len(overlays)):
        prev_end = overlays[i - 1]["timing"]["source_end"]
        curr_start = overlays[i]["timing"]["source_start"]

        if curr_start < prev_end + MIN_OVERLAY_GAP:
            # Shorten the previous overlay
            new_end = curr_start - MIN_OVERLAY_GAP
            if new_end - overlays[i - 1]["timing"]["source_start"] >= 2.0:
                overlays[i - 1]["timing"]["source_end"] = round(new_end, 3)
                overlays[i - 1]["timing"]["display_duration"] = round(
                    new_end - overlays[i - 1]["timing"]["source_start"], 3
                )
            else:
                # Shift the current overlay later instead
                new_start = prev_end + MIN_OVERLAY_GAP
                shift = new_start - curr_start
                overlays[i]["timing"]["source_start"] = round(new_start, 3)
                overlays[i]["timing"]["source_end"] = round(
                    overlays[i]["timing"]["source_end"] + shift, 3
                )

    return overlays


def generate_overlay_spec(mentions_data, edl):
    """Generate overlay specification from mentions and EDL.

    Args:
        mentions_data: Parsed mentions JSON (with assets)
        edl: Parsed EDL JSON

    Returns:
        Overlay spec dict
    """
    mentions = mentions_data.get("mentions", [])
    edl_segments = edl.get("segments", [])
    source_file = mentions_data.get("source_file", "")
    stem = mentions_data.get("stem", "")

    overlays = []
    for mention in mentions:
        timing = compute_display_timing(mention, edl_segments)
        position = DEFAULT_POSITIONS.get(mention["type"], "lower-right")

        overlay = {
            "id": mention["id"],
            "type": mention["type"],
            "label": mention["label"],
            "timing": timing,
            "position": position,
        }

        if mention.get("url"):
            overlay["url"] = mention["url"]
        if mention.get("asset_path"):
            overlay["asset_path"] = mention["asset_path"]
        if mention.get("metadata"):
            overlay["metadata"] = mention["metadata"]

        overlays.append(overlay)

    # Resolve overlapping overlays
    overlays = resolve_overlaps(overlays)

    return {
        "source_file": source_file,
        "stem": stem,
        "overlays": overlays,
    }


# ── CLI ──────────────────────────────────────────────────────────────────────

if __name__ == "__main__":
    if len(sys.argv) < 3:
        print("Usage: uv run generate_overlay_spec.py <mentions_json> <edl_json> [--output-dir <dir>]")
        sys.exit(1)

    mentions_path = sys.argv[1]
    edl_path = sys.argv[2]
    output_dir = None

    args = sys.argv[3:]
    i = 0
    while i < len(args):
        if args[i] == "--output-dir" and i + 1 < len(args):
            output_dir = args[i + 1]
            i += 2
        else:
            i += 1

    with open(mentions_path) as f:
        mentions_data = json.load(f)
    with open(edl_path) as f:
        edl = json.load(f)

    stem = mentions_data.get("stem", "unknown")

    if output_dir is None:
        output_dir = os.path.dirname(os.path.abspath(mentions_path))
    os.makedirs(output_dir, exist_ok=True)

    print(f"Generating overlay spec for: {stem}")
    spec = generate_overlay_spec(mentions_data, edl)
    print(f"  {len(spec['overlays'])} overlays:")

    for o in spec["overlays"]:
        t = o["timing"]
        has_asset = "asset" if o.get("asset_path") else "text-only"
        print(f"  [{t['source_start']:.1f}–{t['source_end']:.1f}s] {o['type']}: {o['label']} ({has_asset})")

    output_path = os.path.join(output_dir, f"{stem}_overlays.json")
    with open(output_path, "w") as f:
        json.dump(spec, f, indent=2)
    print(f"\nOverlay spec saved to {output_path}")