"""Usuwa z bazy sceny zablokowanych uploaderow sxyprn i sciaga smieciowych performerow.

Kontekst (2026-09-04): trzej autorzy (Shizuka, Ajx, Denzel) wrzucaja dobre sceny, ale z
obsada wpisana w tytul i 25-35 hashtagami. Tytul z lista nazwisk to over-attribution
performerow, hashtagi to 27 tagow na scene zamiast 12. Scraper ma juz ich na blokadzie
(`_BLOCKED_AUTHORS` w `app/connectors/direct_scrapers/sxyprn.py`), ten skrypt sprzata
to, co juz weszlo.

Autora poznajemy po sciezce miniatury: `.../<id_autora>/<id_posta>/small.jpg`.

Co robi:
  1. Sceny, ktore maja TYLKO zrodla tych autorow (zadnego innego zywego zrodla, zadnego
     refu tpdb/stashdb): DELETE ze scenes (kaskada zdejmuje refy, performerow, tagi,
     ulubione, postep). To jest utrata danych userow, dlatego domyslnie dry-run.
  2. Sceny scalone z innymi zrodlami: zostaja, wylatuje tylko zrodlo sxyprn tego autora
     i tag „Sizuka" (podpis Shizuki).
  3. Smieciowi performerzy: na scenach, ktorych jedynym zrodlem jest sxyprn, zdejmuje
     przypisania jednowyrazowych performerow, ktorych nazwa nie stoi doslownie w tytule
     (ani jako zlepek „JoannaAngel" przy tytule „Joanna Angel").
     To pozostalosc po dopasowaniu przedrostkiem w obie strony („Thebdck" do „the"),
     naprawionym tego samego dnia w `person_named_in`.

Uzycie (na VPS, z pliku, nie przez stdin):
  docker compose run --rm worker python3 -u /srv/scripts/sxyprn_purge_uploaders.py
  docker compose run --rm worker python3 -u /srv/scripts/sxyprn_purge_uploaders.py --apply
"""
from __future__ import annotations

import argparse
import sys

from sqlalchemy import text

from app.db import session_scope

AUTORZY = {
    "69cab3467ec41": "Shizuka",
    "68e92925d23d9": "Ajx",
    "6a0d125a875d2": "Denzel",
}
_CHUNK = 500

# Zrodlo sxyprn z id autora wyciagnietym z miniatury (drugi segment musi byc id posta).
_SQL_ZRODLA_AUTOROW = text(
    """
    SELECT ps.id, ps.scene_id, a.aid
    FROM playback_sources ps
    CROSS JOIN LATERAL (
        SELECT substring(
            ps.thumbnail_url FROM
            '/([a-f0-9]{13})/' || substring(ps.page_url FROM '/post/([a-f0-9]+)\\.html') || '/'
        ) AS aid
    ) a
    WHERE ps.origin = 'tube:sxyprncom' AND a.aid = ANY(:autorzy)
    """
)

_SQL_KLASYFIKACJA = text(
    """
    SELECT s.id,
      EXISTS (SELECT 1 FROM playback_sources p WHERE p.scene_id = s.id
              AND p.dead_at IS NULL AND p.id <> ALL(:zrodla_autorow)) AS inne_zrodlo,
      EXISTS (SELECT 1 FROM scene_external_refs r JOIN sources so ON so.id = r.source_id
              WHERE r.scene_id = s.id AND so.name IN ('tpdb', 'stashdb')) AS kanon,
      (SELECT count(*) FROM favorite_scenes f WHERE f.scene_id = s.id) AS ulubione,
      (SELECT count(*) FROM scene_play_progress f WHERE f.scene_id = s.id) AS ogladane
    FROM scenes s WHERE s.id = ANY(:sceny)
    """
)

_SQL_SMIECIOWI_PERF = text(
    """
    WITH only_sx AS (
      SELECT s.id, s.title FROM scenes s
      WHERE EXISTS (SELECT 1 FROM playback_sources p WHERE p.scene_id = s.id AND p.origin = 'tube:sxyprncom')
        AND NOT EXISTS (SELECT 1 FROM playback_sources p WHERE p.scene_id = s.id AND p.origin <> 'tube:sxyprncom')
        AND NOT EXISTS (SELECT 1 FROM scene_external_refs r JOIN sources so ON so.id = r.source_id
                        WHERE r.scene_id = s.id AND so.name IN ('tpdb', 'stashdb'))
    )
    SELECT sp.scene_id, sp.performer_id, p.canonical_name
    FROM only_sx o
    JOIN scene_performers sp ON sp.scene_id = o.id
    JOIN performers p ON p.id = sp.performer_id
    WHERE p.canonical_name !~ '\\s'
      AND position(lower(p.canonical_name) IN lower(o.title)) = 0
      AND NOT (length(p.canonical_name) >= 5
               AND position(regexp_replace(lower(p.canonical_name), '[^a-z0-9]', '', 'g')
                            IN regexp_replace(lower(o.title), '[^a-z0-9]', '', 'g')) > 0)
    """
)


def _chunks(seq: list, n: int):
    for i in range(0, len(seq), n):
        yield seq[i : i + n]


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--apply", action="store_true", help="bez tego tylko raport")
    ap.add_argument("--skip-junk", action="store_true", help="pomin krok 3 (smieciowi performerzy)")
    ap.add_argument("--only-junk", action="store_true", help="tylko krok 3, scen i zrodel nie ruszaj")
    ap.add_argument("--authors", default=",".join(AUTORZY), help="id autorow po przecinku")
    args = ap.parse_args()
    autorzy = [a.strip().lower() for a in args.authors.split(",") if a.strip()]
    dry = not args.apply
    print(f"tryb: {'DRY-RUN' if dry else 'APPLY'}; autorzy: {', '.join(AUTORZY.get(a, a) for a in autorzy)}")

    with session_scope() as s:
        rows = s.execute(_SQL_ZRODLA_AUTOROW, {"autorzy": autorzy}).all()
        zrodla = [r[0] for r in rows]
        per_autor: dict[str, set] = {}
        sceny_all: set = set()
        for _, scene_id, aid in rows:
            per_autor.setdefault(aid, set()).add(scene_id)
            sceny_all.add(scene_id)
        for aid, ids in per_autor.items():
            n_src = sum(1 for r in rows if r[2] == aid)
            print(f"  {AUTORZY.get(aid, aid):<8} zrodel/scen: {n_src}/{len(ids)}")
        sceny = sorted(sceny_all, key=str)
        print(f"lacznie: {len(zrodla)} zrodel, {len(sceny)} scen")

        do_usuniecia: list = []
        do_odciecia: list = []
        ulub = ogl = 0
        for chunk in _chunks(sceny, _CHUNK):
            for sid, inne, kanon, u, o in s.execute(
                _SQL_KLASYFIKACJA, {"sceny": chunk, "zrodla_autorow": zrodla}
            ).all():
                if inne or kanon:
                    do_odciecia.append(sid)
                else:
                    do_usuniecia.append(sid)
                    ulub += u
                    ogl += o
        print(
            f"do usuniecia: {len(do_usuniecia)} scen (ulubionych {ulub}, wpisow historii {ogl}); "
            f"do odciecia zrodla (scena zostaje): {len(do_odciecia)}"
        )

        smieci = [] if args.skip_junk else s.execute(_SQL_SMIECIOWI_PERF).all()
        if not args.skip_junk:
            print(
                f"smieciowe przypisania performerow na scenach tylko-sxyprn: {len(smieci)} "
                f"na {len({r[0] for r in smieci})} scenach"
            )
            top: dict[str, int] = {}
            for _, _, name in smieci:
                top[name] = top.get(name, 0) + 1
            najczestsi = sorted(top.items(), key=lambda x: -x[1])[:12]
            print("  najczestsi: " + ", ".join(f"{n} ({c})" for n, c in najczestsi))

        if dry:
            print("dry-run, nic nie zmieniam")
            return 0

        if args.only_junk:
            do_odciecia, do_usuniecia = [], []
            print("tylko krok 3, sceny i zrodla zostaja")

        # 2. sceny scalone: zostaja, wylatuje tylko zrodlo autora + tag Sizuka
        n_ps = 0
        for chunk in _chunks(do_odciecia, _CHUNK):
            n_ps += s.execute(
                text("DELETE FROM playback_sources WHERE scene_id = ANY(:sceny) AND id = ANY(:zrodla)"),
                {"sceny": chunk, "zrodla": zrodla},
            ).rowcount
            s.execute(
                text(
                    "DELETE FROM scene_tags st USING tags t "
                    "WHERE t.id = st.tag_id AND t.slug = 'sizuka' AND st.scene_id = ANY(:sceny)"
                ),
                {"sceny": chunk},
            )
        s.commit()
        print(f"odciete zrodla: {n_ps}")

        # 1. sceny wylacznie tych autorow: DELETE, kaskada robi reszte
        n_sc = 0
        for chunk in _chunks(do_usuniecia, _CHUNK):
            s.execute(
                text("DELETE FROM merge_candidates WHERE left_id = ANY(:sceny) OR right_id = ANY(:sceny)"),
                {"sceny": chunk},
            )
            n_sc += s.execute(text("DELETE FROM scenes WHERE id = ANY(:sceny)"), {"sceny": chunk}).rowcount
            s.commit()
        print(f"usuniete sceny: {n_sc}")

        # 3. smieciowi performerzy
        n_sp = 0
        pary = [(r[0], r[1]) for r in smieci]
        for chunk in _chunks(pary, _CHUNK):
            n_sp += s.execute(
                text(
                    "DELETE FROM scene_performers sp "
                    "USING unnest(CAST(:sc AS uuid[]), CAST(:pf AS uuid[])) AS u(sc, pf) "
                    "WHERE sp.scene_id = u.sc AND sp.performer_id = u.pf"
                ),
                {"sc": [c[0] for c in chunk], "pf": [c[1] for c in chunk]},
            ).rowcount
        s.commit()
        print(f"zdjete smieciowe przypisania: {n_sp}")

        # podpis Shizuki jako tag: bez scen nie ma czego trzymac
        s.execute(
            text(
                "DELETE FROM tags t WHERE t.slug = 'sizuka' "
                "AND NOT EXISTS (SELECT 1 FROM scene_tags st WHERE st.tag_id = t.id)"
            )
        )
        s.commit()
    return 0


if __name__ == "__main__":
    sys.exit(main())