goon/app/enrich/tpdb_movies.py

"""TPDB movie enrichment + dedup.

TPDB `/movies` jest naszym kanonicznym źródłem metadanych filmów: obsada, kategorie
(tagi), studio, reżyser, rok, plus stabilny UUID do dedupu. paradisehill (primary
movie source) prawie nie ma obsady, więc TPDB wypełnia największą lukę.

Zasada: TPDB TYLKO wzbogaca ISTNIEJĄCE filmy (z paradisehill/dooplay) i dedupuje,
NIGDY nie tworzy nowych, TPDB nie ma playbacku, więc nowy film byłby niegrywalny.

Flow per film:
  1. skip, jeśli film ma już TPDB movie ref (movie_external_refs jest tylko-movie,
     więc dowolny ref z tpdb source = już wzbogacony),
  2. search TPDB /movies?q=<tytuł>, wybierz najlepszego kandydata (token-set na tytule
     + guard roku ±2, próg `min_title`),
  3. DEDUP: jeśli ten TPDB UUID jest już przypięty do INNEGO naszego filmu → to ten
     sam film (mirror) → merge_movies (keep = starszy created_at),
  4. wzbogać ocalały film: obsada (resolve_performer → MoviePerformer), tagi
     (resolve_tag → MovieTag source=tpdb), studio (fill studio_id), reżyser/rok/
     poster/rating (fill-only), przypnij MovieExternalRef(tpdb, uuid).

Match jest zachowawczy (próg 0.90 + guard roku), bo TPDB search zwraca dużo (np.
"Pirates" → gay Knightbreeders przed feature), bez scoringu wzięlibyśmy zły film.
"""
from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime

from rapidfuzz import fuzz
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.connectors.tpdb import TPDBConnector, _parse_movie
from app.models.movie import Movie, MovieExternalRef, MoviePerformer
from app.models.movie_playback_source import MoviePlaybackSource
from app.models.performer import Performer
from app.models.studio import Studio
from app.normalize.movies import normalize_movie
from app.normalize.text import normalize
from app.resolve.movie_merge import merge_movies
from app.resolve.movie_resolver import _sync_performers, _sync_tags
from app.resolve.performer_resolver import resolve_performer
from app.resolve.studio_resolver import resolve_studio

log = logging.getLogger(__name__)

_DASH = str.maketrans({"–": "-", "—": "-", "‑": "-"})


def _cand_year(raw: dict) -> int | None:
    d = raw.get("date")
    if d and len(str(d)) >= 4:
        try:
            return int(str(d)[:4])
        except ValueError:
            return None
    return None


def _movie_studio_name(session: Session, movie: Movie) -> str | None:
    if not movie.studio_id:
        return None
    st = session.get(Studio, movie.studio_id)
    return st.name if st is not None else None


def _movie_performer_names(session: Session, movie: Movie) -> set[str]:
    """Znormalizowane nazwy performerów już przypiętych do filmu (do rozróżniania
    kandydatów TPDB o tym samym tytule)."""
    rows = session.execute(
        select(Performer.name_normalized)
        .join(MoviePerformer, MoviePerformer.performer_id == Performer.id)
        .where(MoviePerformer.movie_id == movie.id)
    ).all()
    return {r[0] for r in rows if r[0]}


def _cand_studio_name(raw: dict) -> str:
    site = raw.get("site") or {}
    name = site.get("name") if isinstance(site, dict) else site
    return normalize(name or "")


def _cand_performer_names(raw: dict) -> set[str]:
    return {
        normalize(p.get("name"))
        for p in (raw.get("performers") or [])
        if isinstance(p, dict) and p.get("name")
    }


def _best_match(
    session: Session, connector: TPDBConnector, movie: Movie, *, min_title: float
) -> dict | None:
    """Najlepszy TPDB movie payload dla naszego filmu, albo None.

    Tytuł to gate (`token_sort_ratio` >= min_title): odporny na inwersję i kolejność,
    penalizuje różnicę długości (krótki generyczny tytuł nie łapie dłuższego nadzbioru).
    ALE generyczne tytuły ("Monster Tits", "Pirates") mają w TPDB WIELE różnych filmów
    o tym samym tytule (różne studia/obsady) i identycznym score 1.0. Sam tytuł ich NIE
    rozróżni, a `per_page=10` często w ogóle nie zwracał właściwego (poprawny film bywa
    poza top-10). Dlatego:
      - `per_page=40` (właściwy film generycznego tytułu bywa dalej w wynikach),
      - wśród kandydatów przechodzących title-gate wybieramy po KOMPOZYCIE:
        tytuł + zbieżność studia + pokrycie obsady (nasze studio/obsada z primary
        źródła jak paradisehill rozróżniają który to film),
      - GUARD anty-misattribution: jeśli mamy sygnał (studio na naszym filmie) i jest
        >1 kandydat o ~identycznym tytule, a zwycięzca NIE dzieli studia, to no_match
        (lepiej nie wzbogacić niż przypiąć zły film o tej samej nazwie).
    Precyzja > recall. Guard roku ±2 (inny rok = inna edycja)."""
    query = (movie.title or "").translate(_DASH).strip()[:60]
    if not query:
        return None
    my_norm = normalize(movie.title)
    my_studio = _movie_studio_name(session, movie)
    my_studio_norm = normalize(my_studio) if my_studio else ""
    my_perfs = _movie_performer_names(session, movie)

    best: dict | None = None
    best_key: tuple[float, float, float, float] | None = None
    best_studio_sim = 0.0
    best_perf_overlap = 0.0
    n_exact_title = 0
    for raw in connector.search_movies(query, per_page=40):
        cand_title = raw.get("title")
        if not cand_title:
            continue
        tscore = fuzz.token_sort_ratio(my_norm, normalize(cand_title)) / 100.0
        if tscore < min_title:
            continue
        cy = _cand_year(raw)
        if movie.release_year and cy and abs(movie.release_year - cy) > 2:
            continue  # guard: inny rok → prawdopodobnie inny film (Taxi 2 ≠ Taxi Violeur 2)
        if tscore >= 0.97:
            n_exact_title += 1
        cand_studio = _cand_studio_name(raw)
        studio_sim = (
            fuzz.token_set_ratio(my_studio_norm, cand_studio) / 100.0
            if my_studio_norm and cand_studio
            else 0.0
        )
        cand_perfs = _cand_performer_names(raw)
        perf_overlap = (len(my_perfs & cand_perfs) / len(my_perfs)) if my_perfs else 0.0
        # Kompozyt: tytuł jako baza, studio i obsada jako rozróżniacze (waga 0.6 każdy).
        composite = tscore + 0.6 * studio_sim + 0.6 * perf_overlap
        key = (composite, tscore, studio_sim, perf_overlap)
        if best_key is None or key > best_key:
            best_key = key
            best = raw
            best_studio_sim = studio_sim
            best_perf_overlap = perf_overlap

    if best is None:
        return None
    # Anty-misattribution: mamy czym rozróżnić (studio lub obsada na naszym filmie), jest
    # >1 kandydat o ~identycznym tytule, a zwycięzca nie dzieli NIC (ani studia, ani
    # obsady) → to prawie na pewno inny film o tej samej nazwie. Nie wzbogacaj.
    my_has_signal = bool(my_studio_norm) or bool(my_perfs)
    if (
        my_has_signal
        and n_exact_title >= 2
        and best_studio_sim < 0.5
        and best_perf_overlap < 0.5
    ):
        return None
    return best


def _fill_scalar_fields(movie: Movie, norm) -> None:
    """Fill-only, nie nadpisujemy pól ustawionych przez primary (paradisehill/mirror)."""
    if norm.director and not movie.director:
        movie.director = norm.director
    if norm.release_year and not movie.release_year:
        movie.release_year = norm.release_year
    if norm.release_date and not movie.release_date:
        movie.release_date = norm.release_date
    if norm.duration_sec and not movie.duration_sec:
        movie.duration_sec = norm.duration_sec
    if norm.description and not movie.description:
        movie.description = norm.description
    if norm.poster_url and not movie.poster_url:
        movie.poster_url = norm.poster_url
    if norm.backdrop_url and not movie.backdrop_url:
        movie.backdrop_url = norm.backdrop_url
    if norm.rating is not None and movie.rating is None:
        movie.rating = norm.rating


def enrich_movie(
    session: Session,
    movie: Movie,
    *,
    connector: TPDBConnector,
    source_id: uuid.UUID,
    min_title: float = 0.90,
) -> str:
    """Zwraca: 'skip' | 'no_match' | 'enriched' | 'merged'."""
    already = session.execute(
        select(MovieExternalRef.external_id).where(
            MovieExternalRef.source_id == source_id,
            MovieExternalRef.movie_id == movie.id,
        )
    ).first()
    if already is not None:
        return "skip"

    raw = _best_match(session, connector, movie, min_title=min_title)
    if raw is None:
        return "no_match"
    ext_id = str(raw["id"])

    outcome = "enriched"
    # DEDUP: ten TPDB UUID już przypięty do innego naszego filmu → mirror tego samego.
    other = session.execute(
        select(MovieExternalRef).where(
            MovieExternalRef.source_id == source_id,
            MovieExternalRef.external_id == ext_id,
        )
    ).scalar_one_or_none()
    if other is not None and other.movie_id != movie.id:
        m2 = session.get(Movie, other.movie_id)
        if m2 is not None:
            now = datetime.now(UTC)
            keep, drop = (
                (movie, m2)
                if (movie.created_at or now) <= (m2.created_at or now)
                else (m2, movie)
            )
            movie = merge_movies(
                session, keep_id=keep.id, drop_id=drop.id, resolved_by="tpdb:dedup"
            )
            outcome = "merged"

    rm = _parse_movie(raw)
    if rm is None:
        return "no_match"
    norm = normalize_movie(rm)

    if norm.studio is not None:
        studio = resolve_studio(session, norm=norm.studio, source_id=source_id)
        if studio is not None and not movie.studio_id:
            movie.studio_id = studio.id

    # Dedup wchodzących performerów po external_id (TPDB potrafi wylistować tego samego
    # kanonicznego performera 2×: pod aliasem i kanonicznie), bez tego resolve_performer
    # próbuje wstawić performer_external_refs 2× → UniqueViolation (jak w scene_resolver).
    resolved: list[tuple[uuid.UUID, str | None]] = []
    seen_perf_keys: set[str] = set()
    for p_norm in norm.performers:
        key = p_norm.external_id or normalize(p_norm.name)
        if key in seen_perf_keys:
            continue
        seen_perf_keys.add(key)
        performer = resolve_performer(session, norm=p_norm, source_id=source_id)
        resolved.append((performer.id, p_norm.as_alias_in_scene))
    _sync_performers(session, movie_id=movie.id, resolved=resolved)
    _sync_tags(session, movie_id=movie.id, norm=norm, source_id=source_id)
    _fill_scalar_fields(movie, norm)

    ref = session.execute(
        select(MovieExternalRef).where(
            MovieExternalRef.source_id == source_id,
            MovieExternalRef.external_id == ext_id,
        )
    ).scalar_one_or_none()
    if ref is None:
        session.add(
            MovieExternalRef(
                source_id=source_id,
                external_id=ext_id,
                movie_id=movie.id,
                confidence=1.0,
                url=rm.url,
            )
        )
    elif ref.movie_id != movie.id:
        ref.movie_id = movie.id

    return outcome


def _candidate_movies(session: Session, *, source_id: uuid.UUID, limit: int) -> list[Movie]:
    """Filmy jeszcze nie wzbogacone TPDB, priorytet: mają żywy playback (są grywalne)
    i brak obsady LUB brak studia (największa luka). Reszta później."""
    has_tpdb = (
        select(MovieExternalRef.movie_id)
        .where(MovieExternalRef.source_id == source_id)
        .scalar_subquery()
    )
    has_playback = (
        select(MoviePlaybackSource.movie_id)
        .where(MoviePlaybackSource.dead_at.is_(None))
        .scalar_subquery()
    )
    perf_count = (
        select(func.count())
        .select_from(MoviePerformer)
        .where(MoviePerformer.movie_id == Movie.id)
        .correlate(Movie)
        .scalar_subquery()
    )
    stmt = (
        select(Movie)
        .where(
            Movie.id.not_in(has_tpdb),
            Movie.id.in_(has_playback),
        )
        .where((perf_count == 0) | (Movie.studio_id.is_(None)))
        .order_by(Movie.created_at.desc())
        .limit(limit)
    )
    return list(session.execute(stmt).scalars().all())


def run_tpdb_movie_enrich(
    session_factory,
    *,
    limit: int = 200,
    min_title: float = 0.90,
) -> dict[str, int]:
    """Batch: wzbogać do `limit` filmów. Commit per-film (jeden błąd nie cofa reszty).
    `session_factory` = kontekstowy scope (app.db.session_scope)."""
    from app.ingest import get_or_create_source
    from app.models.source import SourceKind

    connector = TPDBConnector()
    counters = {"seen": 0, "enriched": 0, "merged": 0, "no_match": 0, "skip": 0, "errors": 0}

    with session_factory() as session:
        src = get_or_create_source(session, kind=SourceKind.tpdb, name="tpdb")
        source_id = src.id
        session.commit()
        movies = _candidate_movies(session, source_id=source_id, limit=limit)
        movie_ids = [m.id for m in movies]

    log.info("tpdb-movie-enrich: %d candidate movies", len(movie_ids))
    for mid in movie_ids:
        counters["seen"] += 1
        try:
            with session_factory() as session:
                movie = session.get(Movie, mid)
                if movie is None:
                    continue
                outcome = enrich_movie(
                    session, movie, connector=connector, source_id=source_id, min_title=min_title
                )
                session.commit()
            counters[outcome] = counters.get(outcome, 0) + 1
        except Exception as e:  # pragma: no cover - defensywnie, jeden film nie wywala batcha
            counters["errors"] += 1
            log.warning("tpdb-movie-enrich failed for %s: %s", mid, e)

    log.info("tpdb-movie-enrich done: %s", counters)
    return counters