fix(unmerge): flush po kazdym profilu, dopelnianie pol ze wszystkich luster, shardy

Pierwszy prawdziwy przebieg na "The Doctor" padl na kluczu glownym movie_performers:
helpery resolvera sprawdzaja "czy juz jest" SELECT-em, ktory bez flush nie widzi
wierszy dodanych przez poprzedni profil klastra, wiec ten sam performer z dwoch luster
wpadal dwa razy. Teraz flush po kazdym profilu; transakcja per film, wiec nieudany
film wycofuje sie w calosci i nic nie zostaje w polowie.

Dopelnianie pol (tylko braki) z kazdego pobranego lustra klastra, nie tylko ze wzorca:
wzorcem bywa paradisehill bez roku, a mangoporn ma rok i studio ("Doctor Adventures 14"
tracil rok 2013).

`--shard k/n` po hashtext(id) do rownoleglych kontenerow: 8204 filmy w jednym watku to
~16 h (40 filmow w 4,6 min w probie), w trzech ~6 h.

Wynik na "The Doctor": 16 refow -> 11 filmow, kotwica z jednym refem i rozdzialami,
"Granny's Doctor Games" z dokladnie swoimi dwiema stronami i embedami, 76 zwietrzalych
lub cudzych hosterow skasowanych z kotwicy, 115 przeniesionych do wlasciwych filmow.

scripts/movie_unmerge.py

@@ -278,6 +278,11 @@ def _przypnij_zawartosc(s, movie_id: uuid.UUID, p: Profil) -> None:
     _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)
+    # Klaster to zwykle kilka luster tego samego filmu z ta sama obsada. Helpery
+    # resolvera sprawdzaja "czy juz jest" SELECT-em, ktory bez flush nie widzi wierszy
+    # dodanych przez poprzedni profil, wiec ten sam performer wpadal dwa razy i padal
+    # klucz glowny movie_performers (pierwszy prawdziwy przebieg na "The Doctor").
+    s.flush()
 
 
 def _przetworz(movie_id: uuid.UUID, *, dry_run: bool, liczniki: dict) -> None:
@@ -339,6 +344,10 @@ def _przetworz(movie_id: uuid.UUID, *, dry_run: bool, liczniki: dict) -> None:
             for p in k:
                 if p.norm is not None:
                     _przypnij_zawartosc(s, nowy.id, p)
+                    # Dopelnianie pol z kazdego lustra (tylko braki): wzorcem bywa
+                    # paradisehill bez roku, a mangoporn ma rok i studio.
+                    st = resolve_studio(s, norm=p.norm.studio, source_id=p.source_id) if p.norm.studio else None
+                    _update_movie_fields(nowy, p.norm, studio_id=st.id if st else None)
             liczniki["refy_przeniesione"] += len(k)
             log.info("   -> nowy film %s %r", nowy.id, nowy.title)
 
@@ -371,6 +380,8 @@ def _przetworz(movie_id: uuid.UUID, *, dry_run: bool, liczniki: dict) -> None:
             _update_movie_fields(m, wz.norm, studio_id=studio.id if studio else None)
             for p in wzory:
                 _przypnij_zawartosc(s, movie_id, p)
+                st = resolve_studio(s, norm=p.norm.studio, source_id=p.source_id) if p.norm.studio else None
+                _update_movie_fields(m, p.norm, studio_id=st.id if st else None)
         log.info("   kotwica zostaje jako %r (hostery skasowane: %d)", m.title, r.rowcount)
 
 
@@ -379,6 +390,7 @@ def main(argv: list[str] | None = None) -> int:
     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("--shard", default=None, help="k/n: tylko filmy z hashtext(id) %% n == k (rownolegle kontenery)")
     ap.add_argument("--dry-run", action="store_true")
     a = ap.parse_args(argv)
 
@@ -394,13 +406,19 @@ def main(argv: list[str] | None = None) -> int:
              WHERE movie_id IN (SELECT movie_id FROM movie_external_refs
                                  GROUP BY movie_id, source_id HAVING count(*) >= 2)
         """
+        params: dict = {}
         if a.after_id:
             sql += " AND movie_id > :aid"
+            params["aid"] = a.after_id
+        if a.shard:
+            k, n = (int(x) for x in a.shard.split("/"))
+            sql += " AND abs(hashtext(movie_id::text)) % :n = :k"
+            params.update({"k": k, "n": n})
         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()]
+            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")}