backend/scene_processing/exclusions.py

"""
Exclusion checks for the scene pipeline.

The Search-UI codebase has three places a video can be flagged as
out-of-scope for indexing:

1. Hardcoded music-subcategory check in ``audio_descriptions.is_excluded_music``
   (covers "Sing Out Joyfully" and "Sing to Jehovah" meeting songs).
2. ``video_exclusions`` table in the search DB — manually flagged
   natural keys.
3. ``exclusion_rules`` table in the search DB — auto-exclude rules
   keyed on category / subcategory / title / duration.

This module reads (1)+(2)+(3) once into a cheap snapshot and exposes a
single ``should_exclude(media_item)`` predicate so callers (the rolling
pipeline, the backlog generator) don't each have to re-implement the
chain.

Loading the snapshot does **not** require a full ``SubtitleSearch``
instance — we go straight at the SQLite file via ``get_search_db_path``.
That keeps the rolling pipeline's startup cheap and avoids dragging in
embedding-model loading on the GPU pod just to read a few table rows.
"""

from __future__ import annotations

import sqlite3
from dataclasses import dataclass, field
from typing import Any

from runtime_paths import get_search_db_path


@dataclass
class ExclusionSnapshot:
    """Cached view of the exclusion tables, plus the hardcoded rules."""

    excluded_keys: set[str] = field(default_factory=set)
    rules: list[dict[str, Any]] = field(default_factory=list)

    def __len__(self) -> int:
        return len(self.excluded_keys) + len(self.rules)


def load_exclusions(db_path: str | None = None) -> ExclusionSnapshot:
    """Snapshot the video_exclusions + exclusion_rules tables.

    Returns an empty snapshot if either table is missing — that's a
    valid state on a fresh data root, not an error.
    """
    path = db_path or get_search_db_path()
    snapshot = ExclusionSnapshot()
    try:
        conn = sqlite3.connect(path)
    except sqlite3.OperationalError:
        return snapshot
    try:
        try:
            cur = conn.execute("SELECT natural_key FROM video_exclusions")
            snapshot.excluded_keys = {row[0] for row in cur.fetchall()}
        except sqlite3.OperationalError:
            pass
        try:
            cur = conn.execute(
                """SELECT name, rule_type, category_match, subcategory_match,
                          title_match, duration_min_seconds, enabled
                   FROM exclusion_rules"""
            )
            for name, rule_type, cat, subcat, title, dur, enabled in cur.fetchall():
                if not enabled:
                    continue
                snapshot.rules.append(
                    {
                        "name": name,
                        "rule_type": rule_type or "",
                        "category_match": cat,
                        "subcategory_match": subcat,
                        "title_match": title,
                        "duration_min_seconds": dur,
                    }
                )
        except sqlite3.OperationalError:
            pass
    finally:
        conn.close()
    return snapshot


# ---- Predicates ---------------------------------------------------------


def is_excluded_music(media_item: dict[str, Any]) -> bool:
    """Return True for the hardcoded meeting-song subcategories.

    Mirrors ``backend.audio_descriptions.is_excluded_music`` so the
    rolling pipeline doesn't have to import that module (which carries
    requests / subprocess deps it doesn't otherwise need).
    """
    category = (media_item.get("_category") or "").lower()
    if "music" not in category:
        return False
    subcategory = (media_item.get("_subcategory") or "").lower()
    return "sing out joyfully" in subcategory or "sing to jehovah" in subcategory


def _rule_matches(rule: dict[str, Any], media_item: dict[str, Any]) -> bool:
    """Apply one exclusion rule to one media item.

    Mirrors the matching logic of ``backend.search_db_exclusions.
    matches_exclusion_rules`` — kept private here so callers go through
    ``should_exclude`` which fuses every check.
    """
    category = (media_item.get("_category") or "").lower()
    subcategory = (media_item.get("_subcategory") or "").lower()
    title = (media_item.get("title") or "").lower()
    duration = media_item.get("duration") or 0
    rule_type = rule.get("rule_type") or ""

    if rule_type == "category" and rule.get("category_match"):
        return rule["category_match"].lower() in category
    if rule_type == "subcategory_exact" and rule.get("subcategory_match"):
        return rule["subcategory_match"].lower() == subcategory
    if rule_type == "subcategory" and rule.get("subcategory_match"):
        return rule["subcategory_match"].lower() in subcategory
    if rule_type == "title" and rule.get("title_match"):
        return rule["title_match"].lower() in title
    if rule_type == "duration" and rule.get("duration_min_seconds"):
        return duration >= rule["duration_min_seconds"]
    return False


def should_exclude(
    natural_key: str,
    media_item: dict[str, Any] | None,
    snapshot: ExclusionSnapshot,
) -> tuple[bool, str | None]:
    """Return ``(excluded, reason)`` for one media item.

    The reason string is intended for log lines and the backlog-
    generator's drop-count summary. ``None`` means "no reason because
    not excluded."

    Order of checks (cheap first):
        1. natural_key in the manual ``video_exclusions`` set.
        2. Hardcoded music-subcategory check.
        3. Active exclusion rules (first match wins).
    """
    if natural_key in snapshot.excluded_keys:
        return True, "video_exclusions table"

    if media_item is None:
        # No catalog entry — we can't apply category / rule checks.
        # Surface as a soft exclusion so the caller can decide what to
        # do. The rolling pipeline treats this as a hard failure (it
        # cannot download without the media item anyway).
        return False, None

    if is_excluded_music(media_item):
        return True, "music subcategory (meeting song)"

    for rule in snapshot.rules:
        if _rule_matches(rule, media_item):
            return True, f"rule: {rule.get('name') or rule.get('rule_type', 'unnamed')}"

    return False, None