goon/scripts/review_performer_attributions.py

"""Przegląd i przycinanie przypisań JEDNEGO performera (wzorzec „Rapture").

Dotyczy klasy, której NIE DA SIĘ posprzątać hurtowo: realna osoba (jest w TPDB/StashDB)
o nazwie będącej pospolitym słowem, do której tube-search dokleja wszystko, co ma to
słowo w tytule. „Rapture" miała 30 z 72 przypisań fałszywych, mainstreamowy film „The
Rapture" z Mimi Rogers, literówkę „hymen raptured", tytuły scen („LucidFlix - Tru Kait -
Rapture") i zwykłe użycia („complete rapture", „moments of rapture").

**Dlaczego ręcznie.** Przetestowałem pięć sygnałów automatycznych i żaden nie rozstrzyga:

  - sama provenance ze scrapera, realni twórcy spoza TPDB (Amouranth, MollyRedWolf)
    też mają 100% przypisań ze scrapera
  - obsada z kanonu na scenie, backfill 0026 wypełnił tylko sceny jednoźródłowe, więc
    sceny mieszane, w których leży problem, mają NULL (1 trafienie na 560 tys.)
  - udział nazwy w tytule, u prawdziwych niszowych osób też ~100% (Amouranth 100%,
    Clanddi 100%)
  - „obce" sceny ze słowem w tytule, myli generyczne użycie z pominiętą atrybucją
    (Amouranth ma 137 takich, a to po prostu jej sceny bez przypisanej obsady)
  - stosunek tag/performer ≥1, wyklucza 79 pozycji, ale samej Rapture BY NIE ZŁAPAŁ

Zostaje przeczytanie tytułów. Ten skrypt robi z tego operację na kilka minut zamiast
pół godziny: wypisuje komplet z kontekstem, przyjmuje listę prefiksów do odpięcia i
domyślnie NIE rusza przypisań potwierdzonych przez kanon.

Skala problemu (2026-07-27): 1335 kandydatów z ≥20 scenami, 539 z ≥100 (105 tys.
przypisań). Nie ma sensu przerabiać ich wszystkich, używaj, gdy konkretna osoba
zwróci uwagę, tak jak Rapture wyszła przy audycie phashy.

**Wskaźnik pospolitości** liczony przy każdym przeglądzie: ile scen ma tę nazwę w
tytule, ale NIE jest do niej przypisanych, podzielone przez liczbę jej przypisań.

    Addiction 14,3   Rogue 5,0   Mystery 4,3   Gypsy 3,2   Precious 3,1   Rapture 2,5
    Kwini 1,8   Chyna 0,8   Amouranth 0,45

Wskaźnik NIE mówi, czy osoba jest prawdziwa, mówi tylko, że jej przypisania są
niewiarygodne i trzeba je przeczytać. Rapture przy 2,5 okazała się realna (zostały 42
sceny), „Precious" przy 3,1 okazała się wyłącznie przymiotnikiem („precious
stepdaughter", „her ass is precious") i poszła w całości.

`--candidates` daje ranking, kogo przejrzeć. Działa dzięki indeksowi trigramowemu
(migracja 0027), przedtem pojedyncze szukanie słowa w tytułach było skanem 3,46 mln
wierszy, więc ranking ~800 kandydatów zajmował godziny i został ubity. Teraz jedno
zapytanie to ~26 ms.

Użycie (kontener worker):
  python -m scripts.review_performer_attributions --candidates
  python -m scripts.review_performer_attributions "Rapture"
  python -m scripts.review_performer_attributions "Rapture" --detach-file /tmp/zle.txt
  python -m scripts.review_performer_attributions "Rapture" --detach-file /tmp/zle.txt --yes

`--detach-file` to zwykły plik tekstowy: jedna linia = początek tytułu (bez wielkości
liter). Puste linie i `#` pomijane.
"""
from __future__ import annotations

import argparse
import sys

from sqlalchemy import text

from app.db import session_scope

_LIST_SQL = """
SELECT sc.title,
       sc.duration_sec,
       coalesce(src.kind::text, '?') AS zrodlo,
       (SELECT count(*) FROM scene_performers x WHERE x.scene_id = sc.id) AS ilu_perf,
       sc.id
FROM scene_performers sp
JOIN scenes sc ON sc.id = sp.scene_id
LEFT JOIN sources src ON src.id = sp.source_id
WHERE sp.performer_id = :pid
ORDER BY lower(sc.title)
"""


_FOREIGN_SQL = """
SELECT count(*) FROM scenes s
WHERE s.title ~* ('\\m' || :name || '\\M')
  AND NOT EXISTS (SELECT 1 FROM scene_performers sp
                  WHERE sp.scene_id = s.id AND sp.performer_id = :pid)
"""

# UWAGA: `ORDER BY 3::numeric / scen` NIE działa, po rzutowaniu Postgres traktuje
# trójkę jako literał, nie numer kolumny, i sortuje po 3/scen (czyli po liczbie scen
# rosnąco). Wyrażenie trzeba powtórzyć jawnie, stąd osobne CTE.
_CANDIDATES_SQL = """
WITH kand AS (
  SELECT pf.id, pf.canonical_name,
         (SELECT count(*) FROM scene_performers sp WHERE sp.performer_id = pf.id) AS scen
  FROM performers pf
  WHERE pf.canonical_name !~ ' ' AND length(pf.canonical_name) >= 4
), z AS (
  SELECT k.canonical_name, k.scen,
         (SELECT count(*) FROM scenes s
          WHERE s.title ~* ('\\m' || k.canonical_name || '\\M')
            AND NOT EXISTS (SELECT 1 FROM scene_performers sp
                            WHERE sp.scene_id = s.id AND sp.performer_id = k.id)) AS obce
  FROM kand k WHERE k.scen >= :min_scenes
)
SELECT canonical_name, scen, obce FROM z
ORDER BY obce::numeric / scen DESC
LIMIT :limit
"""


def _find_performer(session, name: str):
    row = session.execute(
        text(
            "SELECT id, canonical_name FROM performers "
            "WHERE lower(canonical_name) = lower(:n) ORDER BY id LIMIT 1"
        ),
        {"n": name},
    ).first()
    return row


def main() -> None:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("performer", nargs="?", help="canonical_name performera")
    ap.add_argument("--candidates", action="store_true", help="ranking: kogo przejrzeć")
    ap.add_argument("--min-scenes", type=int, default=60)
    ap.add_argument("--limit", type=int, default=25)
    ap.add_argument("--detach-file", help="plik z prefiksami tytułów do odpięcia")
    ap.add_argument("--yes", action="store_true", help="wykonaj odpięcie (bez tego dry-run)")
    ap.add_argument(
        "--include-canonical",
        action="store_true",
        help="pozwól odpiąć też przypisania z tpdb/stashdb (domyślnie chronione)",
    )
    args = ap.parse_args()

    if args.candidates:
        with session_scope() as s:
            rows = s.execute(
                text(_CANDIDATES_SQL),
                {"min_scenes": args.min_scenes, "limit": args.limit},
            ).fetchall()
        print("nazwa                  sceny    obce  wskaźnik")
        for nazwa, scen, obce in rows:
            print(f"  {nazwa[:20]:22} {scen:5} {obce:7}  {obce/scen:7.1f}")
        print("\nWskaźnik = obce sceny ze słowem w tytule / przypisania tej osoby.")
        print("Wysoki = nazwa jest zwykłym słowem; NIE znaczy, że osoba jest fikcyjna.")
        return

    if not args.performer:
        print("podaj canonical_name performera albo --candidates")
        sys.exit(1)

    prefixes: list[str] = []
    if args.detach_file:
        with open(args.detach_file, encoding="utf-8") as fh:
            prefixes = [
                ln.strip().lower()
                for ln in fh
                if ln.strip() and not ln.lstrip().startswith("#")
            ]

    with session_scope() as s:
        perf = _find_performer(s, args.performer)
        if perf is None:
            print(f"nie znalazłem performera: {args.performer!r}")
            sys.exit(1)
        pid, name = perf[0], perf[1]
        rows = s.execute(text(_LIST_SQL), {"pid": pid}).fetchall()
        obce = int(s.execute(text(_FOREIGN_SQL), {"name": name, "pid": pid}).scalar_one())

        wsk = obce / len(rows) if rows else 0.0
        print(f"{name} — {len(rows)} przypisań, obcych scen ze słowem: {obce}, "
              f"wskaźnik pospolitości: {wsk:.1f}")
        print("(wysoki = nazwa jest zwykłym słowem, przypisania trzeba przeczytać)\n")
        hit_ids: list[str] = []
        for title, dur, zrodlo, ilu, sid in rows:
            t = (title or "").lower()
            matched = any(t.startswith(p) for p in prefixes)
            protected = matched and zrodlo in ("tpdb", "stashdb") and not args.include_canonical
            if matched and not protected:
                hit_ids.append(sid)
            mark = "ODPNIJ" if (matched and not protected) else ("CHRONIONE" if protected else "      ")
            w_tytule = "T" if name.lower() in t else "-"
            print(f"  {mark:10} [{zrodlo:8}] nazwa-w-tytule={w_tytule} perf={ilu} "
                  f"{dur or '?'}s  {(title or '')[:58]!r}")

        if not prefixes:
            print("\n(sam przegląd — podaj --detach-file, żeby coś odpiąć)")
            return

        print(f"\ndo odpięcia: {len(hit_ids)} / {len(rows)}")
        if not args.yes:
            print("(dry-run — uruchom z --yes)")
            return
        n = s.execute(
            text("DELETE FROM scene_performers WHERE performer_id = :pid AND scene_id = ANY(:ids)"),
            {"pid": pid, "ids": hit_ids},
        ).rowcount
        s.commit()
        print(f"ODPIĘTE: {n}")


if __name__ == "__main__":
    main()