fix(unmerge): szukaj istniejacego filmu przed utworzeniem, sprzataj duble, dziedzicz created_at

Trzy skutki pierwszego przebiegu rozdzielania (2026-09-02, 5846 nowych filmow), oba
zgloszone z apki tego samego dnia:

1. Duble (a5cdda7f). Klaster poza kotwica dostawal ZAWSZE nowy film, choc "Control 2"
   z innego zrodla juz istnial: 798 nowych o tytule sprzed przebiegu, 408 dubli miedzy
   soba. Teraz najpierw `find_movie_candidates` + `score_movie_candidate` (ten sam
   resolver co ingest) i przy >= 0.85 refy ida do istniejacego. Tryb
   `--dedupe-created-after <ts>` scala juz utworzone przez `merge_movies` (691 scalonych).

2. Scorer przepuszczal jeszcze pary z roznymi slowami po obu stronach: "hardcore
   fucking sluts 2" vs "foot fucking sluts 2", "santa list gangbang" vs "gangbang d"
   (token_set 85-94, ze zgodnym studiem i rokiem auto-merge). Lustra nie zmieniaja
   tytulow, wiec: rowne zbiory slow po zdjeciu wypelniaczy SEO = 1.0, podzbior albo
   rozne slowa po obu stronach = sufit 0.6. Ogonki po apostrofie sklejane ("she s" ->
   "shes"), bo normalize() tnie apostrof na spacje. Lista dubli spadla z 941 do 691 i
   sa to same identyczne tytuly.

3. Falszywe "+N nowych" (1db95176). Nowy film dostawal created_at = teraz, a badge w
   ulubionych od lipca sumuje sceny i FILMY: 2196 filmow z obsada z dzisiejsza data
   zapalilo "+N" u kazdego z performerem w ulubionych, przy zerze nowych scen na liscie.
   Teraz created_at = pierwsze zobaczenie refow klastra; 5231 juz utworzonych
   poprawione w bazie tym samym kryterium.

app/resolve/movie_score.py

@@ -117,14 +117,38 @@ def _title_signal(cand_title: str, inc_title: str) -> tuple[float, bool, dict]:
     # 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
+    ta = _slowa(cand_title)
-    ti = set(inc_title.split()) - _SEO_WYPELNIACZE
+    ti = _slowa(inc_title)
-    if ta and ti and ta != ti and (ta < ti or ti < ta):
+    if ta and ti and ta == ti:
+        # Te same slowa (po zdjeciu wypelniaczy i sklejeniu apostrofow): prawdziwe lustro,
+        # niezaleznie od kolejnosci i przedrostkow SEO.
+        return 1.0, False, powody
+    if ta and ti and (ta < ti or ti < ta):
         powody["subset_title_cap"] = sorted((ta ^ ti))[:4]
         return min(ocena, _ONE_SIDED_NUMBER_CAP), False, powody
+    if ta and ti:
+        # Kazda strona ma slowo, ktorego druga nie ma: dla filmow to prawie zawsze inny
+        # tytul, a nie literowka. `token_set_ratio` dawal tu 85-94 ("hardcore fucking sluts
+        # 2" vs "foot fucking sluts 2", "santa list gangbang" vs "gangbang d") i przy
+        # zgodnym studiu i roku wychodzil auto-merge (sprzatanie dubli 2026-09-02).
+        powody["distinct_words_cap"] = sorted(ta ^ ti)[:4]
+        return min(ocena, _ONE_SIDED_NUMBER_CAP), False, powody
     return ocena, False, powody
 
 
+def _slowa(title_normalized: str) -> frozenset[str]:
+    """Zbior slow tytulu do porownania: bez wypelniaczy SEO, z doklejonymi ogonkami po
+    apostrofie ("she s" -> "shes", "gangbang d" -> "gangbangd"), bo `normalize()` tnie
+    apostrof na spacje i "She's In Control 3" rozjezdzalo sie z "Shes In Control 3"."""
+    out: list[str] = []
+    for tok in title_normalized.split():
+        if len(tok) == 1 and tok.isalpha() and out:
+            out[-1] = out[-1] + tok
+        else:
+            out.append(tok)
+    return frozenset(out) - _SEO_WYPELNIACZE
+
+
 def score_movie_candidate(
     session: Session,
     *,

scripts/movie_unmerge.py

@@ -253,6 +253,47 @@ 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 _istniejacy_film(
+    s, *, norm: NormalizedMovie, studio_id, performer_names: set[str], wyklucz: set,
+    tylko_starsze_niz: datetime | None = None,
+) -> Movie | None:
+    """Film, do ktorego resolver dolaczylby ten tytul automatycznie (>= 0.85), albo None.
+
+    Ta sama para `find_movie_candidates` + `score_movie_candidate`, ktorej uzywa ingest,
+    wiec decyzja jest identyczna z ta, ktora zapadlaby przy normalnym wejsciu sceny.
+    Obsada porownywana po nazwach, bo w tym miejscu nie mamy jeszcze performer_id.
+    """
+    from app.resolve.movie_match import find_movie_candidates
+    from app.resolve.movie_score import AUTO_THRESHOLD, score_movie_candidate
+
+    kand = find_movie_candidates(
+        s, title_normalized=norm.title_normalized, studio_id=studio_id, release_year=norm.release_year
+    )
+    najlepszy, najlepsza = None, 0.0
+    for c in kand:
+        if c.id in wyklucz:
+            continue
+        if tylko_starsze_niz is not None and c.created_at and c.created_at >= tylko_starsze_niz:
+            continue
+        pids = [
+            pid for (pid,) in s.execute(
+                text(
+                    "SELECT mp.performer_id FROM movie_performers mp JOIN performers p ON p.id = mp.performer_id "
+                    " WHERE mp.movie_id = :m AND lower(p.canonical_name) = ANY(:n)"
+                ),
+                {"m": str(c.id), "n": list(performer_names) or [""]},
+            ).fetchall()
+        ] if performer_names else []
+        # Do Jaccarda podajemy tylko to, co sie zgadza po nazwie, wiec dolny szacunek.
+        br = score_movie_candidate(
+            s, candidate=c, norm_title_normalized=norm.title_normalized,
+            norm_release_year=norm.release_year, norm_studio_id=studio_id, norm_performer_ids=pids,
+        )
+        if br.composite > najlepsza:
+            najlepszy, najlepsza = c, br.composite
+    return najlepszy if najlepszy is not None and najlepsza >= AUTO_THRESHOLD else None
+
+
 def _zaladuj_refy(s, movie_id: uuid.UUID) -> list[Profil]:
     rows = s.execute(
         text(
@@ -321,6 +362,19 @@ def _przetworz(movie_id: uuid.UUID, *, dry_run: bool, liczniki: dict) -> None:
             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
+                # NAJPIERW istniejacy film, tym samym resolverem co ingest. Bez tego
+                # pierwszy przebieg (2026-09-02) utworzyl 798 filmow o tytule, ktory juz
+                # byl w bazie, i 408 dubli miedzy soba: "Control 2" powstawal na nowo,
+                # chociaz "Control 2" z innego zrodla juz istnial (zgloszenie a5cdda7f).
+                istn = _istniejacy_film(
+                    s, norm=wz.norm, studio_id=studio.id if studio else None,
+                    performer_names=set().union(*(p.cast for p in k)), wyklucz={movie_id},
+                )
+                if istn is not None:
+                    nowy = istn
+                    liczniki["dolaczone_do_istniejacych"] += 1
+                    log.info("   -> istniejacy film %s %r", nowy.id, nowy.title)
+                else:
                     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,
@@ -330,6 +384,13 @@ def _przetworz(movie_id: uuid.UUID, *, dry_run: bool, liczniki: dict) -> None:
                 s.add(nowy)
                 s.flush()
             klucze = set().union(*(p.keys for p in k))
+            # Rozdzielony film to STARA tresc, nie nowy nabytek: created_at = pierwsze
+            # zobaczenie jego refow, nie "teraz". Inaczej kilka tysiecy filmow z dzisiejsza
+            # data wpadlo w badge "+N nowych" u kazdego, kto ma performera w ulubionych,
+            # a na liscie scen nie bylo nic nowego (zgloszenie 1db95176, 2026-09-02).
+            najstarszy = min(p.first_seen for p in k)
+            if nowy.created_at is None or nowy.created_at > najstarszy:
+                nowy.created_at = najstarszy
             for p in k:
                 s.execute(
                     text("UPDATE movie_external_refs SET movie_id=:n WHERE source_id=:sid AND external_id=:e"),
@@ -385,8 +446,62 @@ def _przetworz(movie_id: uuid.UUID, *, dry_run: bool, liczniki: dict) -> None:
         log.info("   kotwica zostaje jako %r (hostery skasowane: %d)", m.title, r.rowcount)
 
 
+def _sprzataj_duble(utworzone_po: datetime, *, dry_run: bool) -> dict:
+    """Scala filmy utworzone przez ten skrypt z filmem, ktory juz istnial (albo z
+    wczesniejszym z tego samego przebiegu), gdy resolver dalby auto-merge.
+
+    Pierwszy przebieg 2026-09-02 tworzyl nowe filmy bez szukania istniejacych. Ten tryb
+    naprawia skutki: dla kazdego filmu z `created_at > utworzone_po` (od najstarszego)
+    szuka kandydata STARSZEGO od niego i przy >= 0.85 robi `merge_movies(keep=starszy,
+    drop=nowy)`, czyli te sama operacje, ktorej uzywa dedup po TPDB.
+    """
+    from app.resolve.movie_merge import merge_movies
+
+    lic = {"sprawdzone": 0, "scalone": 0}
+    with session_scope() as s:
+        ids = [
+            r[0] for r in s.execute(
+                text("SELECT id FROM movies WHERE created_at > :t ORDER BY created_at"), {"t": utworzone_po}
+            ).fetchall()
+        ]
+    log.info("duble: filmow utworzonych po %s: %d", utworzone_po.isoformat(), len(ids))
+    for i, mid in enumerate(ids, 1):
+        with session_scope() as s:
+            m = s.get(Movie, mid)
+            if m is None:
+                continue
+            lic["sprawdzone"] += 1
+            nazwy = {
+                n.lower() for (n,) in s.execute(
+                    text("SELECT p.canonical_name FROM movie_performers mp JOIN performers p ON p.id=mp.performer_id WHERE mp.movie_id=:m"),
+                    {"m": str(mid)},
+                ).fetchall() if n
+            }
+            norm = NormalizedMovie(
+                external_id="", title=m.title, title_normalized=m.title_normalized, slug=m.slug or "",
+                release_year=m.release_year, release_date=m.release_date, description=None, duration_sec=None,
+                director=None, country=None, rating=None, poster_url=None, backdrop_url=None, url=None,
+            )
+            istn = _istniejacy_film(
+                s, norm=norm, studio_id=m.studio_id, performer_names=nazwy, wyklucz={mid},
+                tylko_starsze_niz=m.created_at,
+            )
+            if istn is None:
+                continue
+            lic["scalone"] += 1
+            log.info("DUBEL %r (%s) -> %r (%s)%s", m.title[:40], str(mid)[:8], istn.title[:40], str(istn.id)[:8], " [dry]" if dry_run else "")
+            if not dry_run:
+                merge_movies(s, keep_id=istn.id, drop_id=mid, resolved_by="movie_unmerge:dedupe")
+        if i % 200 == 0:
+            log.info("[duble %d/%d] %s", i, len(ids), lic)
+    log.info("KONIEC duble: %s", lic)
+    return lic
+
+
 def main(argv: list[str] | None = None) -> int:
     ap = argparse.ArgumentParser()
+    ap.add_argument("--dedupe-created-after", default=None,
+                    help="ISO timestamp: zamiast rozdzielania scal duble utworzone po tej chwili")
     ap.add_argument("--movie-id", default=None)
     ap.add_argument("--limit", type=int, default=0)
     ap.add_argument("--after-id", default=None)
@@ -398,6 +513,10 @@ def main(argv: list[str] | None = None) -> int:
     for noisy in ("httpx", "httpcore", "urllib3", "curl_cffi"):
         logging.getLogger(noisy).setLevel(logging.WARNING)
 
+    if a.dedupe_created_after:
+        _sprzataj_duble(datetime.fromisoformat(a.dedupe_created_after), dry_run=a.dry_run)
+        return 0
+
     if a.movie_id:
         ids = [uuid.UUID(a.movie_id)]
     else:
@@ -421,7 +540,7 @@ def main(argv: list[str] | None = None) -> int:
             ids = [r[0] for r in s.execute(text(sql), params).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")}
+    liczniki = {k: 0 for k in ("filmy", "rozdzielone", "nowe_filmy", "dolaczone_do_istniejacych", "refy_przeniesione", "hostery_przeniesione", "hostery_skasowane", "pobrane", "niepobrane")}
     t0 = time.time()
     for i, mid in enumerate(ids, 1):
         try: