4baf6ea896
Generic movie titles ("Monster Tits", "Pirates") map to many different TPDB
films with identical titles (different studios/casts, all title-score 1.0). The
old matcher searched per_page=10 and ranked by title only, so for a generic title
the correct film was often not even in the top 10, and among same-title
candidates it picked arbitrarily. Result: a ~95% no_match rate and silent
misattribution (e.g. "Monster Tits" by Venom Digital Media would get Galaxy
Productions' TPDB entry).
_best_match now:
- searches per_page=40 (the right film for a generic title is often past top 10),
- ranks title-gate survivors by a composite of title + studio similarity +
performer overlap (our studio/cast from the primary source disambiguate which
same-title film it is),
- guards against misattribution: if we have a studio/cast signal and there is
more than one near-identical-title candidate but the winner shares neither
studio nor cast, return no_match instead of attaching a wrong same-title film.
Verified on prod data: a no_match-with-studio sample now matches 18/18 with the
correct studio (fixing Galaxy to Venom, Exquisite to Rodney Moore, and a no_match
to Cherry Boxxx), and an already-enriched sample keeps 16/18 identical picks with
the 2 differences being the same studio (benign TPDB duplicate). No wrong-studio
regressions observed.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
355 lines
14 KiB
Python
355 lines
14 KiB
Python
"""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
|