"""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).

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
"""
from __future__ import annotations

import argparse
import logging

from sqlalchemy import text

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"

_BUILD_SQL = """
CREATE TEMP TABLE ev ON COMMIT DROP AS
SELECT sp.scene_id,
       sp.performer_id,
       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),' ','') || '%'
       END 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
    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:
            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:
                # Archiwum i DELETE w JEDNEJ transakcji: albo wiersz jest zapisany
                # i skasowany, albo nie dzieje się nic. Odwrotna kolejność albo
                # osobne transakcje mogłyby zostawić skasowane wiersze bez śladu.
                archived = session.execute(text(_ARCHIVE_SQL)).rowcount or 0
                deleted = session.execute(text(_DELETE_SQL)).rowcount or 0
                if archived != deleted:
                    log.warning(
                        "zakres %s: zarchiwizowano %d, skasowano %d (rozjazd)",
                        prefix, archived, deleted,
                    )
            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()