feat(cleanup): archiwum przed kasowaniem bledych przypisan performerow

Kasowane wiersze ida najpierw do `scene_performers_pruned`, w TEJ SAMEJ
transakcji co DELETE: albo wiersz jest zapisany i skasowany, albo nie
dzieje sie nic. Cofniecie to `INSERT ... SELECT`, nie backup calej bazy.
Archiwum trzyma tez nazwe performera i tytul sceny, wiec da sie je
przejrzec bez odtwarzania joinow. Tabela swiadomie poza alembikiem: to
slad po jednorazowym sprzataniu, nie czesc schematu.

Kryterium doprecyzowane po dwoch losowych probkach:

1. „brak nazwy w tytule" samo w sobie kasowalo MESKICH performerow (tuby
   nie wypisuja ich w tytulach, przypisanie szlo z listy obsady) -> stad
   warunek „inny performer sceny ma dowod I dzieli czlon".
2. Dowodem dla nazwy o >=2 czlonach jest kazdy czlon jako OSOBNE SLOWO,
   nie spojna fraza: „Silvia and Eveline Dellai" to scena Silvii Dellai,
   ale ciagu „silviadellai" tam nie ma (1 blad na 22 w probce). Nazwy o
   jednym uzytecznym czlonie („Eliza E") zostaja przy ostrym kryterium,
   bo inaczej lapia kazda inna Elize.
3. Granica slowa przez `strpos` na znormalizowanym tekscie, NIE regex:
   nazwy z nawiasami wywalaly caly przebieg („parentheses not balanced"),
   a metaznaki po cichu psulyby dopasowania.

Dry-run: 321 833 przypisania (vs 370 587 przed poprawka 2). Szkoda
rozproszona, nie skoncentrowana: rekordzistka traci 20 z 909.

scripts/prune_token_overlap_attributions.py

@@ -22,6 +22,10 @@ przy Elizie Ibarze, Miss Marisa przy Miss Fufu, CJ Sloan przy Sloan Rider.
 Wsadowo po zakresach `scene_id`, bo pojedynczy DELETE po całej tabeli trzymałby locki
 na `scene_performers` i zablokował ingest (zdarzyło się przy migracji 0026).
 
+Kasowane wiersze lądują najpierw w `scene_performers_pruned` (w tej samej transakcji),
+więc cofnięcie to `INSERT INTO scene_performers SELECT scene_id, performer_id, source_id
+FROM scene_performers_pruned`, a nie odtwarzanie bazy z backupu.
+
 Uruchomienie (kontener worker):
   python -m scripts.prune_token_overlap_attributions            # dry-run
   python -m scripts.prune_token_overlap_attributions --yes      # wykonaj
@@ -37,6 +41,54 @@ from app.db import session_scope
 
 log = logging.getLogger(__name__)
 
+#: Archiwum kasowanych wierszy. Świadomie POZA alembikiem: to tabela operacyjna,
+#: jednorazowy ślad po sprzątaniu, nie część schematu, którego pilnują migracje.
+#: Trzymamy w niej też nazwę performera i tytuł sceny, żeby dało się to przejrzeć
+#: bez odtwarzania joinów, a cofnięcie było jednym `INSERT ... SELECT`.
+_ARCHIVE_DDL = """
+CREATE TABLE IF NOT EXISTS scene_performers_pruned (
+    scene_id uuid NOT NULL,
+    performer_id uuid NOT NULL,
+    source_id uuid,
+    performer_name text,
+    scene_title text,
+    pruned_at timestamptz NOT NULL DEFAULT now(),
+    PRIMARY KEY (scene_id, performer_id)
+)
+"""
+
+#: Kandydaci do archiwum i kasowania — jedna definicja, używana przez oba kroki,
+#: żeby archiwum nie mogło się rozjechać z tym, co faktycznie znika.
+_CANDIDATES_SQL = """
+SELECT e.scene_id, e.performer_id
+FROM ev e
+WHERE NOT e.ma_dowod
+  AND EXISTS (
+      SELECT 1 FROM ev e2
+      WHERE e2.scene_id = e.scene_id
+        AND e2.performer_id <> e.performer_id
+        AND e2.ma_dowod
+        AND e2.tokeny && e.tokeny
+  )
+"""
+
+_ARCHIVE_SQL = f"""
+INSERT INTO scene_performers_pruned
+       (scene_id, performer_id, source_id, performer_name, scene_title)
+SELECT sp.scene_id, sp.performer_id, sp.source_id, p.canonical_name, s.title
+FROM ({_CANDIDATES_SQL}) d
+JOIN scene_performers sp ON sp.scene_id = d.scene_id AND sp.performer_id = d.performer_id
+JOIN performers p ON p.id = sp.performer_id
+JOIN scenes s ON s.id = sp.scene_id
+ON CONFLICT DO NOTHING
+"""
+
+_DELETE_SQL = f"""
+DELETE FROM scene_performers sp
+USING ({_CANDIDATES_SQL}) d
+WHERE sp.scene_id = d.scene_id AND sp.performer_id = d.performer_id
+"""
+
 #: 16 przedziałów po pierwszym nibblu UUID-a. Każdy to osobna transakcja.
 _PREFIXES = "0123456789abcdef"
 
@@ -44,9 +96,37 @@ _BUILD_SQL = """
 CREATE TEMP TABLE ev ON COMMIT DROP AS
 SELECT sp.scene_id,
        sp.performer_id,
-       (replace(replace(replace(
+       CASE
+         -- Nazwa o >=2 uzytecznych czlonach: dowodem jest obecnosc KAZDEGO z nich jako
+         -- OSOBNEGO SLOWA, niekoniecznie obok siebie. Inaczej wypadaly osoby wymieniane
+         -- lacznie: „Silvia and Eveline Dellai" to scena Silvii Dellai, ale ciag
+         -- „silviadellai" w niej nie wystepuje (probka losowa: 1 blad na 22).
+         --
+         -- Granica slowa realizowana przez `strpos` na znormalizowanym, obustronnie
+         -- odspacjowanym tekscie, a NIE przez regex: nazwy potrafia zawierac nawiasy
+         -- („… (Brazil)") i wstawione do wyrazenia wywalaly caly przebieg
+         -- („parentheses not balanced"). Przy okazji `strpos` nie ma metaznakow,
+         -- wiec `%` i `_` w nazwie tez sa bezpieczne.
+         WHEN cardinality(ARRAY(SELECT t FROM unnest(string_to_array(lower(p.canonical_name),' ')) t
+                                WHERE length(t) >= 3)) >= 2
+           THEN NOT EXISTS (
+                  SELECT 1
+                  FROM unnest(string_to_array(lower(p.canonical_name),' ')) t
+                  WHERE length(regexp_replace(t, '[^a-z0-9]+', '', 'g')) >= 3
+                    AND strpos(
+                          ' ' || regexp_replace(
+                                   lower(coalesce(ps.page_url,'') || ' ' || s.title),
+                                   '[^a-z0-9]+', ' ', 'g') || ' ',
+                          ' ' || regexp_replace(t, '[^a-z0-9]+', '', 'g') || ' '
+                        ) = 0
+                )
+         -- Nazwa sprowadzajaca sie do JEDNEGO czlonu („Eliza E", „Pinky 5") degeneruje sie
+         -- do „scena zawiera to imie", czyli lapie kazda inna Elize. Tu wymagamy calej
+         -- nazwy sklejonej, tak jak robi to bramka searcha.
+         ELSE replace(replace(replace(
                 lower(coalesce(ps.page_url,'') || ' ' || s.title), '-',''), '_',''), ' ','')
-        LIKE '%' || replace(lower(p.canonical_name),' ','') || '%') AS ma_dowod,
+              LIKE '%' || replace(lower(p.canonical_name),' ','') || '%'
+       END AS ma_dowod,
        ARRAY(
          SELECT t FROM unnest(string_to_array(lower(p.canonical_name),' ')) t
          WHERE length(t) >= 3
@@ -87,6 +167,9 @@ def _range(prefix: str) -> tuple[str, str]:
 
 def run(*, apply: bool) -> dict[str, int]:
     total = 0
+    if apply:
+        with session_scope() as session:
+            session.execute(text(_ARCHIVE_DDL))
     for prefix in _PREFIXES:
         lo, hi = _range(prefix)
         with session_scope() as session:
@@ -96,20 +179,15 @@ def run(*, apply: bool) -> dict[str, int]:
             rows = session.execute(text(_SELECT_SQL)).all()
             n = len(rows)
             if apply and rows:
-                session.execute(
+                # Archiwum i DELETE w JEDNEJ transakcji: albo wiersz jest zapisany
-                    text(
+                # i skasowany, albo nie dzieje się nic. Odwrotna kolejność albo
-                        """
+                # osobne transakcje mogłyby zostawić skasowane wiersze bez śladu.
-                        DELETE FROM scene_performers sp
+                archived = session.execute(text(_ARCHIVE_SQL)).rowcount or 0
-                        USING (SELECT * FROM ev e
+                deleted = session.execute(text(_DELETE_SQL)).rowcount or 0
-                               WHERE NOT e.ma_dowod
+                if archived != deleted:
-                                 AND EXISTS (SELECT 1 FROM ev e2
+                    log.warning(
-                                             WHERE e2.scene_id = e.scene_id
+                        "zakres %s: zarchiwizowano %d, skasowano %d (rozjazd)",
-                                               AND e2.performer_id <> e.performer_id
+                        prefix, archived, deleted,
-                                               AND e2.ma_dowod
-                                               AND e2.tokeny && e.tokeny)) d
-                        WHERE sp.scene_id = d.scene_id AND sp.performer_id = d.performer_id
-                        """
-                    )
                     )
             total += n
         log.info("zakres %s: %d przypisan%s", prefix, n, " (skasowane)" if apply else " (dry-run)")