"""Composite scoring dla movies, title + year + studio + cast Jaccard. Movies vs Scenes scoring: - Movies rzadko mają fingerprinty, tytuł jest najsilniejszym sygnałem (weight 0.5). - Year (production year) ważny ale czasem nieznany (paradisehill często NULL); wagę redystrybuujemy proporcjonalnie gdy brak. - Studio: binary signal, match=1.0, mismatch=0.0; gdy któraś strona ma NULL, sygnał pomijamy (None waga). - Cast: Jaccard po canonical UUID-ach performerów. Bonus signal, niektóre source'y (paradisehill często, dooplay zawsze) mają cast; tube'y bez cast → None. Threshold (lower niż scenes, title dla movies ma niższy false-positive rate): ≥0.85 auto-merge 0.65-0.85 review <0.65 nowy kanon """ from __future__ import annotations import re import uuid from collections.abc import Iterable from dataclasses import dataclass from rapidfuzz import fuzz from sqlalchemy.orm import Session from app.models.movie import Movie, MoviePerformer from app.models.studio import Studio @dataclass class MovieScoreBreakdown: title: float | None = None year: float | None = None studio_match: bool | None = None cast: float | None = None composite: float = 0.0 reasons: dict | None = None def to_dict(self) -> dict: return { "title": self.title, "year": self.year, "studio_match": self.studio_match, "cast": self.cast, "composite": self.composite, "reasons": self.reasons or {}, } _BASE_WEIGHTS = { "title": 0.50, "year": 0.20, "studio": 0.20, "cast": 0.10, } # Movies-specific thresholds (lower niż scenes ze względu na unique titles) AUTO_THRESHOLD = 0.85 REVIEW_THRESHOLD = 0.65 _ROMAN_RE = re.compile(r"^(x{0,3})(ix|iv|v?i{0,3})$") _ROMAN_VALUES = {"i": 1, "v": 5, "x": 10} def _roman_value(token: str) -> int | None: """Wartosc cyfry rzymskiej II..XXXIX; pojedyncze litery (i, v, x) to zwykle slowa, a "xx"/"xxx" w tych tytulach to slowo ("Star Wars XXX"), nie liczba.""" if len(token) < 2 or not _ROMAN_RE.match(token) or set(token) == {"x"}: return None total, prev = 0, 0 for ch in reversed(token): v = _ROMAN_VALUES[ch] total = total - v if v < prev else total + v prev = max(prev, v) return total if 2 <= total <= 39 else None def _roman_to_arabic(title_normalized: str) -> str: out = [] for t in title_normalized.split(): v = _roman_value(t) out.append(str(v) if v is not None else t) return " ".join(out) def _series_number(title_normalized: str) -> int | None: """Ostatnia samodzielna liczba w tytule, z pominieciem lat (1900-2099). "her way to use tinder 61" -> 61, "the doctor is in 2" -> 2, "the doctor" -> None, "cece 2019" -> None (rok to nie numer odcinka). """ liczby = [ int(t) for t in title_normalized.split() if t.isdigit() and not 1900 <= int(t) <= 2099 ] return liczby[-1] if liczby else None # Tyle najwyzej dostaje tytul, gdy numer serii ma tylko jedna strona. Ponizej progu # auto-merge, zeby sam tytul nie decydowal, ale dosc, zeby ze zgodnym studiem i rokiem # trafic do recznego przegladu zamiast od razu odpadac. _ONE_SIDED_NUMBER_CAP = 0.6 # Slowa, ktore lustra dooplay dokladaja do tytulu w SEO ("Watch Cece Adult Movie Online # Free"). Nie niosa tresci, wiec przy tescie podzbioru je pomijamy. _SEO_WYPELNIACZE = frozenset({ "watch", "adult", "xxx", "movie", "movies", "film", "online", "free", "full", "porn", "hd", "stream", "streaming", "the", "a", "an", "and", "of", }) def _title_signal(cand_title: str, inc_title: str) -> tuple[float, bool, dict]: """(ocena tytulu 0..1, twarde odrzucenie, powody). `token_set_ratio` mierzy czesc wspolna, wiec tytul bedacy podzbiorem drugiego dostaje pelne 100: "the doctor" vs "the doctor is in 2", "fantasies" vs "dark fantasies 3". Numerowane serie wychodza na 90-96%, tez ponad progiem auto-merge. Gdy zrodlo nie ma roku ani studia (lustra dooplay czesto nie maja), o wszystkim decyduje sam tytul i cala seria zapada sie w jeden wpis. Zgloszenie 32100b94 (2026-08-30): pod "The Doctor" siedzialo 9 roznych filmow z pandamovies, w tym granny porn; w bazie 1878 filmow ma co najmniej dwie rozne strony z pandamovies, a jedna to norma. Zasady, obie oparte na numerze serii (ostatnia liczba w tytule, bez lat): * oba tytuly maja numer i sie rozni -> twarde odrzucenie (to inny odcinek), * numer ma tylko jedna strona -> tytul najwyzej 0.6 (inne wydanie, nie lustro). Sam podzbior bez numerow zostaje bez kary celowo: "cece" vs "watch cece adult" to prawdziwe lustro z przedrostkiem SEO i to wlasnie dla niego jest token_set. """ ocena = fuzz.token_set_ratio(cand_title, inc_title) / 100.0 powody: dict = {} n_c, n_i = _series_number(cand_title), _series_number(inc_title) if n_c is not None and n_i is not None and n_c != n_i: powody["series_number_mismatch_reject"] = [n_c, n_i] return ocena, True, powody if (n_c is None) != (n_i is None): powody["series_number_one_sided_cap"] = n_c if n_c is not None else n_i return min(ocena, _ONE_SIDED_NUMBER_CAP), False, powody # Podzbior z PRAWDZIWYMI dodatkowymi slowami to inny film, nie lustro. Po # normalizacji "The Doctor" zostaje samym "doctor", a jedno slowo jest podzbiorem # kazdego tytulu z tym slowem: "granny s doctor games", "we need a doctor", "down on # the doctor" dostawaly 100 i lecialy do kotwicy (rozdzielanie 2026-09-02). Wyjatek to # przedrostki SEO luster ("watch cece adult" = "cece"), stad najpierw zdejmujemy # slownik wypelniaczy i dopiero porownujemy zbiory slow. la, li = _slowa(cand_title), _slowa(inc_title) ta, ti = frozenset(la), frozenset(li) if not ta or not ti: # Po zdjeciu wypelniaczy jedna strona jest pusta: nie ma czego porownywac # zbiorami, zostaje surowy token_set. return ocena, False, powody if not any(t.isalpha() for t in ta | ti): # Zostaly same liczby ("adult movie 2019" vs "porn film 2019" -> {2019} == {2019}). # Rok to nie tytul, wiec rownosc takich zbiorow niczego nie dowodzi. powody["no_words_cap"] = sorted(ta | ti)[:4] return min(ocena, _ONE_SIDED_NUMBER_CAP), False, powody if ta == ti or "".join(la) == "".join(li): # Te same slowa (po zdjeciu wypelniaczy, sklejeniu apostrofow i rozwinieciu # skrotow) albo ten sam ciag po zdjeciu spacji ("gang bang" vs "gangbang"): # prawdziwe lustro, niezaleznie od kolejnosci i przedrostkow SEO. return 1.0, False, powody # Literowka albo liczba mnoga: slowa wlasne kazdej strony musza sie sparowac jeden do # jednego z podobienstwem >= 0.85 na poziomie SLOWA. Porownanie calych posortowanych # ciagow bylo za luzne dla krotkich tytulow: "girls only 4" vs "girls on girls 4" # dawalo 0.91, bo zbior zwija powtorzone "girls", a "only"/"on" to dwie litery. prawie = _parowanie_slow(ta - ti, ti - ta) if prawie is not None: powody["fuzzy_words"] = round(prawie, 3) return prawie, False, powody # Podzbior z prawdziwymi dodatkowymi slowami ("doctor" vs "grannys doctor games") albo # slowa wlasne po obu stronach: dla filmow to inny tytul, nie lustro. `token_set_ratio` # dawal tu 85-100 i przy zgodnym studiu i roku wychodzil auto-merge. klucz = "subset_title_cap" if (ta < ti or ti < ta) else "distinct_words_cap" powody[klucz] = sorted(ta ^ ti)[:4] return min(ocena, _ONE_SIDED_NUMBER_CAP), False, powody # Literowka: dwa slowa od 4 liter o podobienstwie >= 0.92 ("fuking"/"fucking" 0.92). # Prog 0.85 byl za nisko: "sexy"/"sex" 0.86, "intense"/"interns" 0.86, # "infedele"/"fedele" 0.86 (niewierna vs wierna!) laczyly rozne filmy. _FUZZY_WORD_MIN = 0.92 _FUZZY_WORD_MIN_LEN = 4 def _to_samo_slowo(a: str, b: str) -> float | None: """Ocena pary slow uznanych za to samo slowo, albo None. Liczba pojedyncza/mnoga ("slut"/"sluts", "date"/"dates") to 0.95: w tytulach luster to najczestsza roznica. Reszta tylko przez bardzo bliska literowke. """ if a == b: return 1.0 if len(a) < len(b): a, b = b, a if a in (b + "s", b + "es") and len(b) >= 3: return 0.95 if len(a) >= _FUZZY_WORD_MIN_LEN and len(b) >= _FUZZY_WORD_MIN_LEN: r = fuzz.ratio(a, b) / 100.0 if r >= _FUZZY_WORD_MIN: return r return None def _parowanie_slow(tylko_a: frozenset[str], tylko_b: frozenset[str]) -> float | None: """Najslabsze z parowan slow wlasnych obu stron, albo None, gdy sie nie paruja. Kazde slowo wlasne jednej strony musi miec odpowiednik po drugiej (ta sama liczba slow, kazda para wg `_to_samo_slowo`). Wynik to najslabsza para, wiec tytul dostaje ocene dokladnie tak dobra, jak najgorsza literowka. """ if not tylko_a or not tylko_b or len(tylko_a) != len(tylko_b): return None wolne = set(tylko_b) najslabsza = 1.0 for a in sorted(tylko_a): pary = [(b, r) for b in wolne if (r := _to_samo_slowo(a, b)) is not None] if not pary: return None b, r = max(pary, key=lambda x: x[1]) wolne.discard(b) najslabsza = min(najslabsza, r) return najslabsza # Ogonki, ktore `normalize()` odcina od slowa przy apostrofie: "she's" -> "she s", # "gangbang'd" -> "gangbang d". Tylko te sklejamy z powrotem; "a" i "i" to osobne slowa. _OGONKI_APOSTROFU = frozenset({"s", "t", "d"}) # Skroty spotykane w tytulach filmow, rozwijane przed porownaniem zbiorow slow. _SKROTY = {"vol": "volume", "pt": "part", "ep": "episode", "chap": "chapter", "no": "number"} def _slowa(title_normalized: str) -> list[str]: """Slowa tytulu do porownania, w kolejnosci z tytulu. Bez wypelniaczy SEO, z doklejonymi ogonkami apostrofu ("she s" -> "shes") i z rozwinietymi skrotami ("vol" -> "volume"). Pierwsza wersja sklejala KAZDA pojedyncza litere z poprzednim slowem, wiec "watch a whore s tale" dawalo "watcha", a przedimek "a" nigdy nie trafial do wypelniaczy: lustra z "A"/"I" na poczatku tytulu przestawaly sie laczyc (code review 2026-09-02). """ out: list[str] = [] for tok in title_normalized.split(): if tok in _OGONKI_APOSTROFU and out: out[-1] = out[-1] + tok else: out.append(_SKROTY.get(tok, tok)) return [t for t in out if t not in _SEO_WYPELNIACZE] def _same_studio_name(session: Session, a_id: uuid.UUID, b_id: uuid.UUID) -> bool: a, b = session.get(Studio, a_id), session.get(Studio, b_id) if a is None or b is None or not a.name_normalized or not b.name_normalized: return False ka, kb = a.name_normalized.replace(" ", ""), b.name_normalized.replace(" ", "") return len(ka) >= 4 and ka == kb def score_movie_candidate( session: Session, *, candidate: Movie, norm_title_normalized: str, norm_release_year: int | None, norm_studio_id: uuid.UUID | None, norm_performer_ids: Iterable[uuid.UUID], ) -> MovieScoreBreakdown: """Liczy score dopasowania incoming → candidate. Wszystko w [0,1].""" breakdown = MovieScoreBreakdown(reasons={}) # Title (always). Cyfry rzymskie jako arabskie: "hardcore gangbangs viii" i "... 8" to # ten sam numer serii (zgloszenie f8c9486e, xtapes vs inne zrodla). series_reject = False if candidate.title_normalized and norm_title_normalized: breakdown.title, series_reject, powody = _title_signal( _roman_to_arabic(candidate.title_normalized), _roman_to_arabic(norm_title_normalized) ) breakdown.reasons.update(powody) # Year, exact lub ±1 = wysoka, ±2 = średnia if candidate.release_year is not None and norm_release_year is not None: delta = abs(candidate.release_year - norm_release_year) if delta == 0: breakdown.year = 1.0 elif delta == 1: breakdown.year = 0.7 elif delta == 2: breakdown.year = 0.4 else: breakdown.year = 0.0 # Studio match, binary. Rozne wiersze o tej samej nazwie bez spacji ("EvilAngel" z xtapes # i "Evil Angel" z TPDB) to to samo studio; w bazie jest ok. 2 tys. takich par. if candidate.studio_id and norm_studio_id: breakdown.studio_match = candidate.studio_id == norm_studio_id or _same_studio_name( session, candidate.studio_id, norm_studio_id ) # Cast Jaccard, gdy obie strony mają performerów norm_perf_set = {str(pid) for pid in norm_performer_ids if pid is not None} if norm_perf_set: cand_perf_rows = session.execute( MoviePerformer.__table__.select().where( MoviePerformer.movie_id == candidate.id ) ).fetchall() cand_perf_set = {str(r.performer_id) for r in cand_perf_rows} if cand_perf_set: inter = norm_perf_set & cand_perf_set union = norm_perf_set | cand_perf_set breakdown.cast = len(inter) / len(union) if union else 0.0 # Composite, redystrybuuj wagi po dostępnych sygnałach available: dict[str, float] = {} if breakdown.title is not None: available["title"] = breakdown.title if breakdown.year is not None: available["year"] = breakdown.year if breakdown.studio_match is not None: available["studio"] = 1.0 if breakdown.studio_match else 0.0 if breakdown.cast is not None: available["cast"] = breakdown.cast if not available: breakdown.composite = 0.0 breakdown.reasons["no_signals"] = True return breakdown weights = {k: _BASE_WEIGHTS[k] for k in available} total_w = sum(weights.values()) if total_w == 0.0: breakdown.composite = 0.0 return breakdown norm_w = {k: w / total_w for k, w in weights.items()} score = sum(available[k] * norm_w[k] for k in available) # Hard reject: studio mismatch + year mismatch (delta ≥3) → różne filmy z podobnym # tytułem. "Tushy Anal Bliss 7" vs "Anal Bliss 7" z innego studia rok różny, to NIE # jest mirror, to różne wydania. if ( breakdown.studio_match is False and breakdown.year is not None and breakdown.year < 0.4 ): breakdown.reasons["studio_year_mismatch_reject"] = True score = 0.0 # Hard reject: rozne numery serii. Bez tego "Her way to use Tinder 61" wchlaniala # "... Tinder 60" (token_set 96%), az pod jednym tytulem siedzialy 102 strony. if series_reject: score = 0.0 breakdown.composite = max(0.0, min(1.0, score)) breakdown.reasons["weights"] = norm_w breakdown.reasons["sub_scores"] = available return breakdown def triage_movie(score: float) -> str: if score >= AUTO_THRESHOLD: return "auto" if score >= REVIEW_THRESHOLD: return "review" return "reject"