"""Rozdziela filmy zlepione z kilku roznych tytulow (skutek starego scorera).

Tlo: `movie_score` do 2026-09-02 mierzyl tytul przez `token_set_ratio`, ktore daje pelne
100 dla tytulu bedacego podzbiorem drugiego i 90-96 dla numerowanych serii. Efekt:
"The Doctor" = 9 filmow z pandamovies pod jednym wpisem (zgloszenie 32100b94, "w
pierwszym linku granny porn"), "Her way to use Tinder 61" = 102 strony. W bazie 8204
filmy maja >= 2 refy z tego samego zrodla, lacznie 31 299 refow.

Dlaczego to musi pobierac strony na nowo: rekordy zewnetrzne filmow trzymaja tylko
`{slug, html_len}` (dooplay) albo `{hex_id, html_len}` (paradisehill), bez tytulu, roku
ani listy hosterow, a ingest filmow jest przyrostowy (paradisehill `since`, dooplay 3
strony), wiec skasowane rekordy nie odbuduja sie same. Jedyne zrodlo prawdy o tym, ktory
hoster nalezy do ktorej strony, to sama strona.

Algorytm dla kazdego podejrzanego filmu M:
1. Kazdy ref pobieramy na nowo przez konektor zrodla -> RawMovie (tytul, rok, studio,
   obsada, hostery). Zrodla bez konektora (tpdb, tube-scraper, yourdailypornmovies) i
   nieudane pobrania dostaja profil z samego sluga adresu (tylko tytul).
2. Refy laczymy w klastry (union-find): ten sam plik u hostera (wspolny adres embedu,
   BEZ prefiksu zrodla, bo ten sam plik siedzi pod mangoporn, pandamovies i streamporn)
   ALBO nowa ocena zlozona >= 0.85 z regulami numeru serii z `movie_score`.
3. Klaster z najstarszym refem zostaje filmem M (zachowuje id, ulubione i postep).
   Kazdy inny klaster dostaje nowy film; jego refy, hostery (po adresie embedu),
   obsada, tagi i rozdzialy przechodza tam.
4. M zostaje z hosterami potwierdzonymi przez swieze strony kotwicy. Reszta wierszy z
   rodzin luster (mangoporn/pandamovies/streamporn/paradisehill/freeomovie) leci, bo
   albo nalezy do odczepionego filmu, albo jest zwietrzala (pandamovies zrotowal
   hostery: swieza strona "Granny's Doctor Games" ma 0 z 4 kluczy w bazie). Wiersze z
   innych origin (kanoniczne strony studiow) zostaja. Obsada, tagi i rozdzialy M sa
   budowane od nowa z profili kotwicy, o ile choc jeden udalo sie pobrac.

Bezpieczniki: `--dry-run` tylko liczy i wypisuje klastry; `--movie-id` na jeden film;
`--after-id` do wznowienia (przebieg idzie ORDER BY id). Odpalaj z pliku, w osobnym
kontenerze (`docker compose run -d --rm ... python3 -u /srv/scripts/movie_unmerge.py`),
nie przez stdin (patrz browse_catchup.py) i nie w kontenerze workera.

Uzycie:
    python3 -u /srv/scripts/movie_unmerge.py --movie-id 6283c6cb-... --dry-run
    python3 -u /srv/scripts/movie_unmerge.py --limit 50 --dry-run
    python3 -u /srv/scripts/movie_unmerge.py
"""
from __future__ import annotations

import argparse
import logging
import re
import sys
import time
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from urllib.parse import urlparse

from sqlalchemy import text

from app.connectors.base import RawMovie
from app.connectors.dooplay import (
    MangopornConnector,
    PandamoviesConnector,
    StreampornConnector,
    StreampornVipConnector,
)
from app.connectors.freeomovie import FreeoMovieConnector
from app.connectors.paradisehill import ParadisehillConnector
from app.db import session_scope
from app.models.movie import Movie, MovieChapter, MoviePerformer, MovieTag
from app.normalize.movies import NormalizedMovie, normalize_movie
from app.normalize.text import normalize, slugify
from app.resolve.movie_resolver import (
    _create_canonical,
    _sync_chapters,
    _sync_performers,
    _sync_playback_sources,
    _sync_tags,
    _update_movie_fields,
)
from app.resolve.movie_score import _title_signal
from app.resolve.performer_resolver import resolve_performer
from app.resolve.studio_resolver import resolve_studio

log = logging.getLogger("movie-unmerge")

_SPACING_SEC = 0.4
_AUTO = 0.85
_WAGI = {"title": 0.50, "year": 0.20, "studio": 0.20, "cast": 0.10}

# Rodziny origin, ktorych wiersze na filmie-kotwicy wolno skasowac, gdy zadna swieza
# strona ich nie potwierdza. Inne origin (np. kanoniczne strony studiow) zostaja.
_RODZINY_LUSTER = ("mangoporn", "pandamovies", "streamporn", "streampornvip", "paradisehill", "freeomovie")

_SUFIKSY = (
    "-adult-movie-online-free", "-movie-online-free", "-porn-movie", "-online-free",
)


@dataclass
class Profil:
    source_name: str
    source_id: uuid.UUID
    external_id: str
    url: str | None
    first_seen: datetime
    raw: RawMovie | None = None
    norm: NormalizedMovie | None = None
    title_norm: str = ""
    year: int | None = None
    studio_norm: str | None = None
    cast: set[str] = field(default_factory=set)
    keys: set[str] = field(default_factory=set)   # adresy embedow / stron hosterow
    fetched: bool = False

    @property
    def klucz(self) -> tuple[uuid.UUID, str]:
        return (self.source_id, self.external_id)


def _tytul_ze_sluga(url: str | None, external_id: str) -> str:
    seg = ""
    if url:
        seg = urlparse(url).path.rstrip("/").split("/")[-1]
    if not seg or re.fullmatch(r"[0-9a-f]{8,}", seg):
        seg = external_id.split(":")[-1]
    seg = seg.lower()
    for pre in ("watch-xxx-", "watch-"):
        if seg.startswith(pre):
            seg = seg[len(pre):]
    for suf in _SUFIKSY:
        if seg.endswith(suf):
            seg = seg[: -len(suf)]
    # `.../2768634/supermodel-sluts-9-po` -> bez numerycznego id na poczatku
    seg = re.sub(r"^\d{5,}-", "", seg)
    return normalize(seg.replace("-", " ").replace("_", " "))


_KONEKTORY: dict = {}


def _konektor(name: str):
    if name not in _KONEKTORY:
        klasy = {
            "paradisehill": ParadisehillConnector,
            "mangoporn": MangopornConnector,
            "pandamovies": PandamoviesConnector,
            "streamporn": StreampornConnector,
            "streampornvip": StreampornVipConnector,
            "freeomovie": FreeoMovieConnector,
        }
        _KONEKTORY[name] = klasy[name]() if name in klasy else None
    return _KONEKTORY[name]


def _pobierz(p: Profil) -> RawMovie | None:
    conn = _konektor(p.source_name)
    if conn is None:
        return None
    time.sleep(_SPACING_SEC)
    if p.source_name == "paradisehill":
        return conn._fetch_detail(p.external_id)
    if p.source_name == "freeomovie":
        if not p.url:
            return None
        return conn._parse_detail(p.url, _tytul_ze_sluga(p.url, p.external_id).title())
    url = p.url or f"{conn.base_url.rstrip('/')}/movies/{p.external_id}/"
    return conn._fetch_detail(url)


def _profil(p: Profil) -> Profil:
    try:
        raw = _pobierz(p)
    except Exception as e:  # 404, timeout, CF: profil z samego sluga
        log.info("  %s %s: pobranie padlo (%s), profil ze sluga", p.source_name, p.external_id, type(e).__name__)
        raw = None
    if raw is not None:
        p.raw = raw
        p.norm = normalize_movie(raw)
        p.title_norm = p.norm.title_normalized
        p.year = p.norm.release_year
        p.studio_norm = p.norm.studio.name_normalized if p.norm.studio else None
        p.cast = {q.canonical_name.lower() for q in p.norm.performers if q.canonical_name}
        p.keys = {s.page_url for s in raw.playback_sources if s.page_url}
        p.fetched = True
    else:
        p.title_norm = _tytul_ze_sluga(p.url, p.external_id)
    if p.url:
        p.keys.add(p.url)   # wiersz "landing" ma page_url == adres strony
    return p


def _ten_sam_film(a: Profil, b: Profil) -> bool:
    """Odbicie `score_movie_candidate` w pamieci, z tymi samymi wagami i odrzuceniami."""
    if a.keys & b.keys & _bez_landingow(a, b):
        return True
    if not a.title_norm or not b.title_norm:
        return False
    ocena, odrzut, _ = _title_signal(a.title_norm, b.title_norm)
    if odrzut:
        return False
    dost = {"title": ocena}
    rok = None
    if a.year and b.year:
        d = abs(a.year - b.year)
        rok = 1.0 if d == 0 else 0.7 if d == 1 else 0.4 if d == 2 else 0.0
        dost["year"] = rok
    studio = None
    if a.studio_norm and b.studio_norm:
        studio = a.studio_norm == b.studio_norm
        dost["studio"] = 1.0 if studio else 0.0
    if a.cast and b.cast:
        dost["cast"] = len(a.cast & b.cast) / len(a.cast | b.cast)
    if studio is False and rok is not None and rok < 0.4:
        return False
    suma_w = sum(_WAGI[k] for k in dost)
    wynik = sum(v * _WAGI[k] / suma_w for k, v in dost.items())
    return wynik >= _AUTO


def _bez_landingow(a: Profil, b: Profil) -> set[str]:
    """Wspolne klucze licza sie tylko dla hosterow, nie dla adresow stron (te sa unikalne)."""
    return (a.keys | b.keys) - {a.url, b.url}


def _klastry(profile: list[Profil]) -> list[list[Profil]]:
    rodzic = list(range(len(profile)))

    def find(i):
        while rodzic[i] != i:
            rodzic[i] = rodzic[rodzic[i]]
            i = rodzic[i]
        return i

    for i in range(len(profile)):
        for j in range(i + 1, len(profile)):
            if find(i) != find(j) and _ten_sam_film(profile[i], profile[j]):
                rodzic[find(j)] = find(i)
    grupy: dict[int, list[Profil]] = {}
    for i, p in enumerate(profile):
        grupy.setdefault(find(i), []).append(p)
    return list(grupy.values())


def _wzorzec(klaster: list[Profil]) -> Profil | None:
    """Profil, z ktorego bierzemy pola kanoniczne: paradisehill, potem pierwszy pobrany."""
    for p in klaster:
        if p.fetched and p.source_name == "paradisehill":
            return p
    for p in klaster:
        if p.fetched:
            return p
    return None


def _opis(klaster: list[Profil]) -> str:
    return "; ".join(f"{p.source_name}:{p.title_norm[:34]}" for p in klaster[:6]) + (" ..." if len(klaster) > 6 else "")


def _istniejacy_film(
    s, *, title_normalized: str, release_year: int | None, studio_id, performer_ids: list, wyklucz: set,
) -> Movie | None:
    """Film, do ktorego resolver dolaczylby ten tytul automatycznie (>= 0.85), albo None.

    Ta sama para `find_movie_candidates` + `score_movie_candidate`, ktorej uzywa ingest,
    z PELNA lista performer_id (jak w `resolve_movie`), a nie tylko z tymi, ktore pasuja
    do kandydata: pierwsza wersja podawala sam przekroj, przez co Jaccard byl gornym
    oszacowaniem (3 z 5 wspolnych dawalo 1.0 zamiast 0.6) i skrypt laczyl chetniej niz
    ingest. Kandydaci z trigramow PLUS dokladne trafienie w `title_normalized`, bo
    `find_movie_candidates` bierze 50 losowych sasiadow i przy popularnych tytulach
    ("control 2") potrafi pominac ten wlasciwy.
    """
    from app.resolve.movie_match import find_movie_candidates
    from app.resolve.movie_score import AUTO_THRESHOLD, _title_signal, score_movie_candidate

    s.flush()  # blokowanie po studiu i roku czyta kolumny, ktore mogly zmienic sie bez flush
    kand = {c.id: c for c in find_movie_candidates(
        s, title_normalized=title_normalized, studio_id=studio_id, release_year=release_year
    )}
    for c in s.query(Movie).filter(Movie.title_normalized == title_normalized).all():
        kand.setdefault(c.id, c)
    najlepszy, najlepsza = None, 0.0
    for c in kand.values():
        if c.id in wyklucz:
            continue
        # Tytul jest czysto pythonowy i przy sufitach 0.6 nie da sie dojsc do 0.85, wiec
        # obsady z bazy nie ma po co czytac dla takich kandydatow.
        ocena, odrzut, _ = _title_signal(c.title_normalized or "", title_normalized)
        if odrzut or ocena < 0.7:
            continue
        br = score_movie_candidate(
            s, candidate=c, norm_title_normalized=title_normalized,
            norm_release_year=release_year, norm_studio_id=studio_id, norm_performer_ids=performer_ids,
        )
        if br.composite > najlepsza:
            najlepszy, najlepsza = c, br.composite
            if najlepsza >= 1.0:
                break
    return najlepszy if najlepsza >= AUTO_THRESHOLD else None


def _performer_ids_po_nazwach(s, nazwy: set[str]) -> list:
    """Do PODGLADU (dry-run): id performerow po nazwie kanonicznej, bez zapisu."""
    if not nazwy:
        return []
    return [
        pid for (pid,) in s.execute(
            text("SELECT id FROM performers WHERE lower(canonical_name) = ANY(:n)"), {"n": list(nazwy)}
        ).fetchall()
    ]


def _refy_bez_zmian(s, movie_id: uuid.UUID, profile: list[Profil]) -> bool:
    """Czy zbior refow filmu jest ten sam, co przy wczytaniu (inny shard mogl w miedzyczasie
    dolaczyc tu klaster). Przepisanie kotwicy na starych profilach skasowaloby jego
    hostery i obsade jako "niepotwierdzone", wiec taki film pomijamy do nastepnego przebiegu."""
    teraz = {(sid, e) for (sid, e) in s.execute(
        text("SELECT source_id, external_id FROM movie_external_refs WHERE movie_id = :m"), {"m": str(movie_id)}
    ).fetchall()}
    return teraz == {p.klucz for p in profile}


def _zaladuj_refy(s, movie_id: uuid.UUID) -> list[Profil]:
    rows = s.execute(
        text(
            """
            SELECT src.name, r.source_id, r.external_id, r.url, r.first_seen
              FROM movie_external_refs r JOIN sources src ON src.id = r.source_id
             WHERE r.movie_id = :m ORDER BY r.first_seen
            """
        ),
        {"m": str(movie_id)},
    ).fetchall()
    return [Profil(source_name=n, source_id=sid, external_id=e, url=u, first_seen=fs) for n, sid, e, u, fs in rows]


def _przypnij_zawartosc(s, movie_id: uuid.UUID, p: Profil) -> None:
    """Obsada, tagi, rozdzialy i hostery z profilu na wskazany film (jak w resolverze)."""
    assert p.norm is not None
    resolved = []
    for pn in p.norm.performers:
        perf = resolve_performer(s, norm=pn, source_id=p.source_id)
        resolved.append((perf.id, pn.as_alias_in_scene))
    _sync_performers(s, movie_id=movie_id, resolved=resolved)
    _sync_tags(s, movie_id=movie_id, norm=p.norm, source_id=p.source_id)
    _sync_chapters(s, movie_id=movie_id, norm=p.norm)
    _sync_playback_sources(s, movie_id=movie_id, norm=p.norm)
    # Klaster to zwykle kilka luster tego samego filmu z ta sama obsada. Helpery
    # resolvera sprawdzaja "czy juz jest" SELECT-em, ktory bez flush nie widzi wierszy
    # dodanych przez poprzedni profil, wiec ten sam performer wpadal dwa razy i padal
    # klucz glowny movie_performers (pierwszy prawdziwy przebieg na "The Doctor").
    s.flush()


def _przetworz(movie_id: uuid.UUID, *, dry_run: bool, liczniki: dict) -> None:
    with session_scope() as s:
        m = s.get(Movie, movie_id)
        if m is None:
            return
        tytul_m = m.title
        profile = _zaladuj_refy(s, movie_id)
    if len(profile) < 2:
        return
    profile = [_profil(p) for p in profile]
    klastry = _klastry(profile)
    klastry.sort(key=lambda k: min(p.first_seen for p in k))
    kotwica, inne = klastry[0], klastry[1:]
    liczniki["filmy"] += 1
    liczniki["pobrane"] += sum(1 for p in profile if p.fetched)
    liczniki["niepobrane"] += sum(1 for p in profile if not p.fetched)
    if not inne:
        log.info("OK   %-40s refy=%d, jeden klaster, bez zmian", tytul_m[:40], len(profile))
        return
    liczniki["rozdzielone"] += 1
    log.info("SPLIT %-40s refy=%d -> %d klastrow", tytul_m[:40], len(profile), len(klastry))
    log.info("   kotwica: %s", _opis(kotwica))
    if dry_run:
        # Podglad ma pokazac TO SAMO, co zrobi przebieg: takze dolaczanie do istniejacych.
        # Bez tego pierwsza wersja dry-run wypisywala "nowy" dla kazdego klastra i nie
        # dalo sie z gory zobaczyc 798 dubli, ktore potem powstaly.
        with session_scope() as s:
            for k in inne:
                wz = _wzorzec(k)
                istn = None
                if wz is not None and wz.norm is not None:
                    istn = _istniejacy_film(
                        s, title_normalized=wz.norm.title_normalized, release_year=wz.norm.release_year,
                        studio_id=None, performer_ids=_performer_ids_po_nazwach(s, set().union(*(p.cast for p in k))),
                        wyklucz={movie_id},
                    )
                if istn is not None:
                    liczniki["dolaczone_do_istniejacych"] += 1
                    log.info("   dolaczy do %r: %s", istn.title[:36], _opis(k))
                else:
                    liczniki["nowe_filmy"] += 1
                    log.info("   nowy:    %s", _opis(k))
        return

    with session_scope() as s:
        m = s.get(Movie, movie_id)
        assert m is not None
        if not _refy_bez_zmian(s, movie_id, profile):
            log.warning("   %s: refy zmienily sie w trakcie pobierania (inny shard?), pomijam do nastepnego przebiegu", tytul_m[:40])
            liczniki["pominiete_wyscig"] = liczniki.get("pominiete_wyscig", 0) + 1
            return
        # Filmy utworzone w TYM podziale: kolejne klastry nie moga sie do nich dolaczac,
        # bo klastrowanie juz orzeklo, ze to rozne filmy; inaczej rozdzielone przed chwila
        # sklejalo sie z powrotem przez ocene z bazy (dopelnione pola, obsada po id).
        utworzone_tu: set = set()
        # --- nowe filmy dla klastrow poza kotwica --------------------------------
        for k in inne:
            wz = _wzorzec(k)
            if wz is not None and wz.norm is not None:
                studio = resolve_studio(s, norm=wz.norm.studio, source_id=wz.source_id) if wz.norm.studio else None
                # NAJPIERW istniejacy film, tym samym resolverem co ingest. Bez tego
                # pierwszy przebieg (2026-09-02) utworzyl 798 filmow o tytule, ktory juz
                # byl w bazie, i 408 dubli miedzy soba: "Control 2" powstawal na nowo,
                # chociaz "Control 2" z innego zrodla juz istnial (zgloszenie a5cdda7f).
                # Obsada rozwiazana tak jak w resolverze, pelna lista id.
                pids: list = []
                for p in k:
                    if p.norm is not None:
                        for pn in p.norm.performers:
                            pids.append(resolve_performer(s, norm=pn, source_id=p.source_id).id)
                istn = _istniejacy_film(
                    s, title_normalized=wz.norm.title_normalized, release_year=wz.norm.release_year,
                    studio_id=studio.id if studio else None, performer_ids=list(dict.fromkeys(pids)),
                    wyklucz={movie_id} | utworzone_tu,
                )
                if istn is not None:
                    nowy = istn
                    liczniki["dolaczone_do_istniejacych"] += 1
                    log.info("   -> istniejacy film %s %r", nowy.id, nowy.title)
                else:
                    nowy = _create_canonical(s, norm=wz.norm, studio_id=studio.id if studio else None)
                    utworzone_tu.add(nowy.id)
                    liczniki["nowe_filmy"] += 1
            else:
                # Zaden ref klastra nie dal sie pobrac: film z samego tytulu ze sluga,
                # zeby refy nie zostaly na zlym filmie. Reszta doladuje sie przy ingescie.
                t = (k[0].title_norm or "untitled").title()
                nowy = Movie(title=t, title_normalized=normalize(t), slug=slugify(t))
                s.add(nowy)
                s.flush()
                utworzone_tu.add(nowy.id)
                liczniki["nowe_filmy"] += 1
            klucze = set().union(*(p.keys for p in k))
            # Rozdzielony film to STARA tresc, nie nowy nabytek: created_at = pierwsze
            # zobaczenie jego refow, nie "teraz". Inaczej kilka tysiecy filmow z dzisiejsza
            # data wpadlo w badge "+N nowych" u kazdego, kto ma performera w ulubionych,
            # a na liscie scen nie bylo nic nowego (zgloszenie 1db95176, 2026-09-02).
            najstarszy = min(p.first_seen for p in k)
            if nowy.created_at is None or nowy.created_at > najstarszy:
                nowy.created_at = najstarszy
            for p in k:
                s.execute(
                    text("UPDATE movie_external_refs SET movie_id=:n WHERE source_id=:sid AND external_id=:e"),
                    {"n": str(nowy.id), "sid": str(p.source_id), "e": p.external_id},
                )
            if klucze:
                r = s.execute(
                    text("UPDATE movie_playback_sources SET movie_id=:n WHERE movie_id=:m AND page_url = ANY(:k)"),
                    {"n": str(nowy.id), "m": str(movie_id), "k": list(klucze)},
                )
                liczniki["hostery_przeniesione"] += r.rowcount
            for p in k:
                if p.norm is not None:
                    _przypnij_zawartosc(s, nowy.id, p)
                    # Dopelnianie pol z kazdego lustra (tylko braki): wzorcem bywa
                    # paradisehill bez roku, a mangoporn ma rok i studio.
                    st = resolve_studio(s, norm=p.norm.studio, source_id=p.source_id) if p.norm.studio else None
                    _update_movie_fields(nowy, p.norm, studio_id=st.id if st else None)
            liczniki["refy_przeniesione"] += len(k)
            log.info("   -> nowy film %s %r", nowy.id, nowy.title)

        # --- kotwica: zostaje to, co potwierdzaja swieze strony ------------------
        klucze_k = set().union(*(p.keys for p in kotwica))
        wzory = [p for p in kotwica if p.norm is not None]
        r = s.execute(
            text(
                """
                DELETE FROM movie_playback_sources
                 WHERE movie_id = :m AND NOT (page_url = ANY(:k))
                   AND split_part(origin, ':', 1) = ANY(:rodz)
                """
            ),
            {"m": str(movie_id), "k": list(klucze_k) or [""], "rodz": list(_RODZINY_LUSTER)},
        )
        liczniki["hostery_skasowane"] += r.rowcount
        if wzory:
            for tabela in (MoviePerformer, MovieTag, MovieChapter):
                s.query(tabela).filter(tabela.movie_id == movie_id).delete(synchronize_session=False)
            wz = _wzorzec(kotwica)
            assert wz is not None and wz.norm is not None
            studio = resolve_studio(s, norm=wz.norm.studio, source_id=wz.source_id) if wz.norm.studio else None
            m.title = wz.norm.title
            m.title_normalized = wz.norm.title_normalized
            m.slug = wz.norm.slug or slugify(wz.norm.title)
            m.release_year = wz.norm.release_year
            if studio is not None:
                m.studio_id = studio.id
            _update_movie_fields(m, wz.norm, studio_id=studio.id if studio else None)
            for p in wzory:
                _przypnij_zawartosc(s, movie_id, p)
                st = resolve_studio(s, norm=p.norm.studio, source_id=p.source_id) if p.norm.studio else None
                _update_movie_fields(m, p.norm, studio_id=st.id if st else None)
        log.info("   kotwica zostaje jako %r (hostery skasowane: %d)", m.title, r.rowcount)


def _waga_filmu(s, movie_id: uuid.UUID) -> tuple:
    """Ktory z dwoch dubli zostaje: wiecej refow, potem starsze pierwsze zobaczenie, potem
    starsze created_at. NIE sama data utworzenia, bo rozdzielone filmy dostaja created_at
    z historii refow i "starszy" przestal znaczyc "istniejacy wczesniej"."""
    refy, fs = s.execute(
        text("SELECT count(*), min(first_seen) FROM movie_external_refs WHERE movie_id = :m"), {"m": str(movie_id)}
    ).fetchone()
    m = s.get(Movie, movie_id)
    return (-(refy or 0), fs or datetime.max.replace(tzinfo=UTC), m.created_at or datetime.max.replace(tzinfo=UTC))


def _sprzataj_duble(zmienione_po: datetime, *, dry_run: bool, after_id: str | None) -> dict:
    """Scala duble wsrod filmow ruszonych po `zmienione_po` (po `updated_at`, bo
    `created_at` rozdzielonych filmow jest cofane do historii refow i nie nadaje sie na
    znacznik "z tego przebiegu").

    Dla kazdego takiego filmu szuka partnera, do ktorego resolver dolaczylby go
    automatycznie (>= 0.85), i scala przez `merge_movies`. Zostaje film o wiekszej wadze
    (`_waga_filmu`), a `merge_movies` przenosi tez ulubione i postep, wiec kierunek nie
    kosztuje danych usera. Blad przy jednym filmie nie przerywa przebiegu; `--after-id`
    wznawia.
    """
    from app.resolve.movie_merge import merge_movies

    lic = {"sprawdzone": 0, "scalone": 0, "bledy": 0}
    sql = "SELECT id FROM movies WHERE updated_at > :t"
    params: dict = {"t": zmienione_po}
    if after_id:
        sql += " AND id > :aid"
        params["aid"] = after_id
    with session_scope() as s:
        ids = [r[0] for r in s.execute(text(sql + " ORDER BY id"), params).fetchall()]
    log.info("duble: filmow zmienionych po %s: %d", zmienione_po.isoformat(), len(ids))
    for i, mid in enumerate(ids, 1):
        try:
            with session_scope() as s:
                m = s.get(Movie, mid)
                if m is None:
                    continue
                lic["sprawdzone"] += 1
                pids = [pid for (pid,) in s.execute(
                    text("SELECT performer_id FROM movie_performers WHERE movie_id = :m"), {"m": str(mid)}
                ).fetchall()]
                istn = _istniejacy_film(
                    s, title_normalized=m.title_normalized, release_year=m.release_year,
                    studio_id=m.studio_id, performer_ids=pids, wyklucz={mid},
                )
                if istn is None:
                    continue
                keep, drop = (istn, m) if _waga_filmu(s, istn.id) <= _waga_filmu(s, m.id) else (m, istn)
                lic["scalone"] += 1
                log.info("DUBEL %r (%s) -> %r (%s)%s", drop.title[:40], str(drop.id)[:8], keep.title[:40], str(keep.id)[:8], " [dry]" if dry_run else "")
                if not dry_run:
                    merge_movies(s, keep_id=keep.id, drop_id=drop.id, resolved_by="movie_unmerge:dedupe")
        except Exception:
            lic["bledy"] += 1
            log.exception("duble: film %s, pomijam", mid)
        if i % 200 == 0 or i == len(ids):
            log.info("[duble %d/%d] %s (ostatni id %s)", i, len(ids), lic, mid)
            for h in logging.getLogger().handlers:
                h.flush()
    log.info("KONIEC duble: %s", lic)
    return lic


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--dedupe-updated-after", default=None,
                    help="ISO timestamp (bez strefy = UTC): zamiast rozdzielania scal duble wsrod "
                         "filmow zmienionych po tej chwili (po updated_at)")
    ap.add_argument("--movie-id", default=None)
    ap.add_argument("--limit", type=int, default=0)
    ap.add_argument("--after-id", default=None)
    ap.add_argument("--shard", default=None, help="k/n: tylko filmy z hashtext(id) %% n == k (rownolegle kontenery)")
    ap.add_argument("--dry-run", action="store_true")
    a = ap.parse_args(argv)

    logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", stream=sys.stdout)
    for noisy in ("httpx", "httpcore", "urllib3", "curl_cffi"):
        logging.getLogger(noisy).setLevel(logging.WARNING)

    if a.dedupe_updated_after:
        t = datetime.fromisoformat(a.dedupe_updated_after)
        if t.tzinfo is None:
            t = t.replace(tzinfo=UTC)  # inaczej Postgres wzialby strefe sesji i okno by sie przesunelo
        _sprzataj_duble(t, dry_run=a.dry_run, after_id=a.after_id)
        return 0

    if a.movie_id:
        ids = [uuid.UUID(a.movie_id)]
    else:
        sql = """
            SELECT DISTINCT movie_id FROM movie_external_refs
             WHERE movie_id IN (SELECT movie_id FROM movie_external_refs
                                 GROUP BY movie_id, source_id HAVING count(*) >= 2)
        """
        params: dict = {}
        if a.after_id:
            sql += " AND movie_id > :aid"
            params["aid"] = a.after_id
        if a.shard:
            k, n = (int(x) for x in a.shard.split("/"))
            sql += " AND abs(hashtext(movie_id::text)) % :n = :k"
            params.update({"k": k, "n": n})
        sql += " ORDER BY movie_id"
        if a.limit:
            sql += f" LIMIT {int(a.limit)}"
        with session_scope() as s:
            ids = [r[0] for r in s.execute(text(sql), params).fetchall()]
    log.info("filmow do sprawdzenia: %d%s", len(ids), " (DRY RUN)" if a.dry_run else "")

    liczniki = {k: 0 for k in ("filmy", "rozdzielone", "nowe_filmy", "dolaczone_do_istniejacych", "refy_przeniesione", "hostery_przeniesione", "hostery_skasowane", "pobrane", "niepobrane")}
    t0 = time.time()
    for i, mid in enumerate(ids, 1):
        try:
            _przetworz(mid, dry_run=a.dry_run, liczniki=liczniki)
        except Exception:
            log.exception("film %s: blad, pomijam", mid)
            liczniki.setdefault("bledy", 0)
            liczniki["bledy"] += 1
        if i % 25 == 0 or i == len(ids):
            log.info("[%d/%d] %s (%.0f min, ostatni id %s)", i, len(ids), liczniki, (time.time() - t0) / 60, mid)
            for h in logging.getLogger().handlers:
                h.flush()
    log.info("KONIEC: %s | %.0f min", liczniki, (time.time() - t0) / 60)
    return 0


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