feat(movies): rozdzielanie zlepionych filmow + sufit dla tytulu-podzbioru w scorerze

Skrypt `scripts/movie_unmerge.py` rozdziela filmy, ktore stary scorer zlepil z kilku
tytulow (8204 filmy z >=2 refami z jednego zrodla, 31 299 refow). Musi pobierac strony
na nowo, bo rekordy zewnetrzne filmow trzymaja tylko {slug, html_len} / {hex_id,
html_len}, a ingest jest przyrostowy, wiec skasowane rekordy same sie nie odbuduja.

Dla kazdego filmu: refy pobierane przez konektory zrodel (paradisehill, dooplay x4,
freeomovie; reszta profil ze sluga), klastry union-find po wspolnym adresie embedu
(bez prefiksu zrodla, ten sam plik siedzi pod mangoporn/pandamovies/streamporn) albo
po nowej ocenie zlozonej >= 0.85. Klaster z najstarszym refem zostaje filmem
(zachowuje id, ulubione, postep), kazdy inny dostaje nowy film z refami, hosterami,
obsada, tagami i rozdzialami. Kotwica zostaje z hosterami potwierdzonymi przez swieze
strony; reszta z rodzin luster leci, bo pandamovies zdazyl zrotowac hostery (swieza
strona ma 0 z 4 kluczy z bazy), wiec to i tak zwietrzale linki.

Proba na "The Doctor" ujawnila druga luke scorera: po normalizacji zostaje samo
"doctor", a jedno slowo jest podzbiorem kazdego tytulu z tym slowem, wiec "granny s
doctor games" i "we need a doctor" dostawaly 100 i szly do kotwicy. Teraz podzbior z
prawdziwymi dodatkowymi slowami ma sufit 0.6, a przedrostki SEO luster ("watch cece
adult movie online free") sa zdejmowane przed testem, zeby prawdziwe lustra nadal sie
laczyly. Po poprawce "The Doctor" rozpada sie na 11 klastrow, kotwica to sam
paradisehill, a "Doctor Adventures 14" laczy swoje cztery lustra.

app/resolve/movie_score.py

@@ -77,6 +77,13 @@ def _series_number(title_normalized: str) -> int | None:
 # 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).
@@ -104,6 +111,17 @@ def _title_signal(cand_title: str, inc_title: str) -> tuple[float, bool, dict]:
     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.
+    ta = set(cand_title.split()) - _SEO_WYPELNIACZE
+    ti = set(inc_title.split()) - _SEO_WYPELNIACZE
+    if ta and ti and ta != ti and (ta < ti or ti < ta):
+        powody["subset_title_cap"] = sorted((ta ^ ti))[:4]
+        return min(ocena, _ONE_SIDED_NUMBER_CAP), False, powody
     return ocena, False, powody
 
 

scripts/movie_unmerge.py

@@ -0,0 +1,424 @@
+"""Rozdziela filmy zlepione z kilku roznych tytulow (skutek starego scorera).
+
+Tlo: `movie_score` do 2026-09-02 mierzyl tytul przez `token_set_ratio`, ktore daje pelne
+100 dla tytulu bedacego podzbiorem drugiego i 90-96 dla numerowanych serii. Efekt:
+"The Doctor" = 9 filmow z pandamovies pod jednym wpisem (zgloszenie 32100b94, "w
+pierwszym linku granny porn"), "Her way to use Tinder 61" = 102 strony. W bazie 8204
+filmy maja >= 2 refy z tego samego zrodla, lacznie 31 299 refow.
+
+Dlaczego to musi pobierac strony na nowo: rekordy zewnetrzne filmow trzymaja tylko
+`{slug, html_len}` (dooplay) albo `{hex_id, html_len}` (paradisehill), bez tytulu, roku
+ani listy hosterow, a ingest filmow jest przyrostowy (paradisehill `since`, dooplay 3
+strony), wiec skasowane rekordy nie odbuduja sie same. Jedyne zrodlo prawdy o tym, ktory
+hoster nalezy do ktorej strony, to sama strona.
+
+Algorytm dla kazdego podejrzanego filmu M:
+1. Kazdy ref pobieramy na nowo przez konektor zrodla -> RawMovie (tytul, rok, studio,
+   obsada, hostery). Zrodla bez konektora (tpdb, tube-scraper, yourdailypornmovies) i
+   nieudane pobrania dostaja profil z samego sluga adresu (tylko tytul).
+2. Refy laczymy w klastry (union-find): ten sam plik u hostera (wspolny adres embedu,
+   BEZ prefiksu zrodla, bo ten sam plik siedzi pod mangoporn, pandamovies i streamporn)
+   ALBO nowa ocena zlozona >= 0.85 z regulami numeru serii z `movie_score`.
+3. Klaster z najstarszym refem zostaje filmem M (zachowuje id, ulubione i postep).
+   Kazdy inny klaster dostaje nowy film; jego refy, hostery (po adresie embedu),
+   obsada, tagi i rozdzialy przechodza tam.
+4. M zostaje z hosterami potwierdzonymi przez swieze strony kotwicy. Reszta wierszy z
+   rodzin luster (mangoporn/pandamovies/streamporn/paradisehill/freeomovie) leci, bo
+   albo nalezy do odczepionego filmu, albo jest zwietrzala (pandamovies zrotowal
+   hostery: swieza strona "Granny's Doctor Games" ma 0 z 4 kluczy w bazie). Wiersze z
+   innych origin (kanoniczne strony studiow) zostaja. Obsada, tagi i rozdzialy M sa
+   budowane od nowa z profili kotwicy, o ile choc jeden udalo sie pobrac.
+
+Bezpieczniki: `--dry-run` tylko liczy i wypisuje klastry; `--movie-id` na jeden film;
+`--after-id` do wznowienia (przebieg idzie ORDER BY id). Odpalaj z pliku, w osobnym
+kontenerze (`docker compose run -d --rm ... python3 -u /srv/scripts/movie_unmerge.py`),
+nie przez stdin (patrz browse_catchup.py) i nie w kontenerze workera.
+
+Uzycie:
+    python3 -u /srv/scripts/movie_unmerge.py --movie-id 6283c6cb-... --dry-run
+    python3 -u /srv/scripts/movie_unmerge.py --limit 50 --dry-run
+    python3 -u /srv/scripts/movie_unmerge.py
+"""
+from __future__ import annotations
+
+import argparse
+import logging
+import re
+import sys
+import time
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime
+from urllib.parse import urlparse
+
+from sqlalchemy import text
+
+from app.connectors.base import RawMovie
+from app.connectors.dooplay import (
+    MangopornConnector,
+    PandamoviesConnector,
+    StreampornConnector,
+    StreampornVipConnector,
+)
+from app.connectors.freeomovie import FreeoMovieConnector
+from app.connectors.paradisehill import ParadisehillConnector
+from app.db import session_scope
+from app.models.movie import Movie, MovieChapter, MoviePerformer, MovieTag
+from app.normalize.movies import NormalizedMovie, normalize_movie
+from app.normalize.text import normalize, slugify
+from app.resolve.movie_resolver import (
+    _create_canonical,
+    _sync_chapters,
+    _sync_performers,
+    _sync_playback_sources,
+    _sync_tags,
+    _update_movie_fields,
+)
+from app.resolve.movie_score import _title_signal
+from app.resolve.performer_resolver import resolve_performer
+from app.resolve.studio_resolver import resolve_studio
+
+log = logging.getLogger("movie-unmerge")
+
+_SPACING_SEC = 0.4
+_AUTO = 0.85
+_WAGI = {"title": 0.50, "year": 0.20, "studio": 0.20, "cast": 0.10}
+
+# Rodziny origin, ktorych wiersze na filmie-kotwicy wolno skasowac, gdy zadna swieza
+# strona ich nie potwierdza. Inne origin (np. kanoniczne strony studiow) zostaja.
+_RODZINY_LUSTER = ("mangoporn", "pandamovies", "streamporn", "streampornvip", "paradisehill", "freeomovie")
+
+_SUFIKSY = (
+    "-adult-movie-online-free", "-movie-online-free", "-porn-movie", "-online-free",
+)
+
+
+@dataclass
+class Profil:
+    source_name: str
+    source_id: uuid.UUID
+    external_id: str
+    url: str | None
+    first_seen: datetime
+    raw: RawMovie | None = None
+    norm: NormalizedMovie | None = None
+    title_norm: str = ""
+    year: int | None = None
+    studio_norm: str | None = None
+    cast: set[str] = field(default_factory=set)
+    keys: set[str] = field(default_factory=set)   # adresy embedow / stron hosterow
+    fetched: bool = False
+
+    @property
+    def klucz(self) -> tuple[uuid.UUID, str]:
+        return (self.source_id, self.external_id)
+
+
+def _tytul_ze_sluga(url: str | None, external_id: str) -> str:
+    seg = ""
+    if url:
+        seg = urlparse(url).path.rstrip("/").split("/")[-1]
+    if not seg or re.fullmatch(r"[0-9a-f]{8,}", seg):
+        seg = external_id.split(":")[-1]
+    seg = seg.lower()
+    for pre in ("watch-xxx-", "watch-"):
+        if seg.startswith(pre):
+            seg = seg[len(pre):]
+    for suf in _SUFIKSY:
+        if seg.endswith(suf):
+            seg = seg[: -len(suf)]
+    # `.../2768634/supermodel-sluts-9-po` -> bez numerycznego id na poczatku
+    seg = re.sub(r"^\d{5,}-", "", seg)
+    return normalize(seg.replace("-", " ").replace("_", " "))
+
+
+_KONEKTORY: dict = {}
+
+
+def _konektor(name: str):
+    if name not in _KONEKTORY:
+        klasy = {
+            "paradisehill": ParadisehillConnector,
+            "mangoporn": MangopornConnector,
+            "pandamovies": PandamoviesConnector,
+            "streamporn": StreampornConnector,
+            "streampornvip": StreampornVipConnector,
+            "freeomovie": FreeoMovieConnector,
+        }
+        _KONEKTORY[name] = klasy[name]() if name in klasy else None
+    return _KONEKTORY[name]
+
+
+def _pobierz(p: Profil) -> RawMovie | None:
+    conn = _konektor(p.source_name)
+    if conn is None:
+        return None
+    time.sleep(_SPACING_SEC)
+    if p.source_name == "paradisehill":
+        return conn._fetch_detail(p.external_id)
+    if p.source_name == "freeomovie":
+        if not p.url:
+            return None
+        return conn._parse_detail(p.url, _tytul_ze_sluga(p.url, p.external_id).title())
+    url = p.url or f"{conn.base_url.rstrip('/')}/movies/{p.external_id}/"
+    return conn._fetch_detail(url)
+
+
+def _profil(p: Profil) -> Profil:
+    try:
+        raw = _pobierz(p)
+    except Exception as e:  # 404, timeout, CF: profil z samego sluga
+        log.info("  %s %s: pobranie padlo (%s), profil ze sluga", p.source_name, p.external_id, type(e).__name__)
+        raw = None
+    if raw is not None:
+        p.raw = raw
+        p.norm = normalize_movie(raw)
+        p.title_norm = p.norm.title_normalized
+        p.year = p.norm.release_year
+        p.studio_norm = p.norm.studio.name_normalized if p.norm.studio else None
+        p.cast = {q.canonical_name.lower() for q in p.norm.performers if q.canonical_name}
+        p.keys = {s.page_url for s in raw.playback_sources if s.page_url}
+        p.fetched = True
+    else:
+        p.title_norm = _tytul_ze_sluga(p.url, p.external_id)
+    if p.url:
+        p.keys.add(p.url)   # wiersz "landing" ma page_url == adres strony
+    return p
+
+
+def _ten_sam_film(a: Profil, b: Profil) -> bool:
+    """Odbicie `score_movie_candidate` w pamieci, z tymi samymi wagami i odrzuceniami."""
+    if a.keys & b.keys & _bez_landingow(a, b):
+        return True
+    if not a.title_norm or not b.title_norm:
+        return False
+    ocena, odrzut, _ = _title_signal(a.title_norm, b.title_norm)
+    if odrzut:
+        return False
+    dost = {"title": ocena}
+    rok = None
+    if a.year and b.year:
+        d = abs(a.year - b.year)
+        rok = 1.0 if d == 0 else 0.7 if d == 1 else 0.4 if d == 2 else 0.0
+        dost["year"] = rok
+    studio = None
+    if a.studio_norm and b.studio_norm:
+        studio = a.studio_norm == b.studio_norm
+        dost["studio"] = 1.0 if studio else 0.0
+    if a.cast and b.cast:
+        dost["cast"] = len(a.cast & b.cast) / len(a.cast | b.cast)
+    if studio is False and rok is not None and rok < 0.4:
+        return False
+    suma_w = sum(_WAGI[k] for k in dost)
+    wynik = sum(v * _WAGI[k] / suma_w for k, v in dost.items())
+    return wynik >= _AUTO
+
+
+def _bez_landingow(a: Profil, b: Profil) -> set[str]:
+    """Wspolne klucze licza sie tylko dla hosterow, nie dla adresow stron (te sa unikalne)."""
+    return (a.keys | b.keys) - {a.url, b.url}
+
+
+def _klastry(profile: list[Profil]) -> list[list[Profil]]:
+    rodzic = list(range(len(profile)))
+
+    def find(i):
+        while rodzic[i] != i:
+            rodzic[i] = rodzic[rodzic[i]]
+            i = rodzic[i]
+        return i
+
+    for i in range(len(profile)):
+        for j in range(i + 1, len(profile)):
+            if find(i) != find(j) and _ten_sam_film(profile[i], profile[j]):
+                rodzic[find(j)] = find(i)
+    grupy: dict[int, list[Profil]] = {}
+    for i, p in enumerate(profile):
+        grupy.setdefault(find(i), []).append(p)
+    return list(grupy.values())
+
+
+def _wzorzec(klaster: list[Profil]) -> Profil | None:
+    """Profil, z ktorego bierzemy pola kanoniczne: paradisehill, potem pierwszy pobrany."""
+    for p in klaster:
+        if p.fetched and p.source_name == "paradisehill":
+            return p
+    for p in klaster:
+        if p.fetched:
+            return p
+    return None
+
+
+def _opis(klaster: list[Profil]) -> str:
+    return "; ".join(f"{p.source_name}:{p.title_norm[:34]}" for p in klaster[:6]) + (" ..." if len(klaster) > 6 else "")
+
+
+def _zaladuj_refy(s, movie_id: uuid.UUID) -> list[Profil]:
+    rows = s.execute(
+        text(
+            """
+            SELECT src.name, r.source_id, r.external_id, r.url, r.first_seen
+              FROM movie_external_refs r JOIN sources src ON src.id = r.source_id
+             WHERE r.movie_id = :m ORDER BY r.first_seen
+            """
+        ),
+        {"m": str(movie_id)},
+    ).fetchall()
+    return [Profil(source_name=n, source_id=sid, external_id=e, url=u, first_seen=fs) for n, sid, e, u, fs in rows]
+
+
+def _przypnij_zawartosc(s, movie_id: uuid.UUID, p: Profil) -> None:
+    """Obsada, tagi, rozdzialy i hostery z profilu na wskazany film (jak w resolverze)."""
+    assert p.norm is not None
+    resolved = []
+    for pn in p.norm.performers:
+        perf = resolve_performer(s, norm=pn, source_id=p.source_id)
+        resolved.append((perf.id, pn.as_alias_in_scene))
+    _sync_performers(s, movie_id=movie_id, resolved=resolved)
+    _sync_tags(s, movie_id=movie_id, norm=p.norm, source_id=p.source_id)
+    _sync_chapters(s, movie_id=movie_id, norm=p.norm)
+    _sync_playback_sources(s, movie_id=movie_id, norm=p.norm)
+
+
+def _przetworz(movie_id: uuid.UUID, *, dry_run: bool, liczniki: dict) -> None:
+    with session_scope() as s:
+        m = s.get(Movie, movie_id)
+        if m is None:
+            return
+        tytul_m = m.title
+        profile = _zaladuj_refy(s, movie_id)
+    if len(profile) < 2:
+        return
+    profile = [_profil(p) for p in profile]
+    klastry = _klastry(profile)
+    klastry.sort(key=lambda k: min(p.first_seen for p in k))
+    kotwica, inne = klastry[0], klastry[1:]
+    liczniki["filmy"] += 1
+    liczniki["pobrane"] += sum(1 for p in profile if p.fetched)
+    liczniki["niepobrane"] += sum(1 for p in profile if not p.fetched)
+    if not inne:
+        log.info("OK   %-40s refy=%d, jeden klaster, bez zmian", tytul_m[:40], len(profile))
+        return
+    liczniki["rozdzielone"] += 1
+    liczniki["nowe_filmy"] += len(inne)
+    log.info("SPLIT %-40s refy=%d -> %d klastrow", tytul_m[:40], len(profile), len(klastry))
+    log.info("   kotwica: %s", _opis(kotwica))
+    for k in inne:
+        log.info("   nowy:    %s", _opis(k))
+    if dry_run:
+        return
+
+    with session_scope() as s:
+        m = s.get(Movie, movie_id)
+        assert m is not None
+        # --- nowe filmy dla klastrow poza kotwica --------------------------------
+        for k in inne:
+            wz = _wzorzec(k)
+            if wz is not None and wz.norm is not None:
+                studio = resolve_studio(s, norm=wz.norm.studio, source_id=wz.source_id) if wz.norm.studio else None
+                nowy = _create_canonical(s, norm=wz.norm, studio_id=studio.id if studio else None)
+            else:
+                # Zaden ref klastra nie dal sie pobrac: film z samego tytulu ze sluga,
+                # zeby refy nie zostaly na zlym filmie. Reszta doladuje sie przy ingescie.
+                t = (k[0].title_norm or "untitled").title()
+                nowy = Movie(title=t, title_normalized=normalize(t), slug=slugify(t))
+                s.add(nowy)
+                s.flush()
+            klucze = set().union(*(p.keys for p in k))
+            for p in k:
+                s.execute(
+                    text("UPDATE movie_external_refs SET movie_id=:n WHERE source_id=:sid AND external_id=:e"),
+                    {"n": str(nowy.id), "sid": str(p.source_id), "e": p.external_id},
+                )
+            if klucze:
+                r = s.execute(
+                    text("UPDATE movie_playback_sources SET movie_id=:n WHERE movie_id=:m AND page_url = ANY(:k)"),
+                    {"n": str(nowy.id), "m": str(movie_id), "k": list(klucze)},
+                )
+                liczniki["hostery_przeniesione"] += r.rowcount
+            for p in k:
+                if p.norm is not None:
+                    _przypnij_zawartosc(s, nowy.id, p)
+            liczniki["refy_przeniesione"] += len(k)
+            log.info("   -> nowy film %s %r", nowy.id, nowy.title)
+
+        # --- kotwica: zostaje to, co potwierdzaja swieze strony ------------------
+        klucze_k = set().union(*(p.keys for p in kotwica))
+        wzory = [p for p in kotwica if p.norm is not None]
+        r = s.execute(
+            text(
+                """
+                DELETE FROM movie_playback_sources
+                 WHERE movie_id = :m AND NOT (page_url = ANY(:k))
+                   AND split_part(origin, ':', 1) = ANY(:rodz)
+                """
+            ),
+            {"m": str(movie_id), "k": list(klucze_k) or [""], "rodz": list(_RODZINY_LUSTER)},
+        )
+        liczniki["hostery_skasowane"] += r.rowcount
+        if wzory:
+            for tabela in (MoviePerformer, MovieTag, MovieChapter):
+                s.query(tabela).filter(tabela.movie_id == movie_id).delete(synchronize_session=False)
+            wz = _wzorzec(kotwica)
+            assert wz is not None and wz.norm is not None
+            studio = resolve_studio(s, norm=wz.norm.studio, source_id=wz.source_id) if wz.norm.studio else None
+            m.title = wz.norm.title
+            m.title_normalized = wz.norm.title_normalized
+            m.slug = wz.norm.slug or slugify(wz.norm.title)
+            m.release_year = wz.norm.release_year
+            if studio is not None:
+                m.studio_id = studio.id
+            _update_movie_fields(m, wz.norm, studio_id=studio.id if studio else None)
+            for p in wzory:
+                _przypnij_zawartosc(s, movie_id, p)
+        log.info("   kotwica zostaje jako %r (hostery skasowane: %d)", m.title, r.rowcount)
+
+
+def main(argv: list[str] | None = None) -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--movie-id", default=None)
+    ap.add_argument("--limit", type=int, default=0)
+    ap.add_argument("--after-id", default=None)
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+
+    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", stream=sys.stdout)
+    for noisy in ("httpx", "httpcore", "urllib3", "curl_cffi"):
+        logging.getLogger(noisy).setLevel(logging.WARNING)
+
+    if a.movie_id:
+        ids = [uuid.UUID(a.movie_id)]
+    else:
+        sql = """
+            SELECT DISTINCT movie_id FROM movie_external_refs
+             WHERE movie_id IN (SELECT movie_id FROM movie_external_refs
+                                 GROUP BY movie_id, source_id HAVING count(*) >= 2)
+        """
+        if a.after_id:
+            sql += " AND movie_id > :aid"
+        sql += " ORDER BY movie_id"
+        if a.limit:
+            sql += f" LIMIT {int(a.limit)}"
+        with session_scope() as s:
+            ids = [r[0] for r in s.execute(text(sql), {"aid": a.after_id} if a.after_id else {}).fetchall()]
+    log.info("filmow do sprawdzenia: %d%s", len(ids), " (DRY RUN)" if a.dry_run else "")
+
+    liczniki = {k: 0 for k in ("filmy", "rozdzielone", "nowe_filmy", "refy_przeniesione", "hostery_przeniesione", "hostery_skasowane", "pobrane", "niepobrane")}
+    t0 = time.time()
+    for i, mid in enumerate(ids, 1):
+        try:
+            _przetworz(mid, dry_run=a.dry_run, liczniki=liczniki)
+        except Exception:
+            log.exception("film %s: blad, pomijam", mid)
+            liczniki.setdefault("bledy", 0)
+            liczniki["bledy"] += 1
+        if i % 25 == 0 or i == len(ids):
+            log.info("[%d/%d] %s (%.0f min, ostatni id %s)", i, len(ids), liczniki, (time.time() - t0) / 60, mid)
+            for h in logging.getLogger().handlers:
+                h.flush()
+    log.info("KONIEC: %s | %.0f min", liczniki, (time.time() - t0) / 60)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())