"""Rozbija zlepionych wykonawcow typu "Cory Chase, Gigi Sweets" na prawdziwe osoby.

Scraper siski do 2026-09-27 (494b15a) bral pierwszy czlon alt='A, B - Tytul' jako jedno
nazwisko, wiec powstalo ~660 takich "osob". Kazda czesc przechodzi przez
resolve_performer (ta sama sciezka co ingest: nazwa, potem alias, na koncu nowa osoba),
powiazania ze scenami, filmami i ulubionymi przechodza na czesci, zlepek znika.
Czarne listy NIE przechodza: blokada "A, B" to nie blokada wszystkiego z A i z B.

Pomijamy nazwiska z ref do TPDB/StashDB (kanon nie ma przecinkow, wiec to bylby
swiadomy wyjatek) i takie, ktorych czesci nie wygladaja na nazwiska.

  docker compose run --rm worker python3 -u /srv/scripts/split_comma_performers.py         # na sucho
  docker compose run --rm worker python3 -u /srv/scripts/split_comma_performers.py --commit

--commit najpierw kopiuje zlepki i ich powiazania do tabel *_comma_20260927.
"""
from __future__ import annotations

import collections
import sys

from sqlalchemy import text

from app.connectors.base import RawPerformer
from app.db import session_scope
from app.normalize.scenes import normalize_performer
from app.resolve.performer_resolver import resolve_performer

COMMIT = "--commit" in sys.argv
BACKUP_SUFFIX = "comma_20260927"
# Tabele z (..., performer_id) w kluczu glownym, powiazania przenosimy na kazda czesc.
MOVE_TABLES = ("scene_performers", "movie_performers", "favorite_performers")
DROP_TABLES = ("blacklisted_performers",)


def parts_of(name: str) -> list[str] | None:
    parts = [p.strip() for p in name.split(",")]
    if len(parts) < 2 or len(parts) > 10 or any(not p for p in parts):
        return None
    for p in parts:
        words = p.split()
        # Cyfry tylko w jednowyrazowym nicku ("Subgirl0831"), nie w "Scene 2".
        if len(p) < 2 or len(words) > 4 or (len(words) > 1 and any(ch.isdigit() for ch in p)):
            return None
        # "Merce Palau aka Merce", "Ruby Reign. Yoya Grey": inny separator, zostawiamy.
        if "." in p or " aka " in f" {p.lower()} ":
            return None
    return parts


def two_single_words(parts: list[str]) -> bool:
    return len(parts) == 2 and all(len(p.split()) == 1 for p in parts)


def reversed_person(session, pid, parts: list[str]) -> str | None:
    """"Woods, Monique" to jedna osoba w zapisie nazwisko, imie, gdy "Monique Woods" jest
    w bazie i stoi w tytule ktorejs z jej scen. Samo istnienie nazwiska nie wystarcza:
    "Jenna, Chloe" przy "Wild Brunette Threesome" to raczej dwie osoby."""
    if not two_single_words(parts):
        return None
    name = f"{parts[1]} {parts[0]}"
    norm = normalize_performer(RawPerformer(name=name)).name_normalized
    hit = session.execute(text("""
        SELECT 1 FROM performers WHERE name_normalized = :n
        AND EXISTS (SELECT 1 FROM scene_performers sp JOIN scenes sc ON sc.id = sp.scene_id
                    WHERE sp.performer_id = :pid AND sc.title ILIKE :pat)
        LIMIT 1
    """), {"n": norm, "pid": pid, "pat": f"%{name}%"}).first()
    return name if hit else None


def columns(session, table: str) -> list[str]:
    return list(session.execute(text(
        "SELECT column_name FROM information_schema.columns "
        "WHERE table_schema = 'public' AND table_name = :t ORDER BY ordinal_position"
    ), {"t": table}).scalars())


with session_scope() as s:
    tube_source = s.execute(text("SELECT id FROM sources WHERE name = 'tube-scraper'")).scalar_one()
    canon_sources = list(s.execute(text(
        "SELECT id FROM sources WHERE kind IN ('tpdb', 'stashdb')"
    )).scalars())
    junk = s.execute(text("""
        SELECT p.id, p.canonical_name FROM performers p
        WHERE p.canonical_name LIKE '%,%'
          AND NOT EXISTS (
              SELECT 1 FROM performer_external_refs r
              WHERE r.performer_id = p.id AND r.source_id = ANY(:canon))
        ORDER BY p.canonical_name
    """), {"canon": canon_sources}).all()
    print(f"zlepkow z przecinkiem bez refu kanonu: {len(junk)}", flush=True)

    usage = collections.Counter()
    for t in MOVE_TABLES + DROP_TABLES:
        usage[t] = s.execute(text(
            f"SELECT count(*) FROM {t} WHERE performer_id = ANY(:ids)"
        ), {"ids": [j.id for j in junk]}).scalar()
    print("powiazania:", dict(usage), flush=True)

    plan: list[tuple] = []
    skipped: list[str] = []
    for pid, name in junk:
        parts = parts_of(name)
        if parts is None:
            skipped.append(name)
            continue
        whole = reversed_person(s, pid, parts)
        if whole:
            print(f"  nazwisko, imie: {name!r} -> {whole!r}")
            parts = [whole]
        elif two_single_words(parts) and s.execute(
            text("SELECT 1 FROM performers WHERE name_normalized = :n LIMIT 1"),
            {"n": normalize_performer(RawPerformer(name=f"{parts[1]} {parts[0]}")).name_normalized},
        ).first():
            # Odwrocone nazwisko istnieje, ale tytul go nie potwierdza: nie zgadujemy.
            skipped.append(name)
            continue
        plan.append((pid, name, parts))
    print(f"do rozbicia: {len(plan)}, pominiete: {len(skipped)}", flush=True)
    for name in skipped[:30]:
        print("  pomijam:", name)

    existing = new = 0
    samples = []
    targets: dict = {}
    for pid, name, parts in plan:
        ids = []
        for part in parts:
            norm = normalize_performer(RawPerformer(name=part))
            hit = s.execute(text(
                "SELECT id FROM performers WHERE name_normalized = :n "
                "UNION ALL SELECT performer_id FROM performer_aliases WHERE alias_normalized = :n LIMIT 1"
            ), {"n": norm.name_normalized}).scalar()
            if hit is not None and hit != pid:
                existing += 1
            else:
                new += 1
            if COMMIT:
                ids.append(resolve_performer(s, norm=norm, source_id=tube_source).id)
            samples.append((name, part, "jest" if hit is not None and hit != pid else "NOWA"))
        targets[pid] = ids
    print(f"czesci: {existing} juz w bazie, {new} nowych osob", flush=True)
    for name, part, st in samples[:10] + [x for x in samples if x[2] == "NOWA"]:
        print(f"  {name!r} -> {part!r} ({st})")

    if not COMMIT:
        print("NA SUCHO, nic nie zmienione. --commit zeby wykonac.")
        sys.exit(0)

    ids = [pid for pid, _, _ in plan]
    s.execute(text(f"DROP TABLE IF EXISTS performers_{BACKUP_SUFFIX}"))
    s.execute(text(f"CREATE TABLE performers_{BACKUP_SUFFIX} AS SELECT * FROM performers WHERE id = ANY(:ids)"), {"ids": ids})
    for t in MOVE_TABLES + DROP_TABLES:
        s.execute(text(f"DROP TABLE IF EXISTS {t}_{BACKUP_SUFFIX}"))
        s.execute(text(f"CREATE TABLE {t}_{BACKUP_SUFFIX} AS SELECT * FROM {t} WHERE performer_id = ANY(:ids)"), {"ids": ids})

    moved = collections.Counter()
    for t in MOVE_TABLES:
        cols = columns(s, t)
        sel = ", ".join(":new" if c == "performer_id" else c for c in cols)
        for pid, _, _ in plan:
            for new_id in targets[pid]:
                if new_id == pid:
                    continue
                moved[t] += s.execute(text(
                    f"INSERT INTO {t} ({', '.join(cols)}) SELECT {sel} FROM {t} "
                    f"WHERE performer_id = :old ON CONFLICT DO NOTHING"
                ), {"new": new_id, "old": pid}).rowcount or 0
    deleted = s.execute(text("DELETE FROM performers WHERE id = ANY(:ids)"), {"ids": ids}).rowcount
    s.commit()
    print(f"przeniesione powiazania: {dict(moved)}, skasowane zlepki: {deleted}", flush=True)