backend

Dockerfile

@@ -0,0 +1,25 @@
+# Backend image for Railway (or any Docker host).
+FROM python:3.12-slim
+
+# ffmpeg/ffprobe are required at runtime (frame sampling, transcode, audio for whisper).
+RUN apt-get update \
+    && apt-get install -y --no-install-recommends ffmpeg \
+    && rm -rf /var/lib/apt/lists/*
+
+# uv for fast, reproducible dependency installs.
+COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
+
+WORKDIR /app
+
+# Install dependencies first for better layer caching. Use the committed lock for a
+# reproducible build (`requests` etc. come in transitively and are pinned there).
+COPY pyproject.toml uv.lock ./
+RUN uv sync --frozen --no-dev
+
+# Application code.
+COPY app ./app
+
+# Railway injects $PORT; default to 8000 for local `docker run`.
+ENV PORT=8000
+EXPOSE 8000
+CMD uv run uvicorn app.main:app --host 0.0.0.0 --port ${PORT}

README.md

@@ -1,9 +1,10 @@
 ---
 title: Creator Vision Demo
-emoji: 🏆
+emoji: 🎥
+sdk: docker
+app_port: 8000
 colorFrom: gray
 colorTo: red
-sdk: docker
 pinned: false
 license: mit
 short_description: analyze brand presence with a suite of ai tools

app/analysis/__init__.py

@@ -0,0 +1,20 @@
+"""Analysis pipeline (multi-product; several detection modes).
+
+Each analysis tracks N products. A mode's `DetectionStrategy` produces a per-product box
+track; mentions (caption/audio/OCR) are counted globally against the shared keyword list.
+Modes are registered in `registry.py`; the similarity backends (DINOv2/v3, CLIP, OpenAI)
+live in `app/similarity`. See DesignDoc.md for the result schema.
+"""
+
+from .registry import BuildOpts, owlv2_needs_images, requires_name, requires_reference
+from .runner import run_analysis
+from .strategies import ProductInput
+
+__all__ = [
+    "run_analysis",
+    "ProductInput",
+    "requires_reference",
+    "requires_name",
+    "owlv2_needs_images",
+    "BuildOpts",
+]

app/analysis/config.py

@@ -0,0 +1,79 @@
+"""Pipeline tunables — fps, fal endpoints, per-mode thresholds, concurrency, and flags.
+
+Model-intrinsic DINO config (the v2/v3 repos, DINO_THRESHOLD, DINO_MULTI_REF) lives with the
+model in ``app/similarity/dino.py``. Everything here is about *how the pipeline samples,
+segments, crops, and gates*.
+"""
+
+import os
+
+# --- sampling fps ---
+VIDEO_FPS = 3  # fps the video is downsampled to for the SAM-text tracking modes
+FRAME_FPS = 4  # fps frames are sampled at for the per-frame segment+match modes
+
+MAX_WIDTH = 640  # frames/video are scaled to this width before upload
+DETECTION_THRESHOLD = 0.3  # SAM 3 text detection threshold
+GAP_MERGE_FRAMES = 1  # bridge appearance runs separated by ≤1 missing frame
+
+# --- fal concurrency ---
+# fal account's max concurrent inference requests. Upgrade fal -> increase concurrency
+# -> hamemr down latency. Should get 429's if this is increased, but we don't, we just 
+# notice it gets throttled.
+FAL_CONCURRENCY = 10
+
+# --- fal endpoints ---
+SAM3_ENDPOINT = "fal-ai/sam-3/video-rle"
+SAM3_1_ENDPOINT = "fal-ai/sam-3-1/video-rle"  # drop-in, faster + better (Object Multiplex)
+AUTOSEG_ENDPOINT = "fal-ai/sam2/auto-segment"  # SAM 2 segment-everything
+EVF_SAM_ENDPOINT = "fal-ai/evf-sam"  # text-grounded segmentation (Grounding DINO backbone)
+
+# --- segment-per-frame modes (sam2_dino, sam3_clip): one auto-segment call per frame ---
+SEGMENT_FRAME_WORKERS = FAL_CONCURRENCY  # concurrent auto-segment inference calls (≤ fal limit)
+SEGMENT_MASK_WORKERS = 16  # mask downloads in flight per frame (fal CDN, NOT inference-limited)
+
+# --- sam3_clip (SAM 2 segment-everything + CLIP) ---
+# CLIP image-image cosines are high/weakly-calibrated, so keep the top-K most similar
+# segments per frame gated by a high absolute floor.
+CLIP_THRESHOLD = 0.85
+CLIP_TOPK = 1
+CLIP_MIN_AREA_FRAC = 0.0001  # ~ the old per-dimension 1% filter, as an area floor
+
+# --- sam2_dino (SAM 2 segment-everything + DINO) ---
+SAM2_DINO_THRESHOLD = 0.33  # default similarity threshold (the UI slider starts here)
+SAM2_DINO_CANDIDATE_FLOOR = 0.0  # keep best-per-frame above this so the slider can go below
+SAM2_DINO_TOPK = 2  # keep the best-matching segments per product per frame
+SAM2_MIN_AREA_FRAC = 0.004  # drop segments smaller than this fraction of the frame area
+
+# --- gdino_text_dino (EVF-SAM text segmentation + DINO) ---
+GDINO_DINO_THRESHOLD = 0.33  # default similarity threshold (the UI slider starts here)
+GDINO_CANDIDATE_FLOOR = 0.0  # keep detections above this so the slider can go below
+GDINO_MIN_AREA_FRAC = 0.004  # drop a segmentation smaller than this fraction of the frame
+# gdino runs in two stages: (1) network — EVF call + mask download per (frame, product), run
+# concurrently up to GDINO_FRAME_WORKERS; (2) CPU — DINO embeds the surviving crops in fat
+# batches. Throughput is maximized AT your fal concurrency limit: fewer underutilizes it, more
+# triggers 429 throttling (measured ~2.75× slower at 64 vs the limit). So pin it to the limit;
+# the `[gdino] ... s/call` log lets you confirm. (If you ever raise your fal tier, raise this.)
+GDINO_FRAME_WORKERS = FAL_CONCURRENCY  # concurrent EVF calls (one inference each) — ≤ fal limit
+GDINO_EMBED_CHUNK = 64  # crops per batched DINO ONNX pass in the CPU stage (CPU, not fal)
+
+# --- shared DINO crop input ---
+# True: mask the segment's background before embedding (object-only crop → fewer false
+# matches). False: embed the plain bounding-box crop. Applies to the modes with a segment
+# mask (sam2_dino, gdino_text_dino); only changes the embedding input, not the overlay.
+# (Paired with similarity.dino.DINO_MULTI_REF, which controls reference combination.)
+DINO_MASK_BG = False
+
+# --- scene-cut splitting ---
+SCENE_CUT_DIST = 0.45  # L1 distance between normalized 16-bin/channel color histograms
+
+# --- owlv2 (hosted HF Inference Endpoint; see owlv2_endpoint/) ---
+OWLV2_ENDPOINT_URL = os.getenv("OWLV2_ENDPOINT_URL", "")  # required for the owlv2 mode
+OWLV2_FRAME_WORKERS = 4  # concurrent endpoint calls (≤ the endpoint's serving concurrency)
+OWLV2_FRAME_BATCH = 6  # frames per endpoint request — one GPU forward pass serves the batch.
+# Bounded by endpoint GPU memory (B×960×960 ViT) and the 120s request timeout; raise if the
+# endpoint has headroom, lower if you see OOM/timeouts.
+OWLV2_EMBED_CHUNK = 64  # crops per batched DINO ONNX pass in the optional DINO-on-top stage
+OWLV2_SCORE_FLOOR = 0.05  # keep detections with confidence ≥ this; UI sliders re-threshold up
+OWL_CONF_THRESHOLD = 0.2  # default for the confidence slider
+OWL_OBJ_THRESHOLD = 0.2  # default for the objectness slider
+OWL_SIM_THRESHOLD = 0.33  # default for the (optional) DINO-on-top similarity slider

app/analysis/masks.py

@@ -0,0 +1,76 @@
+"""Segmentation-mask + crop utilities: download masks, derive bboxes, RLE-encode for the
+overlay, and produce the crops fed to the embedders."""
+
+import io
+import time
+
+import numpy as np
+import requests
+from PIL import Image
+
+from ..schemas import Box, Mask
+from .config import DINO_MASK_BG
+
+# Neutral fill = ImageNet mean ×255, so masked-out pixels normalize to ~0 (in-distribution
+# for DINO, unlike pure black). Embedding the object on this fill — not the raw bbox crop —
+# keeps background/neighboring objects from contaminating the similarity score.
+_DINO_BG = (124, 116, 104)
+
+
+def fetch_mask(mask_url: str, attempts: int = 3) -> Image.Image | None:
+    """Download an auto-segment mask PNG as a full-frame grayscale image. Tolerant of
+    transient network blips (DNS/connection resets on the fal CDN): retries with backoff,
+    and returns None on persistent failure so the caller skips that one segment instead of
+    failing the whole analysis."""
+    for i in range(attempts):
+        try:
+            resp = requests.get(mask_url, timeout=30)
+            resp.raise_for_status()
+            return Image.open(io.BytesIO(resp.content)).convert("L")
+        except Exception:  # noqa: BLE001 — transient network / decode error
+            if i == attempts - 1:
+                return None
+            time.sleep(0.4 * (i + 1))
+    return None
+
+
+def mask_bbox(mask_url: str) -> tuple[int, int, int, int] | None:
+    """Download a mask PNG and return its (left, upper, right, lower) bbox."""
+    m = fetch_mask(mask_url)
+    return m.getbbox() if m is not None else None
+
+
+def encode_rle(mask: Image.Image, target_w: int = 256) -> Mask:
+    """Downscale a full-frame binary mask and RLE-encode it row-major (alternating runs
+    starting with background). Self-contained — no fal URL leaves the backend."""
+    w, h = mask.size
+    tw = min(target_w, w)
+    th = max(1, round(h * tw / w))
+    flat = (
+        np.asarray(mask.resize((tw, th), Image.NEAREST)) > 127
+    ).astype(np.uint8).ravel()  # row-major
+    bounds = np.concatenate(([0], np.flatnonzero(np.diff(flat)) + 1, [flat.size]))
+    runs = np.diff(bounds).astype(int).tolist()
+    if flat.size and flat[0] == 1:
+        runs = [0, *runs]  # counts must start with a background run
+    return Mask(w=tw, h=th, counts=runs)
+
+
+def masked_crop(frame: Image.Image, mask: Image.Image, bbox: tuple) -> Image.Image:
+    """Composite the segmented object over a neutral fill, then crop to its bbox — so DINO
+    embeds the object alone, not whatever else shares the bounding rectangle. When
+    DINO_MASK_BG is False, returns the plain bbox crop (background kept)."""
+    if not DINO_MASK_BG:
+        return frame.crop(bbox)
+    if mask.size != frame.size:
+        mask = mask.resize(frame.size, Image.NEAREST)
+    bg = Image.new("RGB", frame.size, _DINO_BG)
+    return Image.composite(frame.convert("RGB"), bg, mask).crop(bbox)
+
+
+def crop_box(frame: Image.Image, box: Box) -> Image.Image:
+    """Crop a normalized [0,1] Box region out of a frame (for box-track scoring)."""
+    w, h = frame.size
+    left, top = box.x * w, box.y * h
+    right, bottom = (box.x + box.w) * w, (box.y + box.h) * h
+    return frame.crop((max(0, left), max(0, top), min(w, right), min(h, bottom)))

app/analysis/owlv2_client.py

@@ -0,0 +1,156 @@
+"""Thin client for the hosted OWLv2 Inference Endpoint (see owlv2_endpoint/handler.py).
+
+One call per frame returns per-product detections (box + confidence + objectness). The query
+payload per product is built once by the strategy and reused across frames.
+"""
+
+import base64
+import io
+import os
+import statistics
+import threading
+import time
+
+import requests
+from PIL import Image
+
+from .config import OWLV2_ENDPOINT_URL
+
+# Scale-from-zero cold starts can take a minute+; poll a 503 endpoint this often, up to
+# this total, before giving up — separate from the small transient-error attempt budget.
+COLD_START_POLL_S = 10.0
+COLD_START_BUDGET_S = 4.0 # 180.0
+
+# Per-frame timing fields collected into the Profiler. Client stages are measured here;
+# `server_*` come from the handler's `timing` dict (see owlv2_endpoint/handler.py).
+_PROFILE_FIELDS = (
+    "frame_encode_ms",  # client: JPEG+base64 the frame
+    "network_ms",       # client: POST round trip (handler compute + transfer + queue)
+    "server_total",     # handler: end-to-end inside __call__
+    "server_decode",    # handler: base64 -> PIL
+    "server_forward",   # handler: vision backbone + box/objectness heads
+    "server_queries",   # handler: per-product query embedding (cache miss = expensive)
+    "server_classify",  # handler: class_predictor + NMS + box decode
+    "overhead_ms",      # derived: network_ms - server_total (transfer + queue wait)
+    "dino_ms",          # strategy: DINO crop embedding for this frame (0 if no DINO)
+)
+
+
+class Profiler:
+    """Thread-safe accumulator of per-frame timing records, logged as one summary at the
+    end of a detect() run. Concurrent workers append; aggregation happens once at the end."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._records: list[dict] = []
+
+    def add(self, record: dict) -> None:
+        with self._lock:
+            self._records.append(record)
+
+    def log_summary(self, wall_s: float, n_products: int,
+                    extra: dict | None = None) -> None:
+        recs = self._records
+        if not recs:
+            return
+        lines = [
+            f"[owlv2] {len(recs)} endpoint-calls, {n_products} product(s), "
+            f"wall={wall_s:.1f}s ({len(recs) / wall_s:.1f} calls/s)",
+        ]
+        for k, v in (extra or {}).items():
+            lines.append(f"        {k}: {v}")
+        lines.append(f"        {'stage':<16}{'sum_s':>9}{'mean_ms':>9}"
+                     f"{'p50':>8}{'p95':>8}{'max':>8}")
+        for field in _PROFILE_FIELDS:
+            vals = [r[field] for r in recs if r.get(field) is not None]
+            if not vals:
+                continue
+            p = sorted(vals)
+            lines.append(
+                f"        {field:<16}{sum(vals) / 1000:>9.1f}{statistics.mean(vals):>9.1f}"
+                f"{p[len(p) // 2]:>8.1f}{p[min(len(p) - 1, int(len(p) * 0.95))]:>8.1f}"
+                f"{max(vals):>8.1f}"
+            )
+        hits = sum(r.get("server_cache_hits", 0) for r in recs)
+        lines.append(f"        query-cache hits: {hits}/{n_products * len(recs)}")
+        print("\n".join(lines))
+
+
+def image_b64(img: Image.Image, fmt: str = "JPEG", quality: int = 90) -> str:
+    buf = io.BytesIO()
+    img.convert("RGB").save(buf, format=fmt, quality=quality)
+    return base64.b64encode(buf.getvalue()).decode()
+
+
+def file_b64(path: str) -> str:
+    return image_b64(Image.open(path))
+
+
+def owlv2_detect_batch(
+    frames: list[Image.Image],
+    products_query: list[dict],
+    ref_type: str,
+    floor: float,
+    attempts: int = 3,
+) -> tuple[list[list[list[dict]]], dict]:
+    """POST a batch of frames + all products' queries in one request; return
+    (per_frame_detections, timing_record). per_frame_detections[f][p] = product p's list of
+    {box:[x,y,w,h] normalized, confidence, objectness} in frame f. One forward pass on the
+    endpoint serves the whole batch. timing_record holds this call's client + server stage
+    timings (see _PROFILE_FIELDS). Retries transient errors."""
+    if not OWLV2_ENDPOINT_URL:
+        raise RuntimeError("OWLV2_ENDPOINT_URL is not set (see owlv2_endpoint/README.md).")
+    token = os.getenv("HF_TOKEN", "")
+    headers = {"Content-Type": "application/json"}
+    if token:
+        headers["Authorization"] = f"Bearer {token}"
+    t_enc = time.perf_counter()
+    payload = {
+        "inputs": {
+            "images": [image_b64(f) for f in frames],
+            "ref_type": ref_type,
+            "products": products_query,
+            "score_floor": floor,
+        }
+    }
+    encode_ms = (time.perf_counter() - t_enc) * 1000
+    last: Exception | None = None
+    cold_waited = 0.0
+    i = 0
+    while i < attempts:
+        try:
+            t_net = time.perf_counter()
+            resp = requests.post(OWLV2_ENDPOINT_URL, headers=headers, json=payload, timeout=120)
+            if resp.status_code == 503 and cold_waited < COLD_START_BUDGET_S:
+                # Endpoint scaled to zero / still booting. A scale-from-zero cold start
+                # takes far longer than the generic retry backoff, so wait it out without
+                # spending an attempt — otherwise every run that hits a cold endpoint fails.
+                time.sleep(COLD_START_POLL_S)
+                cold_waited += COLD_START_POLL_S
+                continue
+            if not resp.ok:
+                # HF surfaces handler exceptions in the body; include it so the real
+                # cause (not just "400 Bad Request") reaches the caller.
+                raise RuntimeError(f"{resp.status_code} {resp.reason}: {resp.text[:2000]}")
+            network_ms = (time.perf_counter() - t_net) * 1000
+            body = resp.json()
+            srv = body.get("timing") or {}
+            record = {
+                "frame_encode_ms": round(encode_ms, 2),
+                "network_ms": round(network_ms, 2),
+                "server_total": srv.get("total"),
+                "server_decode": srv.get("decode"),
+                "server_forward": srv.get("forward"),
+                "server_queries": srv.get("queries"),
+                "server_classify": srv.get("classify"),
+                "server_cache_hits": srv.get("query_cache_hits", 0),
+                "frames_in_batch": len(frames),
+                "overhead_ms": round(network_ms - srv["total"], 2) if "total" in srv else None,
+            }
+            return body.get("frames", []), record
+        except Exception as exc:  # noqa: BLE001 — transient endpoint error
+            last = exc
+            i += 1
+            if i < attempts:
+                time.sleep(1.0 * i)  # short backoff for genuinely transient failures
+    raise RuntimeError(f"OWLv2 endpoint call failed: {last}")

app/analysis/registry.py

@@ -0,0 +1,125 @@
+"""The single source of truth mapping each DetectionMode to how it's built and validated.
+
+Add or change a mode here by composing existing strategies + similarity backends — the
+runner and `main.py`'s request validation both read from this table.
+"""
+
+from collections.abc import Callable
+from dataclasses import dataclass
+
+from ..schemas import DetectionMode, ScoreSpec
+from ..similarity import ClipEmbedder, DinoEmbedder
+from .config import (
+    CLIP_MIN_AREA_FRAC,
+    CLIP_THRESHOLD,
+    CLIP_TOPK,
+    GDINO_DINO_THRESHOLD,
+    SAM2_DINO_CANDIDATE_FLOOR,
+    SAM2_DINO_THRESHOLD,
+    SAM2_DINO_TOPK,
+    SAM2_MIN_AREA_FRAC,
+    SAM3_1_ENDPOINT,
+    SAM3_ENDPOINT,
+)
+from .strategies import (
+    DetectionStrategy,
+    DinoRunRefiner,
+    GroundedStrategy,
+    OpenAiRefiner,
+    OwlV2Strategy,
+    SamTextStrategy,
+    SegmentMatchStrategy,
+)
+
+
+def _sim(default: float) -> list[ScoreSpec]:
+    """The single 'similarity' slider used by the DINO/CLIP match modes."""
+    return [ScoreSpec(key="similarity", label="Similarity", default=default)]
+
+
+@dataclass(frozen=True)
+class BuildOpts:
+    """Per-request knobs threaded into a strategy's construction."""
+
+    dino_variant: str = "v2"  # DINO backbone for the *_dino modes
+    owl_ref_type: str = "text"  # OWLv2 reference type: text | image | both
+    owl_dino: str = "none"  # OWLv2 DINO-on-top: none | v2 | v3
+
+
+@dataclass(frozen=True)
+class ModeSpec:
+    # build(opts) -> strategy. None for the debug "none" mode (handled in runner).
+    build: Callable[["BuildOpts"], DetectionStrategy | None]
+    requires_name: bool = True
+    requires_reference: bool = False
+
+
+REGISTRY: dict[DetectionMode, ModeSpec] = {
+    DetectionMode.sam3_text: ModeSpec(
+        lambda o: SamTextStrategy(SAM3_ENDPOINT),
+    ),
+    DetectionMode.sam3_text_dino: ModeSpec(
+        lambda o: SamTextStrategy(SAM3_ENDPOINT, DinoRunRefiner(DinoEmbedder(o.dino_variant))),
+        requires_reference=True,
+    ),
+    DetectionMode.sam3_1_text_dino: ModeSpec(
+        lambda o: SamTextStrategy(SAM3_1_ENDPOINT, DinoRunRefiner(DinoEmbedder(o.dino_variant))),
+        requires_reference=True,
+    ),
+    DetectionMode.sam3_text_openai: ModeSpec(
+        lambda o: SamTextStrategy(SAM3_ENDPOINT, OpenAiRefiner()),
+    ),
+    DetectionMode.sam3_clip: ModeSpec(
+        lambda o: SegmentMatchStrategy(
+            ClipEmbedder(), threshold=CLIP_THRESHOLD, topk=CLIP_TOPK,
+            min_area_frac=CLIP_MIN_AREA_FRAC, mask_overlay=False, smooth=False,
+        ),
+        requires_name=False,  # reference-only; name is just a label
+        requires_reference=True,
+    ),
+    DetectionMode.sam2_dino: ModeSpec(
+        lambda o: SegmentMatchStrategy(
+            DinoEmbedder(o.dino_variant), threshold=SAM2_DINO_CANDIDATE_FLOOR,
+            topk=SAM2_DINO_TOPK, min_area_frac=SAM2_MIN_AREA_FRAC,
+            mask_overlay=True, smooth=True, score_specs=_sim(SAM2_DINO_THRESHOLD),
+        ),
+        requires_reference=True,
+    ),
+    DetectionMode.gdino_text_dino: ModeSpec(
+        lambda o: GroundedStrategy(DinoEmbedder(o.dino_variant), _sim(GDINO_DINO_THRESHOLD)),
+        requires_reference=True,
+    ),
+    DetectionMode.owlv2: ModeSpec(
+        lambda o: OwlV2Strategy(
+            o.owl_ref_type,
+            DinoEmbedder(o.owl_dino) if o.owl_dino != "none" else None,
+        ),
+        requires_name=True,  # the product name labels it + is the text query (text/both)
+        requires_reference=False,  # image refs only when image/both/DINO-on-top (owlv2_needs_images)
+    ),
+    DetectionMode.none: ModeSpec(lambda o: None, requires_name=False),  # debug, runner-handled
+}
+
+
+def spec_for(mode: DetectionMode) -> ModeSpec:
+    return REGISTRY[mode]
+
+
+def strategy_for(mode: DetectionMode, opts: "BuildOpts") -> DetectionStrategy | None:
+    return REGISTRY[mode].build(opts)
+
+
+def requires_reference(mode: DetectionMode) -> bool:
+    return REGISTRY[mode].requires_reference
+
+
+def requires_name(mode: DetectionMode) -> bool:
+    return REGISTRY[mode].requires_name
+
+
+def owlv2_needs_images(mode: DetectionMode, opts: BuildOpts) -> bool:
+    """OWLv2 needs reference images when matching by image (image/both ref types) or when
+    DINO-on-top is enabled (it scores crops vs an image reference)."""
+    return mode == DetectionMode.owlv2 and (
+        opts.owl_ref_type in ("image", "both") or opts.owl_dino != "none"
+    )

app/analysis/runner.py

@@ -0,0 +1,157 @@
+"""Background entrypoint: validate inputs, run the mode's strategy, assemble the result.
+
+The whole pipeline is: validate → strategy.detect (per-product boxes) → build_product_result
+per product (+ scene-cut split / cut-frame export) → enrich mentions → store.
+"""
+
+from ..jobs import store
+from ..mentions import (
+    count_keywords,
+    count_keywords_breakdown,
+    count_ocr_mentions,
+    parse_keywords,
+    transcribe,
+)
+from ..schemas import AnalysisResult, Box, DetectionMode, JobStatus, ProductResult
+from .registry import BuildOpts, owlv2_needs_images, requires_name, requires_reference, strategy_for
+from .strategies import ProductInput
+from .tracks import build_product_result, detect_cut_frames
+from .video import probe_duration
+
+
+def _validate(mode: DetectionMode, products: list[ProductInput], opts: BuildOpts) -> None:
+    if not products:
+        raise ValueError("At least one product is required.")
+    if requires_name(mode) and any(not p.name.strip() for p in products):
+        raise ValueError(f"{mode.value} needs a name for every product.")
+    needs_ref = requires_reference(mode) or owlv2_needs_images(mode, opts)
+    if needs_ref and any(not p.exemplar_paths for p in products):
+        raise ValueError(f"{mode.value} needs a reference image for every product.")
+
+
+def run_analysis(
+    job_id: str,
+    video_path: str,
+    products: list[ProductInput],
+    caption: str,
+    mention_keywords: str,
+    mode: DetectionMode,
+    split_on_cut: bool = False,
+    dino_variant: str = "v2",
+    owl_ref_type: str = "text",
+    owl_dino: str = "none",
+) -> None:
+    """Background entrypoint. Updates the job store in place."""
+    store.set_status(job_id, JobStatus.running)
+
+    # --- DEBUG: no-op mode → dummy result (delete this block to remove) ------ #
+    if mode == DetectionMode.none:
+        store.set_done(job_id, _dummy_result())
+        return
+    # ------------------------------------------------------------------------- #
+
+    try:
+        opts = BuildOpts(
+            dino_variant=dino_variant, owl_ref_type=owl_ref_type, owl_dino=owl_dino
+        )
+        _validate(mode, products, opts)
+        strategy = strategy_for(mode, opts)
+        duration = probe_duration(video_path)
+        fps = strategy.fps
+
+        per_product = strategy.detect(video_path, products)  # {name: {frame: [Box]}}
+
+        # Slider modes re-filter/recount in the UI, so they own the scene-cut split too:
+        # hand the UI the cut boundaries and let it split at any threshold.
+        specs = strategy.score_specs
+        is_slider = bool(specs)
+        cut_frames: list[int] = []
+        if split_on_cut and is_slider:
+            union = sorted({int(k) for boxes in per_product.values() for k in boxes})
+            cut_frames = detect_cut_frames(video_path, fps, union)
+
+        result = AnalysisResult(
+            fps=float(fps),
+            duration_sec=duration,
+            mode=mode,
+            products=[
+                build_product_result(
+                    p.name, per_product.get(p.name, {}), fps, duration, video_path,
+                    split_on_cut and not is_slider,  # backend split only for non-slider modes
+                )
+                for p in products
+            ],
+            brand_name=", ".join(p.name for p in products) or None,
+            score_specs=specs,
+            cut_frames=cut_frames,
+        )
+        _enrich_mentions(
+            result, video_path, caption, [p.name for p in products], mention_keywords
+        )
+        store.set_done(job_id, result)
+    except Exception as exc:  # noqa: BLE001 — surface any failure to the client
+        store.set_error(job_id, f"{type(exc).__name__}: {exc}")
+
+
+def _enrich_mentions(
+    result: AnalysisResult,
+    video_path: str,
+    caption: str,
+    product_names: list[str],
+    mention_keywords: str,
+) -> None:
+    """Phase 4: transcript + mention counts, driven by the dedicated keyword list (falling
+    back to the product names). Each step degrades to None on failure so a flaky model never
+    loses the visual analysis."""
+    keywords = parse_keywords(mention_keywords) or [
+        n.strip() for n in product_names if n.strip()
+    ]
+    if keywords:
+        result.caption_mentions = count_keywords(caption, keywords)
+        result.caption_mention_counts = count_keywords_breakdown(caption, keywords)
+    try:
+        result.transcript = transcribe(video_path)
+        if keywords:
+            result.audio_mentions = count_keywords(result.transcript, keywords)
+            result.audio_mention_counts = count_keywords_breakdown(
+                result.transcript, keywords
+            )
+    except Exception:  # noqa: BLE001
+        pass
+    if keywords:
+        try:
+            result.ocr_mentions, result.ocr_mention_counts = count_ocr_mentions(
+                video_path, keywords
+            )
+        except Exception:  # noqa: BLE001
+            pass
+
+
+# --- DEBUG: dummy result for the no-analysis "none" mode (delete to remove) -- #
+def _dummy_result() -> AnalysisResult:
+    """Fabricated result so the UI can be exercised with zero fal/ML calls."""
+    from .config import VIDEO_FPS
+
+    fps, duration = float(VIDEO_FPS), 10.0
+
+    def track(name: str, frames, conf: float) -> ProductResult:
+        boxes = {
+            str(f): [Box(x=0.2 + 0.01 * (f % 10), y=0.3, w=0.25, h=0.3,
+                         scores={"similarity": conf})]
+            for f in frames
+        }
+        return build_product_result(name, boxes, fps, duration)
+
+    return AnalysisResult(
+        fps=fps, duration_sec=duration, mode=DetectionMode.none,
+        products=[
+            track("debug product A", range(0, 7), 0.92),
+            track("debug product B", list(range(12, 19)) + list(range(24, 28)), 0.74),
+        ],
+        brand_name="debug product A, debug product B",
+        transcript="(debug) no transcription was run.",
+        audio_mentions=3, caption_mentions=1, ocr_mentions=2,
+        audio_mention_counts={"debug": 3},
+        caption_mention_counts={"debug": 1},
+        ocr_mention_counts={"debug": 2},
+    )

app/analysis/strategies.py

@@ -0,0 +1,490 @@
+"""Detection strategies — each produces per-product per-frame boxes for a video.
+
+Every mode is one of four shapes, composed with an injected similarity backend:
+  - SamTextStrategy:     SAM 3/3.1 text video-track per product, optional appearance refiner.
+  - SegmentMatchStrategy: SAM 2 segment-everything per frame, match each crop with an Embedder.
+  - GroundedStrategy:    EVF-SAM (Grounding DINO) text-segment per product, score with an Embedder.
+  - (the debug "none" mode is handled in the runner, not here)
+"""
+
+import tempfile
+import time
+from abc import ABC, abstractmethod
+from concurrent.futures import ThreadPoolExecutor
+from dataclasses import dataclass, field
+from pathlib import Path
+
+import fal_client
+import numpy as np
+from PIL import Image
+
+from ..schemas import Box, ScoreSpec
+from ..similarity import DINO_THRESHOLD, Embedder, OpenAiVerifier
+from .config import (
+    AUTOSEG_ENDPOINT,
+    DETECTION_THRESHOLD,
+    EVF_SAM_ENDPOINT,
+    FRAME_FPS,
+    GDINO_CANDIDATE_FLOOR,
+    GDINO_EMBED_CHUNK,
+    GDINO_FRAME_WORKERS,
+    GDINO_MIN_AREA_FRAC,
+    OWL_CONF_THRESHOLD,
+    OWL_OBJ_THRESHOLD,
+    OWL_SIM_THRESHOLD,
+    OWLV2_EMBED_CHUNK,
+    OWLV2_FRAME_BATCH,
+    OWLV2_FRAME_WORKERS,
+    OWLV2_SCORE_FLOOR,
+    SEGMENT_FRAME_WORKERS,
+    SEGMENT_MASK_WORKERS,
+    VIDEO_FPS,
+)
+from .masks import crop_box, encode_rle, fetch_mask, masked_crop
+from .owlv2_client import Profiler, file_b64, owlv2_detect_batch
+from .tracks import FrameBoxes, fill_micro_gaps, frame_runs
+from .video import downsample_video, extract_frame, sample_frames
+
+
+@dataclass
+class ProductInput:
+    """One product to track: a SAM text prompt + its reference image paths."""
+
+    name: str
+    exemplar_paths: list[str] = field(default_factory=list)
+
+
+class DetectionStrategy(ABC):
+    """Produces ``{product_name: {frame_index: [Box]}}`` for a video.
+
+    `score_specs` declares which named scores this strategy's boxes carry that should get a
+    post-hoc UI slider (empty = none). The runner copies it into the result."""
+
+    fps: float = VIDEO_FPS
+    score_specs: list[ScoreSpec] = []
+
+    @abstractmethod
+    def detect(
+        self, video_path: str, products: list[ProductInput]
+    ) -> dict[str, FrameBoxes]: ...
+
+
+# --------------------------------------------------------------------------- #
+# SAM 3 / 3.1 text tracking (+ optional per-appearance refiner)
+# --------------------------------------------------------------------------- #
+class AppearanceRefiner(ABC):
+    """Re-scores/filters a per-product box track after SAM-text detection."""
+
+    @abstractmethod
+    def refine(
+        self, video_path: str, fps: float, boxes: FrameBoxes, product: ProductInput
+    ) -> FrameBoxes: ...
+
+
+class DinoRunRefiner(AppearanceRefiner):
+    """Score *every* frame of each appearance with DINO; keep runs whose mean clears the
+    threshold, annotating each kept box with its per-frame `similarity` score (the
+    per-appearance average is filled later by build_product_result)."""
+
+    def __init__(self, embedder: Embedder):
+        self.embedder = embedder
+
+    def refine(self, video_path, fps, boxes, product):
+        ref = self.embedder.reference(product.exemplar_paths)
+        kept: FrameBoxes = {}
+        for a, b in frame_runs([int(k) for k in boxes]):
+            candidates = [f for f in range(a, b + 1) if str(f) in boxes]
+            scores = {
+                f: self.embedder.score(
+                    ref, [crop_box(extract_frame(video_path, f / fps), boxes[str(f)][0])]
+                )[0]
+                for f in candidates
+            }
+            if float(np.mean(list(scores.values()))) < DINO_THRESHOLD:
+                continue
+            for f in candidates:
+                kept[str(f)] = [
+                    boxes[str(f)][0].model_copy(
+                        update={"scores": {"similarity": round(scores[f], 3)}}
+                    )
+                ]
+        return kept
+
+
+class OpenAiRefiner(AppearanceRefiner):
+    """Keep only appearances whose representative frame passes the OpenAI yes/no check."""
+
+    def refine(self, video_path, fps, boxes, product):
+        verifier = OpenAiVerifier(product.name)
+        kept: FrameBoxes = {}
+        for a, b in frame_runs([int(k) for k in boxes]):
+            candidates = [f for f in range(a, b + 1) if str(f) in boxes]
+            mid = (a + b) // 2
+            rep = min(candidates, key=lambda f: abs(f - mid))
+            if verifier.verify(extract_frame(video_path, rep / fps), boxes[str(rep)][0]):
+                for f in candidates:
+                    kept[str(f)] = boxes[str(f)]
+        return kept
+
+
+class SamTextStrategy(DetectionStrategy):
+    fps = VIDEO_FPS
+
+    def __init__(self, endpoint: str, refiner: AppearanceRefiner | None = None):
+        self.endpoint = endpoint
+        self.refiner = refiner
+
+    def detect(self, video_path, products):
+        out: dict[str, FrameBoxes] = {}
+        for p in products:
+            boxes = self._text_boxes(video_path, p.name)
+            if self.refiner is not None:
+                boxes = self.refiner.refine(video_path, self.fps, boxes, p)
+            out[p.name] = boxes
+        return out
+
+    def _text_boxes(self, video_path: str, prompt: str) -> FrameBoxes:
+        with tempfile.TemporaryDirectory() as tmp:
+            small = downsample_video(video_path, VIDEO_FPS, Path(tmp))
+            url = fal_client.upload_file(small)
+            result = fal_client.subscribe(
+                self.endpoint,
+                arguments={"video_url": url, "prompt": prompt,
+                           "detection_threshold": DETECTION_THRESHOLD},
+                with_logs=False,
+            )
+        boxes: FrameBoxes = {}
+        for m in result.get("metadata") or []:
+            box = m.get("box")
+            if box:
+                s = m.get("score")
+                boxes[str(int(m["index"]))] = [
+                    Box(x=box[0], y=box[1], w=box[2], h=box[3],
+                        scores={"detection": s} if s is not None else {})
+                ]
+        return boxes
+
+
+# --------------------------------------------------------------------------- #
+# SAM 2 segment-everything + embedder match (sam2_dino, sam3_clip)
+# --------------------------------------------------------------------------- #
+class SegmentMatchStrategy(DetectionStrategy):
+    fps = FRAME_FPS
+
+    def __init__(
+        self,
+        embedder: Embedder,
+        *,
+        threshold: float,
+        topk: int,
+        min_area_frac: float,
+        mask_overlay: bool,
+        smooth: bool,
+        score_specs: list[ScoreSpec] = [],
+    ):
+        self.embedder = embedder
+        self.threshold = threshold
+        self.topk = topk
+        self.min_area_frac = min_area_frac
+        self.mask_overlay = mask_overlay
+        self.smooth = smooth
+        self.score_specs = score_specs
+
+    def detect(self, video_path, products):
+        refs = {p.name: self.embedder.reference(p.exemplar_paths) for p in products}
+        out: dict[str, FrameBoxes] = {p.name: {} for p in products}
+        with tempfile.TemporaryDirectory() as tmp:
+            files = sample_frames(video_path, self.fps, Path(tmp))
+            with ThreadPoolExecutor(max_workers=SEGMENT_FRAME_WORKERS) as pool:
+                for n, frame_out in pool.map(
+                    lambda nf: self._frame(nf[0], nf[1], refs), enumerate(files)
+                ):
+                    for name, boxes in frame_out.items():
+                        out[name][str(n)] = boxes
+        if self.smooth:
+            for boxes in out.values():
+                fill_micro_gaps(boxes)
+        return out
+
+    def _frame(self, n, frame_path, refs):
+        frame = Image.open(frame_path).convert("RGB")
+        fw, fh = frame.size
+        seg = fal_client.subscribe(
+            AUTOSEG_ENDPOINT,
+            arguments={"image_url": fal_client.upload_file(str(frame_path))},
+            with_logs=False,
+        )
+        metas = seg.get("individual_masks") or []
+        with ThreadPoolExecutor(max_workers=SEGMENT_MASK_WORKERS) as pool:
+            survivors = [
+                s for s in pool.map(lambda mm: self._survivor(mm, frame, fw, fh), metas) if s
+            ]
+        if not survivors:
+            return n, {}
+
+        rects = [s[0] for s in survivors]
+        crops = [s[1] for s in survivors]
+        masks = [s[2] for s in survivors]
+        frame_out: FrameBoxes = {}
+        for name, ref in refs.items():
+            sims = self.embedder.score(ref, crops)  # best cosine over the product's refs
+            scored = sorted(
+                ((sims[i], rects[i], masks[i]) for i in range(len(rects))),
+                key=lambda t: t[0],
+                reverse=True,
+            )
+            matched = [
+                Box(
+                    x=left / fw, y=upper / fh,
+                    w=(right - left) / fw, h=(lower - upper) / fh,
+                    scores={"similarity": round(cos, 3)},
+                    mask=encode_rle(m) if self.mask_overlay else None,
+                )
+                for cos, (left, upper, right, lower), m in scored[: self.topk]
+                if cos >= self.threshold
+            ]
+            if matched:
+                frame_out[name] = matched
+        return n, frame_out
+
+    def _survivor(self, mask_meta, frame, fw, fh):
+        m = fetch_mask(mask_meta["url"])
+        if m is None or (bbox := m.getbbox()) is None:
+            return None
+        left, upper, right, lower = bbox
+        if (right - left) * (lower - upper) < self.min_area_frac * fw * fh:
+            return None
+        crop = masked_crop(frame, m, bbox) if self.mask_overlay else frame.crop(bbox)
+        return bbox, crop, m
+
+
+# --------------------------------------------------------------------------- #
+# EVF-SAM text-grounded segmentation + embedder score (gdino_text_dino)
+# --------------------------------------------------------------------------- #
+class GroundedStrategy(DetectionStrategy):
+    """Two-stage to keep DINO off the network threads:
+    (1) network — EVF + mask download per (frame, product), highly concurrent;
+    (2) CPU — embed the surviving crops in fat batches. This avoids running 100s of
+    multi-threaded ONNX inferences at once (which oversubscribes the CPU and gets *slower*
+    with more workers), and replaces batch-of-1 embeds with a few large ones."""
+
+    fps = FRAME_FPS
+
+    def __init__(self, embedder: Embedder, score_specs: list[ScoreSpec] = []):
+        self.embedder = embedder
+        self.score_specs = score_specs
+
+    def detect(self, video_path, products):
+        t0 = time.time()
+        refs = {p.name: self.embedder.reference(p.exemplar_paths) for p in products}
+        out: dict[str, FrameBoxes] = {p.name: {} for p in products}
+        with tempfile.TemporaryDirectory() as tmp:
+            files = sample_frames(video_path, self.fps, Path(tmp))
+            # Stage 1 (network): upload each frame once, then EVF + mask per (frame, product).
+            with ThreadPoolExecutor(max_workers=GDINO_FRAME_WORKERS) as pool:
+                urls = list(pool.map(lambda fp: fal_client.upload_file(str(fp)), files))
+                t_up = time.time()
+                tasks = [(n, p) for n in range(len(files)) for p in products]
+                candidates = [
+                    c for c in pool.map(
+                        lambda t: self._candidate(t[0], urls[t[0]], files[t[0]], t[1]), tasks
+                    ) if c is not None
+                ]
+            t_net = time.time()
+
+            # Stage 2 (CPU): batch-embed each product's crops, gate, build boxes.
+            for name, ref in refs.items():
+                cand = [c for c in candidates if c[0] == name]
+                if not cand:
+                    continue
+                sims = self._scores(ref, [c[3] for c in cand])
+                for (_, n, coords, _crop, mask), cos in zip(cand, sims):
+                    if cos < GDINO_CANDIDATE_FLOOR:  # UI slider re-thresholds above this
+                        continue
+                    x, y, w, h = coords
+                    out[name][str(n)] = [
+                        Box(x=x, y=y, w=w, h=h,
+                            scores={"similarity": round(cos, 3)}, mask=mask)
+                    ]
+        for boxes in out.values():
+            fill_micro_gaps(boxes)
+        # DEBUG timing (remove when tuned): splits upload vs EVF so we can see fal's ceiling.
+        n_evf = len(files) * len(products)
+        print(
+            f"[gdino] {len(files)} frames × {len(products)} products | "
+            f"{len(candidates)} candidates | upload {t_up - t0:.1f}s + "
+            f"evf {t_net - t_up:.1f}s ({n_evf} calls, {(t_net - t_up) / max(n_evf, 1):.2f}s/call) "
+            f"+ embed {time.time() - t_net:.1f}s | workers={GDINO_FRAME_WORKERS}",
+            flush=True,
+        )
+        return out
+
+    def _candidate(self, n, image_url, frame_path, product):
+        """EVF-segment one product in one frame → (name, n, coords, crop, rle) or None.
+        Network/I-O + cropping only; no embedding (that's the batched stage 2)."""
+        try:
+            seg = fal_client.subscribe(
+                EVF_SAM_ENDPOINT,
+                arguments={"image_url": image_url, "prompt": product.name,
+                           "use_grounding_dino": True, "mask_only": True},
+                with_logs=False,
+            )
+        except fal_client.FalClientHTTPError as exc:
+            # EVF-SAM 422s when the prompt matches nothing in the frame — normal "absent".
+            if exc.status_code == 422 and "No objects matching" in str(exc):
+                return None
+            raise
+        img = seg.get("image") or {}
+        if not img.get("url"):
+            return None
+        m = fetch_mask(img["url"])
+        if m is None or (bbox := m.getbbox()) is None:
+            return None
+        fw, fh = m.size  # full-frame mask shares the frame's dimensions
+        left, upper, right, lower = bbox
+        if (right - left) * (lower - upper) < GDINO_MIN_AREA_FRAC * fw * fh:
+            return None
+        frame = Image.open(frame_path).convert("RGB")
+        coords = (left / fw, upper / fh, (right - left) / fw, (lower - upper) / fh)
+        return product.name, n, coords, masked_crop(frame, m, bbox), encode_rle(m)
+
+    def _scores(self, ref, crops):
+        """Embed crops in chunks — a few fat ONNX passes instead of one tiny call each."""
+        sims: list[float] = []
+        for i in range(0, len(crops), GDINO_EMBED_CHUNK):
+            sims.extend(self.embedder.score(ref, crops[i : i + GDINO_EMBED_CHUNK]))
+        return sims
+
+
+# --------------------------------------------------------------------------- #
+# OWLv2 hosted detection (text / image / both refs) + optional DINO-on-top
+# --------------------------------------------------------------------------- #
+class OwlV2Strategy(DetectionStrategy):
+    """Per frame, one call to the hosted OWLv2 endpoint returns each product's best detection
+    with a `confidence` + `objectness`. With DINO-on-top, the crop is also scored vs the
+    product's image reference, adding a `similarity`. Each score gets its own slider."""
+
+    fps = FRAME_FPS
+
+    def __init__(self, ref_type: str, dino: Embedder | None):
+        self.ref_type = ref_type
+        self.dino = dino
+        self.score_specs = [
+            ScoreSpec(key="confidence", label="Confidence", default=OWL_CONF_THRESHOLD),
+            ScoreSpec(key="objectness", label="Objectness", default=OWL_OBJ_THRESHOLD),
+        ]
+        if dino is not None:
+            self.score_specs.append(
+                ScoreSpec(key="similarity", label="Similarity", default=OWL_SIM_THRESHOLD)
+            )
+
+    def _query(self, p: ProductInput) -> dict:
+        q: dict = {}
+        if self.ref_type in ("text", "both"):
+            q["texts"] = [p.name]
+        if self.ref_type in ("image", "both"):
+            q["images"] = [file_b64(path) for path in p.exemplar_paths]
+        return q
+
+    def detect(self, video_path, products):
+        # Two-stage, mirroring GroundedStrategy: (1) network — endpoint calls (each a batch of
+        # frames, one GPU forward pass) collecting best-per-product boxes (+ crops if DINO);
+        # (2) CPU — embed all crops in fat batches per product, not a batch-of-1 per frame.
+        queries = [self._query(p) for p in products]  # built once, reused per frame
+        dino_on = self.dino is not None
+        dino_refs = (
+            {p.name: self.dino.reference(p.exemplar_paths) for p in products}
+            if dino_on else {}
+        )
+        prof = Profiler()
+        out: dict[str, FrameBoxes] = {p.name: {} for p in products}
+        pending: list[tuple[str, Box, Image.Image]] = []  # (product, box, crop) for stage 2
+        wall0 = time.perf_counter()
+        with tempfile.TemporaryDirectory() as tmp:
+            t_sample = time.perf_counter()
+            files = sample_frames(video_path, self.fps, Path(tmp))
+            sample_s = time.perf_counter() - t_sample
+            batches = [
+                list(range(i, min(i + OWLV2_FRAME_BATCH, len(files))))
+                for i in range(0, len(files), OWLV2_FRAME_BATCH)
+            ]
+
+            def consume(batch_results):
+                for n, frame_out, crops in batch_results:
+                    for name, boxes in frame_out.items():
+                        out[name][str(n)] = boxes
+                    pending.extend(crops)
+
+            # Run the first batch alone to warm the endpoint's per-product query cache, so the
+            # concurrent fan-out that follows all hits it (no first-wave recompute race). The
+            # warm-up batch is real work — its detections are kept, nothing wasted.
+            if batches:
+                consume(self._batch(batches[0], files, products, queries, dino_on, prof))
+            if len(batches) > 1:
+                with ThreadPoolExecutor(max_workers=OWLV2_FRAME_WORKERS) as pool:
+                    for results in pool.map(
+                        lambda idxs: self._batch(idxs, files, products, queries, dino_on, prof),
+                        batches[1:],
+                    ):
+                        consume(results)
+
+        # Stage 2 (CPU): batch-embed each product's crops and write similarity into its boxes.
+        dino_s = 0.0
+        if dino_on:
+            t_dino = time.perf_counter()
+            self._score_dino(dino_refs, pending)
+            dino_s = time.perf_counter() - t_dino
+
+        for boxes in out.values():
+            fill_micro_gaps(boxes)
+        extra = {"ref_type": self.ref_type, "dino": dino_on, "frames": len(files),
+                 "batch_size": OWLV2_FRAME_BATCH, "frame_sampling_s": round(sample_s, 1),
+                 "workers": OWLV2_FRAME_WORKERS}
+        if dino_on:
+            extra["dino_batched_s"] = round(dino_s, 1)
+            extra["dino_crops"] = len(pending)
+        prof.log_summary(time.perf_counter() - wall0, len(products), extra=extra)
+        return out
+
+    def _batch(self, idxs, files, products, queries, dino_on, prof):
+        """Stage 1 (network): one endpoint call over a batch of frames. Returns a list of
+        (frame_index, frame_out, crops); crops are deferred to the batched stage 2 (see
+        detect). No embedding here."""
+        frames = [Image.open(files[i]).convert("RGB") for i in idxs]
+        per_frame_dets, record = owlv2_detect_batch(
+            frames, queries, self.ref_type, OWLV2_SCORE_FLOOR
+        )
+        prof.add(record)
+        results: list[tuple[int, FrameBoxes, list[tuple[str, Box, Image.Image]]]] = []
+        for n, frame, dets in zip(idxs, frames, per_frame_dets):
+            frame_out: FrameBoxes = {}
+            crops: list[tuple[str, Box, Image.Image]] = []
+            for p, prod_dets in zip(products, dets):
+                if not prod_dets:
+                    continue
+                best = max(prod_dets, key=lambda d: d["confidence"])  # one track per product
+                x, y, w, h = best["box"]
+                box = Box(x=x, y=y, w=w, h=h, scores={
+                    "confidence": round(float(best["confidence"]), 3),
+                    "objectness": round(float(best["objectness"]), 3),
+                })
+                frame_out[p.name] = [box]
+                if dino_on:
+                    crops.append((p.name, box, crop_box(frame, Box(x=x, y=y, w=w, h=h))))
+            results.append((n, frame_out, crops))
+        return results
+
+    def _score_dino(self, dino_refs, pending):
+        """Group crops by product, embed in fat chunks (a few large ONNX passes instead of
+        one tiny call per frame), and write the similarity into each box in place."""
+        by_product: dict[str, list[tuple[Box, Image.Image]]] = {}
+        for name, box, crop in pending:
+            by_product.setdefault(name, []).append((box, crop))
+        for name, items in by_product.items():
+            ref = dino_refs[name]
+            crops = [c for _, c in items]
+            sims: list[float] = []
+            for i in range(0, len(crops), OWLV2_EMBED_CHUNK):
+                sims.extend(self.dino.score(ref, crops[i : i + OWLV2_EMBED_CHUNK]))
+            for (box, _), cos in zip(items, sims):
+                box.scores["similarity"] = round(float(cos), 3)

app/analysis/tracks.py

@@ -0,0 +1,175 @@
+"""Track post-processing shared by every mode: group per-frame boxes into contiguous
+appearances (bridging micro-gaps, optionally splitting at scene cuts), compute per-appearance
+averages, and assemble the per-product ProductResult.
+"""
+
+import tempfile
+from pathlib import Path
+
+import numpy as np
+from PIL import Image
+
+from ..schemas import Box, ProductResult
+from .config import GAP_MERGE_FRAMES, SCENE_CUT_DIST
+from .video import extract_frame, sample_frames
+
+FrameBoxes = dict[str, list[Box]]
+
+
+# --- contiguous runs / appearances ------------------------------------------- #
+def frame_runs(frames: list[int]) -> list[tuple[int, int]]:
+    """Group present frame indices into contiguous (start,end) runs, bridging gaps."""
+    frames = sorted(set(frames))
+    runs: list[tuple[int, int]] = []
+    for f in frames:
+        if runs and f - runs[-1][1] <= GAP_MERGE_FRAMES + 1:
+            runs[-1] = (runs[-1][0], f)
+        else:
+            runs.append((f, f))
+    return runs
+
+
+def runs_to_appearances(
+    runs: list[tuple[int, int]], fps: float, duration: float
+) -> list[tuple[float, float]]:
+    """(start,end) seconds for each (first,last) frame-index run."""
+    out: list[tuple[float, float]] = []
+    for a, b in runs:
+        start = round(a / fps, 2)
+        end = round(min((b + 1) / fps, duration or (b + 1) / fps), 2)
+        out.append((start, end))
+    return out
+
+
+# --- scene-cut splitting ----------------------------------------------------- #
+def frame_hist(img: Image.Image) -> np.ndarray:
+    arr = np.asarray(img.convert("RGB").resize((64, 64), Image.BILINEAR), dtype=np.float32)
+    hist = np.concatenate(
+        [np.histogram(arr[..., c], bins=16, range=(0, 255))[0] for c in range(3)]
+    ).astype(np.float32)
+    return hist / (hist.sum() + 1e-8)
+
+
+def split_run_on_cuts(
+    present: list[int], fps: float, video_path: str, threshold: float = SCENE_CUT_DIST
+) -> list[tuple[int, int]]:
+    """Split one run of present frame indices wherever a hard cut sits between frames."""
+    runs: list[tuple[int, int]] = []
+    seg_start = prev = present[0]
+    prev_h = frame_hist(extract_frame(video_path, prev / fps))
+    for f in present[1:]:
+        h = frame_hist(extract_frame(video_path, f / fps))
+        if float(np.abs(h - prev_h).sum()) > threshold:
+            runs.append((seg_start, prev))
+            seg_start = f
+        prev, prev_h = f, h
+    runs.append((seg_start, prev))
+    return runs
+
+
+def split_runs_on_cuts(
+    boxes: FrameBoxes, fps: float, video_path: str
+) -> list[tuple[int, int]]:
+    """All frame-index runs, with each time-contiguous run further split at cuts."""
+    runs: list[tuple[int, int]] = []
+    for a, b in frame_runs([int(k) for k in boxes]):
+        present = [f for f in range(a, b + 1) if str(f) in boxes]
+        runs.extend(split_run_on_cuts(present, fps, video_path))
+    return runs
+
+
+def detect_cut_frames(video_path: str, fps: float, frames: list[int]) -> list[int]:
+    """Frame indices where a hard scene cut *begins* — threshold-independent (a property of
+    the pixels), so the UI can re-split appearances at these boundaries after the similarity
+    slider moves. Only adjacent, bridgeable frames are compared. One ffmpeg dump (frame n →
+    index n, matching the detection pass) keeps it cheap."""
+    frames = sorted(set(frames))
+    if len(frames) < 2:
+        return []
+    with tempfile.TemporaryDirectory() as tmp:
+        files = sample_frames(video_path, fps, Path(tmp))
+
+        def hist_at(idx: int) -> np.ndarray | None:
+            return frame_hist(Image.open(files[idx])) if 0 <= idx < len(files) else None
+
+        cuts: list[int] = []
+        prev, prev_h = frames[0], hist_at(frames[0])
+        for f in frames[1:]:
+            h = hist_at(f)
+            if (
+                prev_h is not None and h is not None
+                and f - prev <= GAP_MERGE_FRAMES + 1
+                and float(np.abs(h - prev_h).sum()) > SCENE_CUT_DIST
+            ):
+                cuts.append(f)
+            prev, prev_h = f, h
+        return cuts
+
+
+# --- per-appearance averages / gap fill -------------------------------------- #
+def set_avg_scores(boxes: FrameBoxes, runs: list[tuple[int, int]]) -> None:
+    """For each run, write every box's `avg_scores[key]` = mean of that key's per-frame score
+    over the run — one pass, every score key. Run-scoped so scene-cut-split appearances get
+    independent averages; the UI recomputes the same way as sliders move."""
+    for a, b in runs:
+        frames = [f for f in range(a, b + 1) if str(f) in boxes]
+        keys = {k for f in frames for k in boxes[str(f)][0].scores}
+        for key in keys:
+            vals = [
+                boxes[str(f)][0].scores[key] for f in frames if key in boxes[str(f)][0].scores
+            ]
+            if not vals:
+                continue
+            avg = round(float(np.mean(vals)), 3)
+            for f in frames:
+                if key in boxes[str(f)][0].scores:
+                    boxes[str(f)][0].avg_scores[key] = avg
+
+
+def fill_micro_gaps(boxes: FrameBoxes, max_gap: int = 1) -> None:
+    """Fill single-frame (≤ max_gap) holes between two present detections by linearly
+    interpolating the box (and averaging each shared score), so a one-frame miss doesn't
+    flicker or split a run."""
+    present = sorted(int(k) for k in boxes)
+    for a, b in zip(present, present[1:]):
+        if not (1 <= b - a - 1 <= max_gap):
+            continue
+        ba, bb = boxes[str(a)][0], boxes[str(b)][0]
+        shared = ba.scores.keys() & bb.scores.keys()
+        for f in range(a + 1, b):
+            t = (f - a) / (b - a)
+            boxes[str(f)] = [
+                Box(
+                    x=ba.x + (bb.x - ba.x) * t,
+                    y=ba.y + (bb.y - ba.y) * t,
+                    w=ba.w + (bb.w - ba.w) * t,
+                    h=ba.h + (bb.h - ba.h) * t,
+                    scores={k: round((ba.scores[k] + bb.scores[k]) / 2, 3) for k in shared},
+                )
+            ]
+
+
+# --- assembly ---------------------------------------------------------------- #
+def build_product_result(
+    name: str,
+    boxes: FrameBoxes,
+    fps: float,
+    duration: float,
+    video_path: str | None = None,
+    split_on_cut: bool = False,
+) -> ProductResult:
+    """Assemble one product's track: group frames into appearances (optionally split at scene
+    cuts), set per-appearance average scores, and return the ProductResult."""
+    if split_on_cut and boxes and video_path:
+        runs = split_runs_on_cuts(boxes, fps, video_path)
+    else:
+        runs = frame_runs([int(k) for k in boxes])
+    set_avg_scores(boxes, runs)  # per-appearance averages, every score key
+    appearances = runs_to_appearances(runs, fps, duration)
+    return ProductResult(
+        name=name,
+        boxes=boxes,
+        appearances=appearances,
+        contiguous_appearances=len(appearances),
+        total_on_screen_sec=round(sum(e - s for s, e in appearances), 2),
+    )

app/analysis/video.py

@@ -0,0 +1,69 @@
+"""ffmpeg video I/O: probe duration, extract a single frame, and sample frames to disk."""
+
+import json
+import subprocess
+import tempfile
+from pathlib import Path
+
+from PIL import Image
+
+from .config import MAX_WIDTH
+
+
+def probe_duration(path: str) -> float:
+    try:
+        out = subprocess.run(
+            ["ffprobe", "-v", "error", "-show_entries", "format=duration",
+             "-of", "json", path],
+            capture_output=True, text=True, check=True,
+        )
+        return round(float(json.loads(out.stdout)["format"]["duration"]), 3)
+    except Exception:
+        return 0.0
+
+
+def sample_frames(video_path: str, fps: float, dest: Path, width: int = MAX_WIDTH) -> list[Path]:
+    """Dump frames at `fps` into `dest` (frame n → file index n) and return them sorted.
+    The single ffmpeg pass shared by every per-frame mode and by cut detection."""
+    subprocess.run(
+        ["ffmpeg", "-y", "-i", video_path, "-r", str(fps),
+         "-vf", f"scale={width}:-2", str(dest / "f%05d.jpg")],
+        check=True, capture_output=True,
+    )
+    return sorted(dest.glob("f*.jpg"))
+
+
+def downsample_video(video_path: str, fps: float, dest: Path, width: int = MAX_WIDTH) -> str:
+    """Downsample + scale to a small mp4 (no audio) for upload to a hosted video model."""
+    out = str(dest / "ds.mp4")
+    subprocess.run(
+        ["ffmpeg", "-y", "-i", video_path, "-r", str(fps),
+         "-vf", f"scale={width}:-2", "-an", out],
+        check=True, capture_output=True,
+    )
+    return out
+
+
+def extract_frame(video_path: str, t: float) -> Image.Image:
+    """Pull a single frame near time `t` (seconds) as a PIL image.
+
+    Robust to seeks that land at/just past EOF: sampled frame indices can round a hair
+    beyond the real duration, and ffmpeg input-seek past the end yields no frame and a
+    non-zero exit. So we retry a touch earlier, then fall back to the final frame."""
+    with tempfile.TemporaryDirectory() as tmp:
+        out = Path(tmp) / "f.jpg"
+
+        def grab(seek_args: list[str]) -> bool:
+            res = subprocess.run(
+                ["ffmpeg", "-y", *seek_args, "-frames:v", "1",
+                 "-vf", "scale=720:-2", str(out)],
+                capture_output=True,
+            )
+            return res.returncode == 0 and out.exists() and out.stat().st_size > 0
+
+        for ss in (t, max(0.0, t - 0.2), max(0.0, t - 0.5)):
+            if grab(["-ss", f"{ss:.3f}", "-i", video_path]):
+                return Image.open(out).copy()
+        if grab(["-sseof", "-0.2", "-i", video_path]):  # last frame, seeking from end
+            return Image.open(out).copy()
+        raise RuntimeError(f"ffmpeg could not extract a frame near t={t:.3f}s")

app/jobs.py

@@ -0,0 +1,55 @@
+"""In-memory job store.
+
+Single-process, single-user spike — a plain dict guarded by a lock is enough.
+Swap for Redis/DB if we ever need persistence or multiple workers.
+"""
+
+import threading
+import uuid
+from dataclasses import dataclass, field
+
+from .schemas import AnalysisResult, JobStatus
+
+
+@dataclass
+class Job:
+    id: str
+    status: JobStatus = JobStatus.queued
+    result: AnalysisResult | None = None
+    error: str | None = None
+
+
+class JobStore:
+    def __init__(self) -> None:
+        self._jobs: dict[str, Job] = {}
+        self._lock = threading.Lock()
+
+    def create(self) -> Job:
+        job = Job(id=uuid.uuid4().hex)
+        with self._lock:
+            self._jobs[job.id] = job
+        return job
+
+    def get(self, job_id: str) -> Job | None:
+        with self._lock:
+            return self._jobs.get(job_id)
+
+    def set_status(self, job_id: str, status: JobStatus) -> None:
+        with self._lock:
+            if job := self._jobs.get(job_id):
+                job.status = status
+
+    def set_done(self, job_id: str, result: AnalysisResult) -> None:
+        with self._lock:
+            if job := self._jobs.get(job_id):
+                job.status = JobStatus.done
+                job.result = result
+
+    def set_error(self, job_id: str, message: str) -> None:
+        with self._lock:
+            if job := self._jobs.get(job_id):
+                job.status = JobStatus.error
+                job.error = message
+
+
+store = JobStore()

app/main.py

@@ -0,0 +1,139 @@
+"""Creator Vision API.
+
+Phase 1: POST /analyze (BackgroundTasks + in-memory job store) and
+GET /status/{job_id}. The analysis is currently stubbed (see analysis.py);
+Phase 2 wires real SAM 3. See DesignDoc.md for the result schema.
+"""
+
+import json
+import os
+import shutil
+import uuid
+from pathlib import Path
+
+from dotenv import load_dotenv
+from fastapi import BackgroundTasks, FastAPI, File, Form, HTTPException, UploadFile
+from fastapi.middleware.cors import CORSMiddleware
+
+from .analysis import (
+    BuildOpts,
+    ProductInput,
+    owlv2_needs_images,
+    requires_name,
+    requires_reference,
+    run_analysis,
+)
+from .jobs import store
+from .schemas import DetectionMode, JobResponse, StatusResponse
+
+# Load backend/.env (and fall back to repo-root .env).
+load_dotenv()
+
+UPLOAD_DIR = Path(__file__).resolve().parent.parent / "uploads"
+UPLOAD_DIR.mkdir(exist_ok=True)
+
+app = FastAPI(title="Creator Vision API", version="0.1.0")
+
+# Allowed browser origins. Defaults to the local Next.js dev server; in prod set
+# ALLOWED_ORIGINS to your deployed frontend URL(s), comma-separated.
+_origins = os.getenv("ALLOWED_ORIGINS", "http://localhost:3000")
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=[o.strip() for o in _origins.split(",") if o.strip()],
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+@app.get("/health")
+def health() -> dict:
+    """Liveness + config sanity check (does the SAM3 key exist?)."""
+    return {
+        "status": "ok",
+        "fal_key_configured": bool(os.getenv("FAL_KEY")),
+    }
+
+
+def _save_upload(upload: UploadFile, dest_dir: Path) -> str:
+    """Persist an UploadFile under dest_dir with a collision-proof name."""
+    suffix = Path(upload.filename or "").suffix
+    dest = dest_dir / f"{uuid.uuid4().hex}{suffix}"
+    with dest.open("wb") as f:
+        shutil.copyfileobj(upload.file, f)
+    return str(dest)
+
+
+@app.post("/analyze", response_model=JobResponse)
+def analyze(
+    background_tasks: BackgroundTasks,
+    video: UploadFile = File(...),
+    # `products` is JSON: [{"name": str, "image_count": int}, ...]. `exemplars` is a
+    # flat file list in product order, sliced back per product by `image_count`.
+    products: str = Form(...),
+    exemplars: list[UploadFile] = File(default=[]),
+    caption: str = Form(default=""),
+    mention_keywords: str = Form(default=""),
+    mode: DetectionMode = Form(default=DetectionMode.sam3_text),
+    split_on_cut: bool = Form(default=False),
+    dino_variant: str = Form(default="v2"),
+    owl_ref_type: str = Form(default="text"),  # owlv2: text | image | both
+    owl_dino: str = Form(default="none"),  # owlv2 DINO-on-top: none | v2 | v3
+) -> JobResponse:
+    """Accept the uploads, kick off background analysis, return a job id."""
+    if not video.filename:
+        raise HTTPException(status_code=400, detail="A video file is required.")
+
+    try:
+        product_meta = json.loads(products)
+        assert isinstance(product_meta, list) and product_meta
+    except (json.JSONDecodeError, AssertionError):
+        raise HTTPException(status_code=400, detail="`products` must be a non-empty JSON list.")
+
+    job = store.create()
+    job_dir = UPLOAD_DIR / job.id
+    job_dir.mkdir(parents=True, exist_ok=True)
+
+    video_path = _save_upload(video, job_dir)
+    exemplar_files = [ex for ex in exemplars if ex.filename]
+
+    # Slice the flat exemplar list back into per-product groups by image_count.
+    product_inputs: list[ProductInput] = []
+    cursor = 0
+    for p in product_meta:
+        name = str(p.get("name", "")).strip()
+        count = int(p.get("image_count", 0))
+        group = exemplar_files[cursor : cursor + count]
+        cursor += count
+        paths = [_save_upload(ex, job_dir) for ex in group]
+        product_inputs.append(ProductInput(name=name, exemplar_paths=paths))
+
+    # Validation is driven by the mode registry (single source of truth).
+    opts = BuildOpts(dino_variant=dino_variant, owl_ref_type=owl_ref_type, owl_dino=owl_dino)
+    if requires_name(mode) and any(not p.name for p in product_inputs):
+        raise HTTPException(status_code=400, detail="Every product needs a name in this mode.")
+    if (requires_reference(mode) or owlv2_needs_images(mode, opts)) and any(
+        not p.exemplar_paths for p in product_inputs
+    ):
+        raise HTTPException(
+            status_code=400,
+            detail=f"{mode.value} mode requires a reference image for every product.",
+        )
+
+    variant = dino_variant if dino_variant in ("v2", "v3") else "v2"
+    owl_dino_v = owl_dino if owl_dino in ("none", "v2", "v3") else "none"
+    owl_ref = owl_ref_type if owl_ref_type in ("text", "image", "both") else "text"
+    background_tasks.add_task(
+        run_analysis, job.id, video_path, product_inputs, caption,
+        mention_keywords, mode, split_on_cut, variant, owl_ref, owl_dino_v,
+    )
+    return JobResponse(job_id=job.id)
+
+
+@app.get("/status/{job_id}", response_model=StatusResponse)
+def status(job_id: str) -> StatusResponse:
+    job = store.get(job_id)
+    if job is None:
+        raise HTTPException(status_code=404, detail="Unknown job_id.")
+    return StatusResponse(
+        job_id=job.id, status=job.status, result=job.result, error=job.error
+    )

app/mentions.py

@@ -0,0 +1,109 @@
+"""Phase 4 — mention counting (independent of the visual detection mode).
+
+- caption / audio: case-insensitive count of the brand name in text.
+- transcript: faster-whisper (local, CPU int8 — no PyTorch).
+- on-screen text: RapidOCR (ONNX) over sampled frames.
+
+Models load lazily as singletons; failures degrade to None rather than failing
+the whole job.
+"""
+
+import re
+import subprocess
+import tempfile
+from pathlib import Path
+
+OCR_FPS = 1  # frames/sec to OCR
+OCR_WIDTH = 720  # OCR likes a bit more resolution than the tracking pass
+WHISPER_MODEL = "base"
+
+_whisper = None
+_ocr = None
+
+
+def _get_whisper():
+    global _whisper
+    if _whisper is None:
+        from faster_whisper import WhisperModel
+
+        _whisper = WhisperModel(WHISPER_MODEL, device="cpu", compute_type="int8")
+    return _whisper
+
+
+def _get_ocr():
+    global _ocr
+    if _ocr is None:
+        from rapidocr_onnxruntime import RapidOCR
+
+        _ocr = RapidOCR()
+    return _ocr
+
+
+def parse_keywords(raw: str) -> list[str]:
+    """Split a comma-separated keyword string into a clean list."""
+    return [k.strip() for k in raw.split(",") if k.strip()]
+
+
+def count_mentions(text: str, keyword: str) -> int:
+    """Case-insensitive, non-overlapping count of `keyword` in `text`."""
+    if not text or not keyword.strip():
+        return 0
+    return len(re.findall(re.escape(keyword.strip()), text, flags=re.IGNORECASE))
+
+
+def count_keywords(text: str, keywords: list[str]) -> int:
+    """Total mentions of any keyword in `text`."""
+    return sum(count_mentions(text, k) for k in keywords)
+
+
+def count_keywords_breakdown(text: str, keywords: list[str]) -> dict[str, int]:
+    """Per-keyword mention counts in `text` (sums to count_keywords)."""
+    return {k: count_mentions(text, k) for k in keywords}
+
+
+def transcribe(video_path: str) -> str:
+    """Full transcript of the video's audio."""
+    segments, _info = _get_whisper().transcribe(video_path)
+    return " ".join(seg.text.strip() for seg in segments).strip()
+
+
+def count_ocr_mentions(
+    video_path: str, keywords: list[str]
+) -> tuple[int, dict[str, int]]:
+    """OCR sampled frames once and report (total, per-keyword) on-screen appearances.
+
+    Contiguous sampled frames are collapsed into a single appearance, so a keyword
+    that stays on screen for many frames counts once (a new appearance starts only on
+    a rising edge — present now, absent in the previous frame). `total` counts
+    appearances where *any* keyword is on screen; the per-keyword map counts each
+    keyword's appearances independently (a frame can match several keywords, so the
+    map need not sum to `total`)."""
+    if not keywords:
+        return 0, {}
+    ocr = _get_ocr()
+    needles = {k: k.lower() for k in keywords}
+    per_keyword = {k: 0 for k in keywords}
+    prev_present = {k: False for k in keywords}
+    hits = 0
+    prev_any = False
+    with tempfile.TemporaryDirectory() as tmp:
+        tmpd = Path(tmp)
+        subprocess.run(
+            ["ffmpeg", "-y", "-i", video_path, "-r", str(OCR_FPS),
+             "-vf", f"scale={OCR_WIDTH}:-2", str(tmpd / "f%05d.jpg")],
+            check=True, capture_output=True,
+        )
+        for frame in sorted(tmpd.glob("f*.jpg")):
+            result, _ = ocr(str(frame))
+            text = " ".join((line[1] or "") for line in (result or [])).lower()
+            matched_any = False
+            for k, needle in needles.items():
+                present = needle in text
+                if present and not prev_present[k]:  # rising edge → new appearance
+                    per_keyword[k] += 1
+                prev_present[k] = present
+                matched_any = matched_any or present
+            if matched_any and not prev_any:
+                hits += 1
+            prev_any = matched_any
+    return hits, per_keyword

app/schemas.py

@@ -0,0 +1,109 @@
+"""Pydantic models — the FE/BE contract. Mirrors the result schema in DesignDoc.md."""
+
+from enum import Enum
+
+from pydantic import BaseModel, Field
+
+
+class JobStatus(str, Enum):
+    queued = "queued"
+    running = "running"
+    done = "done"
+    error = "error"
+
+
+class DetectionMode(str, Enum):
+    sam3_text = "sam3_text"  # SAM 3 promptable by the product name (text)
+    sam3_clip = "sam3_clip"  # SAM 2 segment-everything + CLIP match vs reference image
+    sam3_text_dino = "sam3_text_dino"  # SAM 3 text + DINOv2 per-appearance verification
+    sam3_text_openai = "sam3_text_openai"  # SAM 3 text + OpenAI VLM per-appearance check
+    sam3_1_text_dino = "sam3_1_text_dino"  # SAM 3.1 text + DINOv2 (faster/better endpoint)
+    sam2_dino = "sam2_dino"  # SAM 2 segment-everything + DINOv2 match vs reference image
+    gdino_text_dino = "gdino_text_dino"  # Grounding DINO text seg (EVF-SAM) + DINOv2 score
+    owlv2 = "owlv2"  # OWLv2 detection (text/image/both refs) + confidence/objectness (+DINO)
+    none = "none"  # DEBUG: no analysis, returns a dummy result (easy to comment out)
+
+
+class Mask(BaseModel):
+    """A binary segmentation mask over a `w`×`h` grid (the full frame, downscaled),
+    run-length encoded row-major as alternating runs starting with background (0)."""
+
+    w: int
+    h: int
+    counts: list[int]
+
+
+class Box(BaseModel):
+    """A detection box, normalized to [0, 1] relative to frame size.
+
+    Carries one or more *named* scores (e.g. {"similarity": .8} for the DINO modes, or
+    {"confidence": .7, "objectness": .5, "similarity": .8} for OWLv2). `scores` are per-frame;
+    `avg_scores` are the per-appearance means (filled by build_product_result / the UI). Each
+    score key gets its own display + threshold + slider (see AnalysisResult.score_specs)."""
+
+    x: float
+    y: float
+    w: float
+    h: float
+    scores: dict[str, float] = Field(default_factory=dict)
+    avg_scores: dict[str, float] = Field(default_factory=dict)
+    mask: Mask | None = None  # target-object segmentation mask (segment/grounded modes)
+
+
+class ScoreSpec(BaseModel):
+    """Declares one score dimension a mode emits, and its post-hoc slider default."""
+
+    key: str  # matches Box.scores keys, e.g. "similarity" | "confidence" | "objectness"
+    label: str  # UI label, e.g. "Similarity"
+    default: float  # slider start (and the threshold the backend's candidate view uses)
+
+
+class ProductResult(BaseModel):
+    """Per-product visual track (Phase 2b — N products per analysis)."""
+
+    name: str
+    boxes: dict[str, list[Box]] = Field(default_factory=dict)  # keyed by frame index
+    appearances: list[tuple[float, float]] = Field(default_factory=list)
+    contiguous_appearances: int = 0
+    total_on_screen_sec: float = 0.0
+
+
+class AnalysisResult(BaseModel):
+    # video metadata
+    fps: float = 0.0
+    duration_sec: float = 0.0
+    # visual tracking — one entry per product (Phase 2b)
+    products: list[ProductResult] = Field(default_factory=list)
+    # mentions (Phase 4)
+    transcript: str | None = None
+    audio_mentions: int | None = None
+    caption_mentions: int | None = None
+    ocr_mentions: int | None = None  # contiguous on-screen-text appearances of the brand
+    # per-keyword breakdowns (keyword -> count), parallel to the totals above
+    audio_mention_counts: dict[str, int] | None = None
+    caption_mention_counts: dict[str, int] | None = None
+    ocr_mention_counts: dict[str, int] | None = None
+    # echo of the inputs we matched against
+    brand_name: str | None = None
+    mode: DetectionMode | None = None
+    # one entry per post-hoc slider (empty = no sliders for this mode). Boxes carry per-frame
+    # `scores`, so the UI re-filters/recounts live across all of them.
+    score_specs: list[ScoreSpec] = Field(default_factory=list)
+    # frame indices where a scene cut begins (only when split-on-cut is on for a slider
+    # mode); lets the UI re-split appearances at cuts after re-thresholding.
+    cut_frames: list[int] = Field(default_factory=list)
+
+
+class JobResponse(BaseModel):
+    """Returned by POST /analyze."""
+
+    job_id: str
+
+
+class StatusResponse(BaseModel):
+    """Returned by GET /status/{job_id}."""
+
+    job_id: str
+    status: JobStatus
+    result: AnalysisResult | None = None
+    error: str | None = None

app/similarity/__init__.py

@@ -0,0 +1,35 @@
+"""Similarity backends — interchangeable scorers reused across detection modes.
+
+`Embedder` (DinoEmbedder, ClipEmbedder) encodes reference image(s) and scores crops by
+cosine similarity; `OpenAiVerifier` is a frame-level yes/no verifier. Detection strategies
+inject whichever they need.
+"""
+
+from typing import Protocol
+
+import numpy as np
+from PIL import Image
+
+from .clip import ClipEmbedder
+from .dino import DEFAULT_DINO_VARIANT, DINO_THRESHOLD, DinoEmbedder
+from .vlm import OpenAiVerifier
+
+
+class Embedder(Protocol):
+    """Encodes references and scores crops against them (best similarity per crop)."""
+
+    def reference(self, paths: list[str]) -> np.ndarray:
+        """Encode reference image path(s) into rows (R, D) to match candidates against."""
+
+    def score(self, ref: np.ndarray, crops: list[Image.Image]) -> list[float]:
+        """Best similarity of each crop vs the reference rows."""
+
+
+__all__ = [
+    "Embedder",
+    "DinoEmbedder",
+    "ClipEmbedder",
+    "OpenAiVerifier",
+    "DINO_THRESHOLD",
+    "DEFAULT_DINO_VARIANT",
+]

app/similarity/clip.py

@@ -0,0 +1,43 @@
+"""CLIP image embedder (fastembed ONNX) — the original sam3_clip similarity backend.
+
+Exposes the same `Embedder` interface as DinoEmbedder, but `reference()` returns a single
+averaged centroid (1, D) — CLIP image-image cosines are weakly calibrated, so the historical
+behavior matched the mean reference with a high absolute threshold (see config.CLIP_*).
+"""
+
+import numpy as np
+from PIL import Image
+
+CLIP_MODEL = "Qdrant/clip-ViT-B-32-vision"
+_model = None  # lazy singleton (first use downloads ~350MB to the HF cache)
+
+
+def _embedder():
+    global _model
+    if _model is None:
+        from fastembed import ImageEmbedding
+
+        _model = ImageEmbedding(CLIP_MODEL)
+    return _model
+
+
+def _embed(images: list) -> np.ndarray:
+    """(N, D) L2-normalized rows. fastembed accepts PIL images or paths directly."""
+    if not images:
+        return np.zeros((0, 1), dtype=np.float32)
+    embs = np.stack([np.asarray(e, dtype=np.float32) for e in _embedder().embed(images)])
+    return embs / (np.linalg.norm(embs, axis=1, keepdims=True) + 1e-8)
+
+
+class ClipEmbedder:
+    """An `Embedder` backed by CLIP, matching against an averaged reference centroid."""
+
+    def reference(self, paths: list[str]) -> np.ndarray:
+        ref = _embed(paths).mean(axis=0)
+        return (ref / (np.linalg.norm(ref) + 1e-8))[None, :]  # (1, D)
+
+    def score(self, ref: np.ndarray, crops: list[Image.Image]) -> list[float]:
+        if not crops:
+            return []
+        sims = (_embed(crops) @ ref.T).max(axis=1)
+        return [float(s) for s in sims]

app/similarity/dino.py

@@ -0,0 +1,110 @@
+"""DINOv2 / DINOv3 image embedder (ONNX, local, no PyTorch).
+
+Two interchangeable backbones, picked per analysis with `variant`:
+  "v2" — Xenova/dinov2-small (DINOv2 ViT-S/14, Apache-2.0, ungated).
+  "v3" — onnx-community/dinov3-vits16-pretrain-lvd1689m-ONNX (DINOv3 ViT-S/16,
+         dinov3-license; this re-export downloads ungated, though the upstream Facebook
+         weights are gated). Weights are external data (model.onnx_data) fetched beside the
+         graph; the graph adds 4 register tokens after CLS, which the CLS-at-index-0 read
+         ignores.
+Both share preprocessing (224px, ImageNet norm) and CLS-token extraction; same 384-d
+embedding, so v2/v3 scores are directly comparable.
+"""
+
+import numpy as np
+from PIL import Image
+
+DINO_REPOS = {
+    "v2": "Xenova/dinov2-small",
+    "v3": "onnx-community/dinov3-vits16-pretrain-lvd1689m-ONNX",
+}
+DEFAULT_DINO_VARIANT = "v2"
+DINO_THRESHOLD = 0.0  # cosine gate for the SAM-text DINO refiner (same-object ~0.8)
+# True: score a candidate against the BEST matching reference (max over views) — distinct
+# reference angles/lighting stay distinct. False: collapse references to one averaged
+# centroid. (Paired with analysis.config.DINO_MASK_BG, which controls the crop input.)
+DINO_MULTI_REF = False
+
+_MEAN = np.array([0.485, 0.456, 0.406], dtype=np.float32)
+_STD = np.array([0.229, 0.224, 0.225], dtype=np.float32)
+_sessions: dict = {}  # variant -> onnxruntime.InferenceSession (lazy)
+
+
+def _session(variant: str = DEFAULT_DINO_VARIANT):
+    if variant not in _sessions:
+        import onnxruntime as ort
+        from huggingface_hub import hf_hub_download
+
+        repo = DINO_REPOS.get(variant, DINO_REPOS[DEFAULT_DINO_VARIANT])
+        path = hf_hub_download(repo, "onnx/model.onnx")
+        # Some exports (DINOv3) store weights as external data beside the graph; pull the
+        # companion into the same snapshot dir so onnxruntime can resolve the relative path.
+        try:
+            hf_hub_download(repo, "onnx/model.onnx_data")
+        except Exception:  # self-contained graph (e.g. DINOv2) — nothing to fetch
+            pass
+        _sessions[variant] = ort.InferenceSession(path, providers=["CPUExecutionProvider"])
+    return _sessions[variant]
+
+
+def _preprocess(img: Image.Image) -> np.ndarray:
+    img = img.convert("RGB").resize((224, 224), Image.BICUBIC)
+    arr = (np.asarray(img, dtype=np.float32) / 255.0 - _MEAN) / _STD
+    return arr.transpose(2, 0, 1)  # CHW
+
+
+def _cls_tokens(sess, outputs: list[np.ndarray]) -> np.ndarray:
+    """(N, D) CLS embeddings, robust to output ordering/naming across the v2 and v3
+    exports. DINOv3 prepends register tokens after CLS, but CLS stays at index 0."""
+    names = [o.name for o in sess.get_outputs()]
+    for arr, name in zip(outputs, names):
+        if "last_hidden_state" in name and arr.ndim == 3:
+            return arr[:, 0]
+    for arr in outputs:
+        if arr.ndim == 3:  # (N, tokens, D) -> CLS at 0
+            return arr[:, 0]
+        if arr.ndim == 2:  # already pooled (N, D)
+            return arr
+    return outputs[0][:, 0]
+
+
+def _embed_batch(imgs: list[Image.Image], variant: str) -> np.ndarray:
+    """Embed many crops in a single ONNX run; returns (N, D) L2-normalized rows."""
+    if not imgs:
+        return np.zeros((0, 1), dtype=np.float32)
+    batch = np.stack([_preprocess(im) for im in imgs])
+    sess = _session(variant)
+    out = sess.run(None, {sess.get_inputs()[0].name: batch})
+    v = _cls_tokens(sess, out)
+    return v / (np.linalg.norm(v, axis=1, keepdims=True) + 1e-8)
+
+
+def _reference_set(paths: list[str], variant: str) -> np.ndarray:
+    """Per-reference normalized embeddings (R, D) to match a candidate against with the MAX
+    cosine over rows. When DINO_MULTI_REF is False, collapse to one averaged-centroid row
+    (1, D) so the max degenerates to the old single-cosine behavior."""
+    embs = _embed_batch([Image.open(p) for p in paths], variant)
+    if not DINO_MULTI_REF and embs.shape[0] > 1:
+        ref = embs.mean(axis=0)
+        return (ref / (np.linalg.norm(ref) + 1e-8))[None, :]
+    return embs
+
+
+def _max_sim(ref: np.ndarray, embs: np.ndarray) -> np.ndarray:
+    """(N,) best cosine of each row of `embs` (N, D) against any reference row `ref` (R, D)."""
+    return (embs @ ref.T).max(axis=1)
+
+
+class DinoEmbedder:
+    """An `Embedder`: encodes references and scores crops by best cosine over reference views."""
+
+    def __init__(self, variant: str = DEFAULT_DINO_VARIANT):
+        self.variant = variant if variant in DINO_REPOS else DEFAULT_DINO_VARIANT
+
+    def reference(self, paths: list[str]) -> np.ndarray:
+        return _reference_set(paths, self.variant)
+
+    def score(self, ref: np.ndarray, crops: list[Image.Image]) -> list[float]:
+        if not crops:
+            return []
+        return [float(s) for s in _max_sim(ref, _embed_batch(crops, self.variant))]

app/similarity/vlm.py

@@ -0,0 +1,53 @@
+"""OpenAI vision verifier — asks a VLM "is <product> in this frame?" (frame-level yes/no).
+
+Unlike the embedders, this is a `FrameVerifier`: it judges a whole frame, not a crop, so it
+returns a bool rather than a similarity. Used by the sam3_text_openai mode.
+"""
+
+import base64
+import io
+import os
+
+from PIL import Image
+
+OPENAI_MODEL = "gpt-4o-mini"
+_client = None
+
+
+def _openai_client():
+    global _client
+    if _client is None:
+        from openai import OpenAI
+
+        if not os.getenv("OPENAI_API_KEY"):
+            raise RuntimeError("OPENAI_API_KEY is not set (see GettingStarted.md).")
+        _client = OpenAI()
+    return _client
+
+
+class OpenAiVerifier:
+    """Frame-level verifier for a single product/brand."""
+
+    def __init__(self, brand: str):
+        self.brand = brand
+
+    def verify(self, frame: Image.Image, box=None) -> bool:  # noqa: ARG002 — frame-level
+        buf = io.BytesIO()
+        frame.convert("RGB").save(buf, format="JPEG", quality=85)
+        b64 = base64.b64encode(buf.getvalue()).decode()
+        resp = _openai_client().chat.completions.create(
+            model=OPENAI_MODEL,
+            max_tokens=3,
+            messages=[
+                {
+                    "role": "user",
+                    "content": [
+                        {"type": "text",
+                         "text": f"Is {self.brand} visible in this image? Answer only 'yes' or 'no'."},
+                        {"type": "image_url",
+                         "image_url": {"url": f"data:image/jpeg;base64,{b64}"}},
+                    ],
+                }
+            ],
+        )
+        return "yes" in (resp.choices[0].message.content or "").strip().lower()

pyproject.toml

@@ -0,0 +1,21 @@
+[project]
+name = "creator-vision-backend"
+version = "0.1.0"
+description = "Creator Vision API — brand-presence analysis for video"
+requires-python = ">=3.12"
+dependencies = [
+    "fastapi>=0.115",
+    "uvicorn[standard]>=0.32",
+    "python-multipart>=0.0.12",
+    "python-dotenv>=1.0",
+    "fal-client>=0.5",
+    "pillow>=11.0",
+    "fastembed>=0.4",
+    "numpy>=1.26",
+    "faster-whisper>=1.0",
+    "rapidocr-onnxruntime>=1.3",
+    "openai>=1.50",
+]
+
+[tool.uv]
+package = false