"""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=, 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, timedelta from rapidfuzz import fuzz from sqlalchemy import func, select, text 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__) # Film bez trafienia w TPDB wraca do kolejki dopiero po tym czasie (TPDB dopisuje filmy, # wiec kiedys warto sprawdzic ponownie, ale nie cztery razy dziennie). _RECHECK_AFTER = timedelta(days=30) _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. Kolejność: najpierw nigdy nie sprawdzane (najnowsze pierwsze), potem te sprawdzone najdawniej, ale nie wcześniej niż _RECHECK_AFTER od ostatniego razu. Bez tego każdy przebieg brał te same 200 najnowszych bez trafienia, a starszych nie ruszał wcale (migracja 0051).""" 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))) .where( Movie.tpdb_checked_at.is_(None) | (Movie.tpdb_checked_at < datetime.now(UTC) - _RECHECK_AFTER) ) .order_by(Movie.tpdb_checked_at.asc().nulls_first(), 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 ) # Surowy UPDATE: nie rusza updated_at, a po merge skasowany film po prostu # nie ma czego oznaczyc. Blad (wyjatek nizej) nie oznacza, wroci w nastepnym # przebiegu, bo moze to byc chwilowa awaria TPDB. session.execute( text("UPDATE movies SET tpdb_checked_at = now() WHERE id = :id"), {"id": mid} ) 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