app/analysis/runner.py

"""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},
    )