goon/scripts/prune_token_overlap_attributions.py

"""Kasuje przypisania performerów powstałe przez CZĘŚCIOWE dopasowanie nazwy.

Mechanizm, który to produkował (naprawiony 2026-08-13 w `_search_base._slug_matches_query`):
search po nazwie doklejał szukanego performera do KAŻDEJ sceny, która przeszła filtr
tokenów, a filtr testował PODCIĄG na luźnych tokenach. Więc „Tiffany Time" łapało się
w „s-tiffany-love-first-time", a „Eliza E" w „eliza-ibarra". Zgłoszenie 2026-08-13:
strona Tiffani Time pełna scen o Tiffany Tatum, Tiffani Madison i Victorii Tiffani.

**Kryterium (trzy warunki naraz, każdy konieczny):**

1. przypisanie pochodzi ze scrapera (kanon nietykalny, `scene_performers.source_id`),
2. nazwy performera NIE ma w tytule ani URL-u sceny (brak dowodu),
3. na tej samej scenie jest INNY performer, który nazwę ma **i dzieli z nim człon**.

Trzeci warunek jest tym, co odróżnia pomyłkę od braku danych, i został dołożony po
obejrzeniu próbki. Bez niego reguła kasowała **męskich performerów**: tuby zwyczajowo
nie wypisują ich w tytule („Charles Dera" przy scenie „HouseHumpers – Demi Hawks and
Serena Hill"), a przypisanie przyszło z listy obsady, nie z searcha, więc było poprawne.
Po zawężeniu próbka 14/14 to realne pomyłki: Nikki Nevada przy Nikki Brooks, Eliza E
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).

Uruchomienie (kontener worker):
  python -m scripts.prune_token_overlap_attributions            # dry-run
  python -m scripts.prune_token_overlap_attributions --yes      # wykonaj
"""
from __future__ import annotations

import argparse
import logging

from sqlalchemy import text

from app.db import session_scope

log = logging.getLogger(__name__)

#: 16 przedziałów po pierwszym nibblu UUID-a. Każdy to osobna transakcja.
_PREFIXES = "0123456789abcdef"

_BUILD_SQL = """
CREATE TEMP TABLE ev ON COMMIT DROP AS
SELECT sp.scene_id,
       sp.performer_id,
       (replace(replace(replace(
            lower(coalesce(ps.page_url,'') || ' ' || s.title), '-',''), '_',''), ' ','')
        LIKE '%' || replace(lower(p.canonical_name),' ','') || '%') AS ma_dowod,
       ARRAY(
         SELECT t FROM unnest(string_to_array(lower(p.canonical_name),' ')) t
         WHERE length(t) >= 3
       ) AS tokeny
FROM scene_performers sp
JOIN sources src ON src.id = sp.source_id AND src.name = 'tube-scraper'
JOIN performers p ON p.id = sp.performer_id
JOIN scenes s ON s.id = sp.scene_id
LEFT JOIN LATERAL (
    SELECT page_url FROM playback_sources x WHERE x.scene_id = s.id LIMIT 1
) ps ON true
WHERE sp.scene_id >= CAST(:lo AS uuid) AND sp.scene_id < CAST(:hi AS uuid);
CREATE INDEX ON ev (scene_id);
ANALYZE ev;
"""

#: Kandydaci: bez dowodu, ale inny performer TEJ SCENY dowód ma i dzieli człon.
_SELECT_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
  )
"""


def _range(prefix: str) -> tuple[str, str]:
    lo = prefix + "0000000-0000-0000-0000-000000000000"
    nxt = _PREFIXES[_PREFIXES.index(prefix) + 1] if prefix != "f" else None
    hi = (nxt + "0000000-0000-0000-0000-000000000000") if nxt else "ffffffff-ffff-ffff-ffff-ffffffffffff"
    return lo, hi


def run(*, apply: bool) -> dict[str, int]:
    total = 0
    for prefix in _PREFIXES:
        lo, hi = _range(prefix)
        with session_scope() as session:
            for stmt in _BUILD_SQL.strip().split(";\n"):
                if stmt.strip():
                    session.execute(text(stmt), {"lo": lo, "hi": hi})
            rows = session.execute(text(_SELECT_SQL)).all()
            n = len(rows)
            if apply and rows:
                session.execute(
                    text(
                        """
                        DELETE FROM scene_performers sp
                        USING (SELECT * 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)) 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)")
        print(f"  zakres {prefix}: {n}{' skasowanych' if apply else ' do skasowania'}", flush=True)
    return {"matched": total}


def main() -> None:
    logging.basicConfig(level=logging.INFO, format="%(message)s")
    ap = argparse.ArgumentParser()
    ap.add_argument("--yes", action="store_true", help="wykonaj DELETE (bez tego dry-run)")
    args = ap.parse_args()
    out = run(apply=args.yes)
    print(f"\nRAZEM: {out['matched']} {'skasowanych' if args.yes else 'do skasowania (dry-run)'}")


if __name__ == "__main__":
    main()