fix(movies): poprawki po code review scorera, merge_movies i rozdzielania

Scorer tytulu (`movie_score._title_signal`), na zywym ingescie:
* sklejanie ogonkow apostrofu obejmowalo KAZDA pojedyncza litere, wiec "watch a whore s
  tale" dawalo "watcha", a przedimek "a" nigdy nie wpadal do wypelniaczy; lustra z A/I
  na poczatku tytulu przestawaly sie laczyc. Teraz tylko s/t/d.
* rowne zbiory zlozone z samych liczb ("adult movie 2019" vs "porn film 2019" -> {2019})
  dawaly 1.0; teraz sufit.
* zaostrzenie "rozne slowa = sufit" zabijalo warianty zapisu: Vol./Volume, Pt./Part,
  Gang-Bang/Gangbang, literowki, liczbe mnoga. Skroty rozwijane, zlepki porownywane po
  zdjeciu spacji, slowa wlasne obu stron parowane jeden do jednego: liczba mnoga = 0.95,
  literowka tylko przy >= 0.92 i slowach od 4 liter. Prog 0.85 na slowie przepuszczal
  "sexy"/"sex", "intense"/"interns" i "infedele"/"fedele" (niewierna vs wierna), a
  porownanie calych posortowanych ciagow "girls only 4" vs "girls on girls 4".
Sprawdzone na 28 parach, w tym wszystkich przykladach z review.

`merge_movies`: ulubione i postep odtwarzania przenoszone na `keep` przed skasowaniem
`drop`. Oba maja FK ondelete=CASCADE, wiec kazde scalenie po cichu wymazywalo userowi
dane. Dotyczy tez dedup po TPDB.

Skrypt rozdzielania:
* `_istniejacy_film` dostaje pelna liste performer_id (jak resolver), nie tylko przekroj,
  ktory zawyzal Jaccarda (3 z 5 wspolnych dawalo 1.0); kandydaci z trigramow PLUS
  dokladne trafienie w title_normalized, bo LIMIT 50 gubil "control 2" wsrod sasiadow;
  wstepny test tytulu oszczedza zapytan o obsade; flush przed blokowaniem.
* klastry rozdzielone w tym samym filmie nie moga sie do siebie dolaczyc (klastrowanie
  juz orzeklo, ze to rozne filmy).
* przed przepisaniem kotwicy sprawdzenie, czy zbior refow nie zmienil sie w trakcie
  pobierania (wyscig miedzy shardami).
* dry-run pokazuje tez dolaczanie do istniejacych; liczniki liczone przy tworzeniu.
* sprzatanie dubli po `updated_at` (created_at jest cofane i nie nadaje sie na znacznik
  przebiegu), keep = wiecej refow, potem starsze pierwsze zobaczenie; try/except per
  film; --after-id; naiwny timestamp = UTC.

app/resolve/movie_merge.py

@@ -14,7 +14,7 @@ import logging
 import uuid
 from datetime import UTC, datetime
 
-from sqlalchemy import or_, select, update
+from sqlalchemy import or_, select, text, update
 from sqlalchemy.orm import Session
 
 from app.models.merge_candidate import MergeCandidate, MergeKind, MergeStatus
@@ -53,6 +53,7 @@ def merge_movies(
     _move_tags(session, keep_id=keep_id, drop_id=drop_id)
     _move_chapters(session, keep_id=keep_id, drop_id=drop_id)
     _move_playback_sources(session, keep_id=keep_id, drop_id=drop_id)
+    _move_user_data(session, keep_id=keep_id, drop_id=drop_id)
     _coalesce_canonical_fields(keep, drop)
 
     session.delete(drop)
@@ -142,6 +143,24 @@ def _move_playback_sources(session: Session, *, keep_id: uuid.UUID, drop_id: uui
     )
 
 
+def _move_user_data(session: Session, *, keep_id: uuid.UUID, drop_id: uuid.UUID) -> None:
+    """Ulubione i postep odtwarzania z `drop` na `keep`.
+
+    Oba maja FK `ondelete=CASCADE` na movies.id, wiec bez tego kroku skasowanie `drop`
+    po cichu wymazywalo userowi ulubione i miejsce, w ktorym skonczyl ogladac (code review
+    2026-09-02, sprzatanie dubli po rozdzielaniu filmow). Gdy urzadzenie ma juz wpis na
+    `keep`, ten na `drop` zostaje i idzie z kaskada: nie ma czego laczyc.
+    """
+    for tabela in ("favorite_movies", "movie_play_progress"):
+        session.execute(
+            text(
+                f"UPDATE {tabela} SET movie_id = :keep WHERE movie_id = :drop "
+                f" AND device_id NOT IN (SELECT device_id FROM {tabela} WHERE movie_id = :keep)"
+            ),
+            {"keep": str(keep_id), "drop": str(drop_id)},
+        )
+
+
 def _coalesce_canonical_fields(keep: Movie, drop: Movie) -> None:
     """Wypełnij braki w `keep` polami z `drop` (bez nadpisywania ustawionych)."""
     if not keep.description and drop.description:

app/resolve/movie_score.py

@@ -117,36 +117,108 @@ 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 = _slowa(cand_title)
+    la, li = _slowa(cand_title), _slowa(inc_title)
-    ti = _slowa(inc_title)
+    ta, ti = frozenset(la), frozenset(li)
-    if ta and ti and ta == ti:
+    if not ta or not ti:
-        # Te same slowa (po zdjeciu wypelniaczy i sklejeniu apostrofow): prawdziwe lustro,
+        # Po zdjeciu wypelniaczy jedna strona jest pusta: nie ma czego porownywac
-        # niezaleznie od kolejnosci i przedrostkow SEO.
+        # 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
-    if ta and ti and (ta < ti or ti < ta):
+    # Literowka albo liczba mnoga: slowa wlasne kazdej strony musza sie sparowac jeden do
-        powody["subset_title_cap"] = sorted((ta ^ ti))[:4]
+    # jednego z podobienstwem >= 0.85 na poziomie SLOWA. Porownanie calych posortowanych
-        return min(ocena, _ONE_SIDED_NUMBER_CAP), False, powody
+    # ciagow bylo za luzne dla krotkich tytulow: "girls only 4" vs "girls on girls 4"
-    if ta and ti:
+    # dawalo 0.91, bo zbior zwija powtorzone "girls", a "only"/"on" to dwie litery.
-        # Kazda strona ma slowo, ktorego druga nie ma: dla filmow to prawie zawsze inny
+    prawie = _parowanie_slow(ta - ti, ti - ta)
-        # tytul, a nie literowka. `token_set_ratio` dawal tu 85-94 ("hardcore fucking sluts
+    if prawie is not None:
-        # 2" vs "foot fucking sluts 2", "santa list gangbang" vs "gangbang d") i przy
+        powody["fuzzy_words"] = round(prawie, 3)
-        # zgodnym studiu i roku wychodzil auto-merge (sprzatanie dubli 2026-09-02).
+        return prawie, False, powody
-        powody["distinct_words_cap"] = sorted(ta ^ ti)[:4]
+    # Podzbior z prawdziwymi dodatkowymi slowami ("doctor" vs "grannys doctor games") albo
-        return min(ocena, _ONE_SIDED_NUMBER_CAP), False, powody
+    # slowa wlasne po obu stronach: dla filmow to inny tytul, nie lustro. `token_set_ratio`
-    return ocena, False, powody
+    # 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
 
 
-def _slowa(title_normalized: str) -> frozenset[str]:
+# Literowka: dwa slowa od 4 liter o podobienstwie >= 0.92 ("fuking"/"fucking" 0.92).
-    """Zbior slow tytulu do porownania: bez wypelniaczy SEO, z doklejonymi ogonkami po
+# Prog 0.85 byl za nisko: "sexy"/"sex" 0.86, "intense"/"interns" 0.86,
-    apostrofie ("she s" -> "shes", "gangbang d" -> "gangbangd"), bo `normalize()` tnie
+# "infedele"/"fedele" 0.86 (niewierna vs wierna!) laczyly rozne filmy.
-    apostrof na spacje i "She's In Control 3" rozjezdzalo sie z "Shes In Control 3"."""
+_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 len(tok) == 1 and tok.isalpha() and out:
+        if tok in _OGONKI_APOSTROFU and out:
             out[-1] = out[-1] + tok
         else:
-            out.append(tok)
+            out.append(_SKROTY.get(tok, tok))
-    return frozenset(out) - _SEO_WYPELNIACZE
+    return [t for t in out if t not in _SEO_WYPELNIACZE]
 
 
 def score_movie_candidate(

scripts/movie_unmerge.py

@@ -48,7 +48,7 @@ import sys
 import time
 import uuid
 from dataclasses import dataclass, field
-from datetime import datetime
+from datetime import UTC, datetime
 from urllib.parse import urlparse
 
 from sqlalchemy import text
@@ -254,44 +254,66 @@ def _opis(klaster: list[Profil]) -> str:
 
 
 def _istniejacy_film(
-    s, *, norm: NormalizedMovie, studio_id, performer_names: set[str], wyklucz: set,
+    s, *, title_normalized: str, release_year: int | None, studio_id, performer_ids: list, 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.
+    z PELNA lista performer_id (jak w `resolve_movie`), a nie tylko z tymi, ktore pasuja
-    Obsada porownywana po nazwach, bo w tym miejscu nie mamy jeszcze performer_id.
+    do kandydata: pierwsza wersja podawala sam przekroj, przez co Jaccard byl gornym
+    oszacowaniem (3 z 5 wspolnych dawalo 1.0 zamiast 0.6) i skrypt laczyl chetniej niz
+    ingest. Kandydaci z trigramow PLUS dokladne trafienie w `title_normalized`, bo
+    `find_movie_candidates` bierze 50 losowych sasiadow i przy popularnych tytulach
+    ("control 2") potrafi pominac ten wlasciwy.
     """
     from app.resolve.movie_match import find_movie_candidates
-    from app.resolve.movie_score import AUTO_THRESHOLD, score_movie_candidate
+    from app.resolve.movie_score import AUTO_THRESHOLD, _title_signal, score_movie_candidate
 
-    kand = find_movie_candidates(
+    s.flush()  # blokowanie po studiu i roku czyta kolumny, ktore mogly zmienic sie bez flush
-        s, title_normalized=norm.title_normalized, studio_id=studio_id, release_year=norm.release_year
+    kand = {c.id: c for c in find_movie_candidates(
-    )
+        s, title_normalized=title_normalized, studio_id=studio_id, release_year=release_year
+    )}
+    for c in s.query(Movie).filter(Movie.title_normalized == title_normalized).all():
+        kand.setdefault(c.id, c)
     najlepszy, najlepsza = None, 0.0
-    for c in kand:
+    for c in kand.values():
         if c.id in wyklucz:
             continue
-        if tylko_starsze_niz is not None and c.created_at and c.created_at >= tylko_starsze_niz:
+        # Tytul jest czysto pythonowy i przy sufitach 0.6 nie da sie dojsc do 0.85, wiec
+        # obsady z bazy nie ma po co czytac dla takich kandydatow.
+        ocena, odrzut, _ = _title_signal(c.title_normalized or "", title_normalized)
+        if odrzut or ocena < 0.7:
             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,
+            s, candidate=c, norm_title_normalized=title_normalized,
-            norm_release_year=norm.release_year, norm_studio_id=studio_id, norm_performer_ids=pids,
+            norm_release_year=release_year, norm_studio_id=studio_id, norm_performer_ids=performer_ids,
         )
         if br.composite > najlepsza:
             najlepszy, najlepsza = c, br.composite
-    return najlepszy if najlepszy is not None and najlepsza >= AUTO_THRESHOLD else None
+            if najlepsza >= 1.0:
+                break
+    return najlepszy if najlepsza >= AUTO_THRESHOLD else None
+
+
+def _performer_ids_po_nazwach(s, nazwy: set[str]) -> list:
+    """Do PODGLADU (dry-run): id performerow po nazwie kanonicznej, bez zapisu."""
+    if not nazwy:
+        return []
+    return [
+        pid for (pid,) in s.execute(
+            text("SELECT id FROM performers WHERE lower(canonical_name) = ANY(:n)"), {"n": list(nazwy)}
+        ).fetchall()
+    ]
+
+
+def _refy_bez_zmian(s, movie_id: uuid.UUID, profile: list[Profil]) -> bool:
+    """Czy zbior refow filmu jest ten sam, co przy wczytaniu (inny shard mogl w miedzyczasie
+    dolaczyc tu klaster). Przepisanie kotwicy na starych profilach skasowaloby jego
+    hostery i obsade jako "niepotwierdzone", wiec taki film pomijamy do nastepnego przebiegu."""
+    teraz = {(sid, e) for (sid, e) in s.execute(
+        text("SELECT source_id, external_id FROM movie_external_refs WHERE movie_id = :m"), {"m": str(movie_id)}
+    ).fetchall()}
+    return teraz == {p.klucz for p in profile}
 
 
 def _zaladuj_refy(s, movie_id: uuid.UUID) -> list[Profil]:
@@ -346,17 +368,41 @@ def _przetworz(movie_id: uuid.UUID, *, dry_run: bool, liczniki: dict) -> None:
         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:
+        # Podglad ma pokazac TO SAMO, co zrobi przebieg: takze dolaczanie do istniejacych.
+        # Bez tego pierwsza wersja dry-run wypisywala "nowy" dla kazdego klastra i nie
+        # dalo sie z gory zobaczyc 798 dubli, ktore potem powstaly.
+        with session_scope() as s:
+            for k in inne:
+                wz = _wzorzec(k)
+                istn = None
+                if wz is not None and wz.norm is not None:
+                    istn = _istniejacy_film(
+                        s, title_normalized=wz.norm.title_normalized, release_year=wz.norm.release_year,
+                        studio_id=None, performer_ids=_performer_ids_po_nazwach(s, set().union(*(p.cast for p in k))),
+                        wyklucz={movie_id},
+                    )
+                if istn is not None:
+                    liczniki["dolaczone_do_istniejacych"] += 1
+                    log.info("   dolaczy do %r: %s", istn.title[:36], _opis(k))
+                else:
+                    liczniki["nowe_filmy"] += 1
+                    log.info("   nowy:    %s", _opis(k))
         return
 
     with session_scope() as s:
         m = s.get(Movie, movie_id)
         assert m is not None
+        if not _refy_bez_zmian(s, movie_id, profile):
+            log.warning("   %s: refy zmienily sie w trakcie pobierania (inny shard?), pomijam do nastepnego przebiegu", tytul_m[:40])
+            liczniki["pominiete_wyscig"] = liczniki.get("pominiete_wyscig", 0) + 1
+            return
+        # Filmy utworzone w TYM podziale: kolejne klastry nie moga sie do nich dolaczac,
+        # bo klastrowanie juz orzeklo, ze to rozne filmy; inaczej rozdzielone przed chwila
+        # sklejalo sie z powrotem przez ocene z bazy (dopelnione pola, obsada po id).
+        utworzone_tu: set = set()
         # --- nowe filmy dla klastrow poza kotwica --------------------------------
         for k in inne:
             wz = _wzorzec(k)
@@ -366,9 +412,16 @@ def _przetworz(movie_id: uuid.UUID, *, dry_run: bool, liczniki: dict) -> None:
                 # 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).
+                # Obsada rozwiazana tak jak w resolverze, pelna lista id.
+                pids: list = []
+                for p in k:
+                    if p.norm is not None:
+                        for pn in p.norm.performers:
+                            pids.append(resolve_performer(s, norm=pn, source_id=p.source_id).id)
                 istn = _istniejacy_film(
-                    s, norm=wz.norm, studio_id=studio.id if studio else None,
+                    s, title_normalized=wz.norm.title_normalized, release_year=wz.norm.release_year,
-                    performer_names=set().union(*(p.cast for p in k)), wyklucz={movie_id},
+                    studio_id=studio.id if studio else None, performer_ids=list(dict.fromkeys(pids)),
+                    wyklucz={movie_id} | utworzone_tu,
                 )
                 if istn is not None:
                     nowy = istn
@@ -376,6 +429,8 @@ def _przetworz(movie_id: uuid.UUID, *, dry_run: bool, liczniki: dict) -> None:
                     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)
+                    utworzone_tu.add(nowy.id)
+                    liczniki["nowe_filmy"] += 1
             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.
@@ -383,6 +438,8 @@ def _przetworz(movie_id: uuid.UUID, *, dry_run: bool, liczniki: dict) -> None:
                 nowy = Movie(title=t, title_normalized=normalize(t), slug=slugify(t))
                 s.add(nowy)
                 s.flush()
+                utworzone_tu.add(nowy.id)
+                liczniki["nowe_filmy"] += 1
             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
@@ -446,62 +503,76 @@ 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:
+def _waga_filmu(s, movie_id: uuid.UUID) -> tuple:
-    """Scala filmy utworzone przez ten skrypt z filmem, ktory juz istnial (albo z
+    """Ktory z dwoch dubli zostaje: wiecej refow, potem starsze pierwsze zobaczenie, potem
-    wczesniejszym z tego samego przebiegu), gdy resolver dalby auto-merge.
+    starsze created_at. NIE sama data utworzenia, bo rozdzielone filmy dostaja created_at
+    z historii refow i "starszy" przestal znaczyc "istniejacy wczesniej"."""
+    refy, fs = s.execute(
+        text("SELECT count(*), min(first_seen) FROM movie_external_refs WHERE movie_id = :m"), {"m": str(movie_id)}
+    ).fetchone()
+    m = s.get(Movie, movie_id)
+    return (-(refy or 0), fs or datetime.max.replace(tzinfo=UTC), m.created_at or datetime.max.replace(tzinfo=UTC))
 
-    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)
+def _sprzataj_duble(zmienione_po: datetime, *, dry_run: bool, after_id: str | None) -> dict:
-    szuka kandydata STARSZEGO od niego i przy >= 0.85 robi `merge_movies(keep=starszy,
+    """Scala duble wsrod filmow ruszonych po `zmienione_po` (po `updated_at`, bo
-    drop=nowy)`, czyli te sama operacje, ktorej uzywa dedup po TPDB.
+    `created_at` rozdzielonych filmow jest cofane do historii refow i nie nadaje sie na
+    znacznik "z tego przebiegu").
+
+    Dla kazdego takiego filmu szuka partnera, do ktorego resolver dolaczylby go
+    automatycznie (>= 0.85), i scala przez `merge_movies`. Zostaje film o wiekszej wadze
+    (`_waga_filmu`), a `merge_movies` przenosi tez ulubione i postep, wiec kierunek nie
+    kosztuje danych usera. Blad przy jednym filmie nie przerywa przebiegu; `--after-id`
+    wznawia.
     """
     from app.resolve.movie_merge import merge_movies
 
-    lic = {"sprawdzone": 0, "scalone": 0}
+    lic = {"sprawdzone": 0, "scalone": 0, "bledy": 0}
+    sql = "SELECT id FROM movies WHERE updated_at > :t"
+    params: dict = {"t": zmienione_po}
+    if after_id:
+        sql += " AND id > :aid"
+        params["aid"] = after_id
     with session_scope() as s:
-        ids = [
+        ids = [r[0] for r in s.execute(text(sql + " ORDER BY id"), params).fetchall()]
-            r[0] for r in s.execute(
+    log.info("duble: filmow zmienionych po %s: %d", zmienione_po.isoformat(), len(ids))
-                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:
+        try:
-            m = s.get(Movie, mid)
+            with session_scope() as s:
-            if m is None:
+                m = s.get(Movie, mid)
-                continue
+                if m is None:
-            lic["sprawdzone"] += 1
+                    continue
-            nazwy = {
+                lic["sprawdzone"] += 1
-                n.lower() for (n,) in s.execute(
+                pids = [pid for (pid,) 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"),
+                    text("SELECT performer_id FROM movie_performers WHERE movie_id = :m"), {"m": str(mid)}
-                    {"m": str(mid)},
+                ).fetchall()]
-                ).fetchall() if n
+                istn = _istniejacy_film(
-            }
+                    s, title_normalized=m.title_normalized, release_year=m.release_year,
-            norm = NormalizedMovie(
+                    studio_id=m.studio_id, performer_ids=pids, wyklucz={mid},
-                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,
+                if istn is None:
-                director=None, country=None, rating=None, poster_url=None, backdrop_url=None, url=None,
+                    continue
-            )
+                keep, drop = (istn, m) if _waga_filmu(s, istn.id) <= _waga_filmu(s, m.id) else (m, istn)
-            istn = _istniejacy_film(
+                lic["scalone"] += 1
-                s, norm=norm, studio_id=m.studio_id, performer_names=nazwy, wyklucz={mid},
+                log.info("DUBEL %r (%s) -> %r (%s)%s", drop.title[:40], str(drop.id)[:8], keep.title[:40], str(keep.id)[:8], " [dry]" if dry_run else "")
-                tylko_starsze_niz=m.created_at,
+                if not dry_run:
-            )
+                    merge_movies(s, keep_id=keep.id, drop_id=drop.id, resolved_by="movie_unmerge:dedupe")
-            if istn is None:
+        except Exception:
-                continue
+            lic["bledy"] += 1
-            lic["scalone"] += 1
+            log.exception("duble: film %s, pomijam", mid)
-            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 i % 200 == 0 or i == len(ids):
-            if not dry_run:
+            log.info("[duble %d/%d] %s (ostatni id %s)", i, len(ids), lic, mid)
-                merge_movies(s, keep_id=istn.id, drop_id=mid, resolved_by="movie_unmerge:dedupe")
+            for h in logging.getLogger().handlers:
-        if i % 200 == 0:
+                h.flush()
-            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,
+    ap.add_argument("--dedupe-updated-after", default=None,
-                    help="ISO timestamp: zamiast rozdzielania scal duble utworzone po tej chwili")
+                    help="ISO timestamp (bez strefy = UTC): zamiast rozdzielania scal duble wsrod "
+                         "filmow zmienionych po tej chwili (po updated_at)")
     ap.add_argument("--movie-id", default=None)
     ap.add_argument("--limit", type=int, default=0)
     ap.add_argument("--after-id", default=None)
@@ -513,8 +584,11 @@ 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:
+    if a.dedupe_updated_after:
-        _sprzataj_duble(datetime.fromisoformat(a.dedupe_created_after), dry_run=a.dry_run)
+        t = datetime.fromisoformat(a.dedupe_updated_after)
+        if t.tzinfo is None:
+            t = t.replace(tzinfo=UTC)  # inaczej Postgres wzialby strefe sesji i okno by sie przesunelo
+        _sprzataj_duble(t, dry_run=a.dry_run, after_id=a.after_id)
         return 0
 
     if a.movie_id: