"""Kasuje błędne przypisania performerów do scen z sxyprn (over-atrybucja).

Zgłoszenie `e28f2517` z 2026-08-24. Scena „Bridgette B, Gigi Rivera, Julia Ann, Phoenix
Marie, ..." miała przypisaną **Julię B.**, której w tytule nie ma.

Skąd się to brało: scraper sxyprn woła stronę performera i sprawdza, czy scena naprawdę
go dotyczy, porównując człony nazwiska z tytułem. Sprawdzał je jednak KAŻDY Z OSOBNA i
tylko od trzech znaków. Dla „Julia B." człon „b." odpadał na długości, zostawało samo
„julia" i przypisanie łapało każdą scenę, w której tytule pada to imię. Ten sam
mechanizm dał „Chloe S" przy „Chloe Kreams", „Iris N" przy „Iris Rodriguez" i tak dalej.

Poprawka w samym scraperze siedzi w `person_named_in` (`app/normalize/text.py`) i wymaga,
by człony stały OBOK SIEBIE, z tolerancją na wariant zapisu od trzech znaków. Ten skrypt
stosuje dokładnie tę samą regułę wstecz, do tego, co już jest w bazie.

Zakres: TYLKO sceny, których jedynym źródłem jest sxyprn, i TYLKO przypisania od
scrapera. Kanon (tpdb/stashdb) nie jest ruszany, bo jest wiarygodniejszy niż tytuł z
tuba. Sceny mające obok sxyprn inne źródło zostawiamy, bo performer mógł tam trafić z
tamtego tuba i tytuł sxyprn nic o tym nie mówi.

Skala przed uruchomieniem (2026-08-24): 125 524 scen wyłącznie z sxyprn, 244 309
przypisań od scrapera, z czego 85 711 bez nazwiska w tytule.

Weryfikacja reguły przed kasowaniem: dwie próbki po ~22 pozycje obejrzane ręcznie,
wszystkie do skasowania były faktycznie błędne. Wyjątki, które reguła CELOWO zachowuje,
to warianty zapisu w rodzaju „Savanna Bond" przy tytule „Savannah Bond".

Czego ten skrypt NIE robi: nie rusza śmieciowych performerów nazwanych od tagów
(„Rope Bondage", „Sub Girl"). To osobny problem, od niego jest kryterium
promiskuityczności taga.

Użycie (w kontenerze workera):
    python3 /srv/scripts/sxyprn_fix_overattribution.py --dry-run
    python3 /srv/scripts/sxyprn_fix_overattribution.py
"""
from __future__ import annotations

import argparse
import logging
import sys
import time

from sqlalchemy import text

from app.db import session_scope
from app.normalize.text import person_named_in

log = logging.getLogger("sxyprn-overattr")

_ORIGIN = "tube:sxyprncom"
_BATCH = 2000

_SELECT = """
    WITH tylko_sx AS (
        SELECT DISTINCT ps.scene_id
          FROM playback_sources ps
         WHERE ps.origin = :origin
           AND NOT EXISTS (
               SELECT 1 FROM playback_sources p2
                WHERE p2.scene_id = ps.scene_id AND p2.origin <> :origin
           )
    )
    SELECT sp.scene_id, sp.performer_id, p.canonical_name, s.title
      FROM scene_performers sp
      JOIN tylko_sx t ON t.scene_id = sp.scene_id
      JOIN performers p ON p.id = sp.performer_id
      JOIN scenes s ON s.id = sp.scene_id
      JOIN sources src ON src.id = sp.source_id AND src.name = 'tube-scraper'
"""


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser()
    p.add_argument("--dry-run", action="store_true", help="nie kasuj, tylko policz i pokaż próbkę")
    p.add_argument("--limit", type=int, default=0, help="ogranicz liczbę sprawdzanych wierszy")
    p.add_argument("--sample", type=int, default=15, help="ile przykładów wypisać")
    a = p.parse_args(argv)

    logging.basicConfig(
        level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", stream=sys.stdout
    )
    t0 = time.time()

    sql = _SELECT + (f" LIMIT {int(a.limit)}" if a.limit else "")
    with session_scope() as s:
        rows = s.execute(text(sql), {"origin": _ORIGIN}).fetchall()
    log.info("przypisań do sprawdzenia: %d (%.0f s)", len(rows), time.time() - t0)

    do_kasacji: list[tuple] = []
    przyklady: list[str] = []
    for scene_id, performer_id, nazwa, tytul in rows:
        if person_named_in(tytul, nazwa):
            continue
        do_kasacji.append((scene_id, performer_id))
        if len(przyklady) < a.sample:
            przyklady.append(f"  {nazwa!r:32} nie występuje w {str(tytul)[:72]!r}")

    log.info(
        "do skasowania: %d z %d (%.1f%%)",
        len(do_kasacji), len(rows), 100.0 * len(do_kasacji) / max(1, len(rows)),
    )
    for line in przyklady:
        log.info("%s", line)

    if a.dry_run:
        log.info("dry-run, nic nie skasowane")
        return 0

    skasowane = 0
    for i in range(0, len(do_kasacji), _BATCH):
        chunk = do_kasacji[i : i + _BATCH]
        with session_scope() as s:
            skasowane += s.execute(
                text(
                    """
                    DELETE FROM scene_performers sp
                     USING unnest(CAST(:sids AS uuid[]), CAST(:pids AS uuid[])) AS u(sid, pid)
                     WHERE sp.scene_id = u.sid AND sp.performer_id = u.pid
                    """
                ),
                {"sids": [str(x[0]) for x in chunk], "pids": [str(x[1]) for x in chunk]},
            ).rowcount
        log.info("skasowane %d/%d (%.0f s)", skasowane, len(do_kasacji), time.time() - t0)

    log.info("KONIEC: skasowane %d przypisań w %.0f s", skasowane, time.time() - t0)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())