"""Dedup w obrebie studia po rdzeniu tytulu.

Kazdy tube zapisuje tytul tej samej sceny po swojemu: "[Milfy] Koda Monroe (Voluptuous
Therapist Fucks Prisoner)", "Milfy 26 09 16 Koda Monroe Voluptuous Therapist Fucks Prisoner",
"Koda Monroe Voluptuous Therapist Fucks Prisoner 4K". Miniatury sa z innych klatek, wiec
phash sie nie pokrywa, a title_duration_dedup wymaga identycznego tytulu. Skutek: ta sama
scena Milfy wisiala w katalogu 8 razy (zgloszenie bb1a357d, 2026-09-18).

Rdzen tytulu = slowa bez nazwy studia, dat, znacznikow jakosci, lacznikow i nazwisk obsady
(suma obsady obu porownywanych scen, bo tube czesto nie ma obsady, a nazwisko ma w tytule).
Pojedyncze liczby ZOSTAJA: to numery czesci ("Whip 2", "Part 1"), ktore pierwsza wersja
reguly sklejala w jedno (suchy przebieg 2026-09-18).

Para jest ta sama scena, gdy to samo studio, rowne rdzenie i nic jej nie wyklucza:
  - dlugosci (gdy obie znane) roznia sie o <= 5 s,
  - daty premiery (gdy obie znane i zadna nie jest samym rokiem, czyli 1 stycznia) o <= 7 dni,
  - bez obu dlugosci rdzen musi miec >= 4 slowa, bo zostaje sam tytul,
  - obsady (gdy obie niepuste) maja wspolna osobe.
Grupy skladamy przez sume par, ale scalamy tylko sceny zgodne BEZPOSREDNIO z zostawiana,
zeby lancuch A~B~C nie wciagnal C, ktore z A sie nie zgadza (inna dlugosc przez scene bez
dlugosci w srodku).

Kopie bez studia (tube nie podal wytworni, 3 z 8 kopii Milfy) dopinamy do scen studyjnych
ostrzej: obie dlugosci znane i rowne co do 3 s, rdzen >= 4 slowa. Dwie sceny bez studia nie
maja wspolnej kotwicy, wiec tu sie nie lacza (identyczne tytuly zalatwia title_duration_dedup).

Okno czasowe, bo dupy splywaja z tubow w ciagu dni od premiery; cyklicznie w schedulerze.
"""
from __future__ import annotations

import collections
import logging
import re
import unicodedata
import uuid as _u
from dataclasses import dataclass
from datetime import date, datetime

from sqlalchemy import text

from app.db import session_scope
from app.resolve.scene_merge import merge_scenes

log = logging.getLogger(__name__)

_QUALITY = {"4k", "8k", "uhd", "fhd", "hd", "sd", "2160p", "1440p", "1080p", "720p", "480p", "360p", "full", "mp4", "vr"}
_CONNECT = {"and", "x", "ft", "feat", "featuring", "starring", "with"}
# Data wpisana w tytul przez tube: "26 09 16", "(17/9/26)", "21-08-2026", "2026.09.16".
_DATE_RE = re.compile(r"(?<![0-9])[0-9]{1,4}[ ./_-][0-9]{1,2}[ ./_-][0-9]{1,4}(?![0-9])")
_MAX_DURATION_DIFF = 5
_MAX_RELEASE_DIFF_DAYS = 7
# Kubelek (studio + ostatnie slowo rdzenia) wiekszy od tego to generyczny tytul w studiu-
# zbieraczu ("Only Fans"), gdzie porownanie parami jest drogie i ryzykowne.
_MAX_BUCKET = 400
_JUNK_TAG_MIN_SCENES = 1000
_ORPHAN_MAX_DURATION_DIFF = 3
_ORPHAN_MIN_CORE = 4


@dataclass(frozen=True)
class Cand:
    id: str
    title: str
    studio_id: str
    studio_name: str
    studio_slug: str
    duration_sec: int | None
    release_date: date | None
    created_at: datetime
    perfs: tuple[str, ...]
    canon: bool
    live: int


def title_words(title: str) -> list[str]:
    t = unicodedata.normalize("NFKD", title or "").encode("ascii", "ignore").decode().lower()
    t = _DATE_RE.sub(" ", t)
    return re.sub(r"[^a-z0-9]+", " ", t).split()


def studio_tokens(name: str, slug: str) -> set[str]:
    words = title_words(name)
    # "[BrazzersExxtra]" i "Brazzers Exxtra" to to samo studio.
    return set(words) | set(title_words(slug)) | {"".join(words)}


def title_core(title: str, st_tokens: set[str], perf_tokens: set[str]) -> tuple[str, ...]:
    return tuple(
        w for w in title_words(title)
        if w not in _QUALITY and w not in _CONNECT and w not in st_tokens and w not in perf_tokens
    )


def _year_only(d: date | None) -> bool:
    # Czesc tubow podaje sam rok, zapisany jako 1 stycznia. Taka data niczego nie wyklucza.
    return d is not None and d.month == 1 and d.day == 1


def same_scene(a: Cand, b: Cand, st_tokens: set[str], junk_names: frozenset[str] = frozenset()) -> bool:
    if a.studio_id and b.studio_id and a.studio_id != b.studio_id:
        return False
    if not a.studio_id and not b.studio_id:
        return False
    orphan = not (a.studio_id and b.studio_id)
    perf_tokens = {
        w for n in a.perfs + b.perfs if n.lower() not in junk_names for w in title_words(n)
    }
    ca = title_core(a.title, st_tokens, perf_tokens)
    if len(ca) < (_ORPHAN_MIN_CORE if orphan else 3) or ca != title_core(b.title, st_tokens, perf_tokens):
        return False
    if a.duration_sec and b.duration_sec:
        if abs(a.duration_sec - b.duration_sec) > (_ORPHAN_MAX_DURATION_DIFF if orphan else _MAX_DURATION_DIFF):
            return False
    elif orphan or len(ca) < 4:
        return False
    ra, rb = a.release_date, b.release_date
    if ra and rb and not _year_only(ra) and not _year_only(rb):
        if abs((ra - rb).days) > _MAX_RELEASE_DIFF_DAYS:
            return False
    if a.perfs and b.perfs and not (set(a.perfs) & set(b.perfs)):
        return False
    return True


def _keep_rank(c: Cand) -> tuple:
    # Zostaje scena z kanonem (TPDB/StashDB), potem z najwieksza liczba zywych zrodel.
    return (not c.canon, not c.studio_id, -c.live, not c.perfs, c.created_at)


def plan_merges(cands: list[Cand], junk_names: frozenset[str] = frozenset()) -> list[tuple[Cand, list[Cand]]]:
    """Zwraca [(zostaje, [do_scalenia...])]. Czysta funkcja, bez bazy."""
    by_studio: dict[str, list[Cand]] = collections.defaultdict(list)
    orphans: list[Cand] = []
    for c in cands:
        if c.studio_id:
            by_studio[c.studio_id].append(c)
        elif c.duration_sec:
            orphans.append(c)

    parent: dict[str, str] = {}

    def find(x: str) -> str:
        while parent.setdefault(x, x) != x:
            parent[x] = parent[parent[x]]
            x = parent[x]
        return x

    tokens_by_studio: dict[str, set[str]] = {}
    by_id: dict[str, Cand] = {}
    for sid, items in by_studio.items():
        st = studio_tokens(items[0].studio_name, items[0].studio_slug)
        tokens_by_studio[sid] = st
        buckets: dict[str, list[Cand]] = collections.defaultdict(list)
        for c in items:
            core = title_core(c.title, st, set())
            if core:
                buckets[core[-1]].append(c)
        for bucket in buckets.values():
            if len(bucket) < 2 or len(bucket) > _MAX_BUCKET:
                continue
            for i, a in enumerate(bucket):
                for b in bucket[i + 1:]:
                    if same_scene(a, b, st, junk_names):
                        parent[find(a.id)] = find(b.id)
                        by_id[a.id], by_id[b.id] = a, b

    # Kopie bez studia: kandydaci z tym samym ostatnim slowem rdzenia i dlugoscia w tym samym
    # albo sasiednim przedziale 10 s. Scena studyjna trafia pod oba warianty rdzenia, bo nazwa
    # studia bywa na koncu tytulu ("Hotwife Hall Pass - TouchMyWife").
    index: dict[tuple[str, int], list[Cand]] = collections.defaultdict(list)
    for sid, items in by_studio.items():
        st = tokens_by_studio[sid]
        for c in items:
            if not c.duration_sec:
                continue
            for core in {title_core(c.title, st, set()), title_core(c.title, set(), set())}:
                if core:
                    index[(core[-1], c.duration_sec // 10)].append(c)
    for x in orphans:
        core = title_core(x.title, set(), set())
        if not core:
            continue
        seen: set[str] = set()
        slot = x.duration_sec // 10
        for key in ((core[-1], slot - 1), (core[-1], slot), (core[-1], slot + 1)):
            for y in index.get(key, ()):
                if y.id in seen:
                    continue
                seen.add(y.id)
                if same_scene(x, y, tokens_by_studio[y.studio_id], junk_names):
                    parent[find(x.id)] = find(y.id)
                    by_id[x.id], by_id[y.id] = x, y

    groups: dict[str, list[Cand]] = collections.defaultdict(list)
    for cid, c in by_id.items():
        groups[find(cid)].append(c)

    plan: list[tuple[Cand, list[Cand]]] = []
    for members in groups.values():
        keep = min(members, key=_keep_rank)
        st = tokens_by_studio[keep.studio_id]
        drops = [
            m for m in members
            if m.id != keep.id
            and (m.live or keep.live)  # dwie niewidoczne w apce sceny zostawiamy bulk_dedup
            and same_scene(keep, m, st, junk_names)
        ]
        if drops:
            plan.append((keep, drops))
    return plan


def _load(days: int) -> tuple[list[Cand], frozenset[str]]:
    sql = text("""
        SELECT s.id::text, s.title, s.studio_id::text, st.name, st.slug, s.duration_sec,
               s.release_date, s.created_at,
               COALESCE((SELECT array_agg(p.canonical_name) FROM scene_performers sp
                         JOIN performers p ON p.id = sp.performer_id
                         WHERE sp.scene_id = s.id), '{}') AS perfs,
               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.kind IN ('tpdb', 'stashdb')) AS canon,
               (SELECT count(*) FROM playback_sources p
                WHERE p.scene_id = s.id AND p.dead_at IS NULL) AS live
        FROM scenes s LEFT JOIN studios st ON st.id = s.studio_id
        WHERE (s.created_at > now() - make_interval(days => :d) OR s.release_date > current_date - :d)
          AND (s.studio_id IS NOT NULL OR s.duration_sec IS NOT NULL)
    """)
    with session_scope() as s:
        rows = s.execute(sql, {"d": days}).all()
        # Tube wpisuja czasem kategorie jako obsade ("Big Tits", "Deep Throat"). Takie
        # "nazwiska" wycinalyby z rdzenia zwykle slowa tytulu, wiec ich nie uzywamy. Sama
        # obecnosc wsrod tagow nie wystarcza: hashtagi z nazwiskami tez sa tagami ("Koda
        # Monroe" 51 scen, "Angela White" 239), a kategorie maja ich tysiace ("Ball Licking"
        # 11 tys., "Big Tits" 685 tys.). Pomiar 2026-09-18.
        junk = frozenset(
            n.lower()
            for (n,) in s.execute(text("SELECT name FROM tags WHERE scene_count >= :m"), {"m": _JUNK_TAG_MIN_SCENES}).all()
            if n
        )
    cands = [
        Cand(
            id=r[0], title=r[1] or "", studio_id=r[2] or "", studio_name=r[3] or "", studio_slug=r[4] or "",
            duration_sec=r[5], release_date=r[6], created_at=r[7], perfs=tuple(r[8] or ()),
            canon=bool(r[9]), live=int(r[10] or 0),
        )
        for r in rows
    ]
    return cands, junk


def run_studio_title_dedup(*, days: int = 45, commit: bool = True) -> dict[str, int]:
    """Scal dupy w obrebie studia po rdzeniu tytulu. Zwraca {groups, merges, merged, errors}."""
    cands, junk = _load(days)
    plan = plan_merges(cands, junk)
    merges = sum(len(d) for _, d in plan)
    merged = errors = 0
    if commit:
        for keep, drops in plan:
            for drop in drops:
                try:
                    with session_scope() as s:
                        merge_scenes(
                            s, keep_id=_u.UUID(keep.id), drop_id=_u.UUID(drop.id),
                            resolved_by="studio_title_dedup",
                        )
                    merged += 1
                except Exception as e:  # pragma: no cover - obronnie
                    errors += 1
                    if errors <= 20:
                        log.warning("studio_title_dedup keep %s drop %s: %s", keep.id[:8], drop.id[:8], str(e)[:120])
    result = {"candidates": len(cands), "groups": len(plan), "merges": merges, "merged": merged, "errors": errors}
    log.info("studio_title_dedup: %s", result)
    return result