"""Oznacza jako martwe źródła z rodziny seekplayer, których hoster już nie serwuje.

Dlaczego: zgłoszenia `d59fc396`, `ccbc75da` i sąsiednie z 2026-08-24 („Easyvidplayer
nie chce odpalić", „Embedseek też nie rusza"). Sprawdzone: to NIE jest nasz błąd
parsowania. Endpoint `/api/v1/video` odpowiada 200 i poprawnie się odszyfrowuje, tylko
w odpowiedzi nie ma pola `source`. Ich własny player w tej sytuacji pisze wprost
„Video is not ready yet", co potwierdziłem w przeglądarce z łącza domowego, i to samo
widać z VPS oraz z brytyjskiego proxy, więc nie chodzi o blokadę po IP ani po kraju.

Skala (próbka 45 źródeł, 2026-08-24): 26 razy HTTP 404, 18 razy odpowiedź bez strumienia,
1 raz działający strumień. Powtórka 2026-08-27 na sześciu markach: 0 grających na 36.

Bezpieczeństwo: filmów zależnych WYŁĄCZNIE od tej rodziny jest w bazie jeden, pozostałe
mają inne źródło. Sprzątanie nie chowa więc katalogu, tylko przestaje podsuwać wybór,
który i tak kończy się czarnym ekranem.

Oszczędność: domeny bliźniacze (`p.` i `vip.easyvidplayer.com`, `my.` i
`vip.player4me`) trzymają ten sam identyfikator wideo, więc pytamy raz na parę
(marka, hash) i werdykt stosujemy do wszystkich źródeł z tym kluczem.

Wznawianie za darmo: bierzemy tylko żywe źródła, więc kolejne odpalenie samo pomija
to, co już oznaczone. Żadnego pliku stanu.

STAN NA 2026-08-27: cała rodzina została wygaszona hurtowo dla FILMÓW (338 904 źródeł),
bo dowody były jednoznaczne: świeże identyfikatory prosto ze strony serwisu też nic nie
zwracają, 0 grających na 36 próbek w sześciu markach, a ich własny player kończy
komunikatem „Media source is not playable", który pada dopiero po wyczerpaniu WSZYSTKICH
czterech ścieżek CDN. Zostawiono 4 źródła jednego filmu, który nie miał nic innego, żeby
nic nie zniknęło z katalogu. Ten skrypt zostaje do punktowego sprawdzania i do scen.

Użycie (w kontenerze workera):
    python3 /srv/scripts/seekplayer_reap.py --limit 300 --dry-run
    python3 /srv/scripts/seekplayer_reap.py
"""
from __future__ import annotations

import argparse
import collections
import logging
import re
import sys
import time
from concurrent.futures import ThreadPoolExecutor
from urllib.parse import urlparse

from sqlalchemy import text

from app.db import session_scope
from app.extractors.hosters.seekplayer_engine import _HOST_RE

log = logging.getLogger("seekplayer-reap")

# Ta sama rodzina co `seekplayer_engine._HOST_RE`, w formie nadającej się do SQL-a.
_FAMILY_SQL = (
    "easyvidplayer|embedseek|seekplayer|upns|player4me|rpmplay"
    "|4meplayer|seekplays|ezplayer|seeks"
)
_CONCURRENCY = 3
# Odstęp przed każdym pobraniem. Bez niego przy pięciu wątkach 38% odpowiedzi to było
# HTTP 429, a limit nie jest dowodem na nic, więc takie źródło trzeba zostawić i wrócić
# do niego później. Wolniej, ale za to prawie każda próba coś rozstrzyga.
_SPACING_SEC = 0.4
_TIMEOUT = 20.0

_ALIVE, _DEAD, _UNKNOWN = "alive", "dead", "unknown"


def _brand(hostname: str) -> str:
    """`vip.easyvidplayer.com` → `easyvidplayer`. Klucz wspólny dla domen bliźniaczych."""
    m = re.search(_FAMILY_SQL, hostname)
    return m.group(0) if m else hostname


def _probe(item: tuple) -> tuple:
    """((brand, hash), verdict, detail). Sieć i parsowanie, bez dotykania bazy.

    Deleguje do `seekplayer_engine.extract`, ŻEBY WERDYKT BYŁ TEN SAM co u resolvera,
    z którego korzysta aplikacja. Wcześniej skrypt miał własną, luźniejszą regułę
    (są pola = żywe), przez co uznawał za żywe źródła z adresem do usuniętego pliku,
    na których CDN oddaje 502. Silnik sprawdza teraz każdą ścieżkę naprawdę.
    """
    key, url = item
    time.sleep(_SPACING_SEC)
    from app.extractors._models import HosterDead
    from app.extractors.hosters import seekplayer_engine

    try:
        out = seekplayer_engine.extract(url, timeout=_TIMEOUT)
    except HosterDead as e:
        return (key, _DEAD, str(e)[-80:])
    except Exception as e:
        # Rate limit i błędy sieci NIE są dowodem na nic: zostawiamy nietknięte.
        return (key, _UNKNOWN, type(e).__name__)
    if out:
        return (key, _ALIVE, (out[0].raw or {}).get("sciezka") or "")
    return (key, _UNKNOWN, "brak zrodla bez wyjatku")


def _load_work() -> tuple[dict, dict, dict]:
    """{(brand,hash): url}, {(brand,hash): [movie_src_id]}, {(brand,hash): [scene_src_id]}"""
    probes: dict = {}
    movie_ids: dict = collections.defaultdict(list)
    scene_ids: dict = collections.defaultdict(list)
    with session_scope() as s:
        for table, bucket in (
            ("movie_playback_sources", movie_ids),
            ("playback_sources", scene_ids),
        ):
            rows = s.execute(
                text(
                    f"""
                    SELECT id, coalesce(embed_url, page_url) AS u
                      FROM {table}
                     WHERE dead_at IS NULL
                       AND coalesce(embed_url, page_url) ~ :fam
                    """
                ),
                {"fam": _FAMILY_SQL},
            ).fetchall()
            for sid, u in rows:
                p = urlparse(u)
                # Regex w SQL to tylko tani przesiew i potrafi trafić w zwykły adres
                # tuba, gdy w slugu tytułu siedzi np. "seeks". Autorytetem jest
                # `_HOST_RE` z samego ekstraktora, ono patrzy wyłącznie na hosta.
                if not p.hostname or not p.fragment or not _HOST_RE.match(p.hostname):
                    continue
                key = (_brand(p.hostname), p.fragment.strip())
                probes.setdefault(key, u)
                bucket[key].append(sid)
    return probes, movie_ids, scene_ids


def _mark_dead(table: str, ids: list, reason: str) -> int:
    if not ids:
        return 0
    with session_scope() as s:
        return s.execute(
            text(
                f"UPDATE {table} SET dead_at = now(), dead_reason = :r "
                " WHERE id = ANY(:ids) AND dead_at IS NULL"
            ),
            {"r": f"seekplayer-reap: {reason}"[:512], "ids": ids},
        ).rowcount


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser()
    p.add_argument("--limit", type=int, default=0, help="ile kluczy sprawdzić (0 = wszystkie)")
    p.add_argument("--dry-run", action="store_true", help="nie zapisuj, tylko policz")
    a = p.parse_args(argv)

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

    probes, movie_ids, scene_ids = _load_work()
    keys = list(probes)
    if a.limit:
        keys = keys[: a.limit]
    log.info(
        "do sprawdzenia %d kluczy (marka+hash), zrodel: filmy %d, sceny %d",
        len(keys),
        sum(len(v) for v in movie_ids.values()),
        sum(len(v) for v in scene_ids.values()),
    )

    cnt: collections.Counter = collections.Counter()
    powody: collections.Counter = collections.Counter()
    nieznane: collections.Counter = collections.Counter()
    ubite_f = ubite_s = 0
    t0 = time.time()

    with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool:
        for i, (key, verdict, detail) in enumerate(
            pool.map(_probe, [(k, probes[k]) for k in keys]), 1
        ):
            cnt[verdict] += 1
            if verdict == _UNKNOWN:
                nieznane[detail[:40]] += 1
            if verdict == _DEAD:
                powody[detail[:40]] += 1
                if not a.dry_run:
                    ubite_f += _mark_dead("movie_playback_sources", movie_ids.get(key, []), detail)
                    ubite_s += _mark_dead("playback_sources", scene_ids.get(key, []), detail)
            if i % 250 == 0:
                log.info(
                    "[%d/%d] zywe=%d martwe=%d nieznane=%d, ubite zrodla: filmy %d sceny %d (%.0f min)",
                    i, len(keys), cnt[_ALIVE], cnt[_DEAD], cnt[_UNKNOWN],
                    ubite_f, ubite_s, (time.time() - t0) / 60,
                )

    log.info(
        "KONIEC: zywe=%d martwe=%d nieznane=%d | ubite zrodla: filmy %d sceny %d | %.0f min",
        cnt[_ALIVE], cnt[_DEAD], cnt[_UNKNOWN], ubite_f, ubite_s, (time.time() - t0) / 60,
    )
    log.info("powody martwych: %s", dict(powody.most_common(6)))
    log.info("powody nieznanych: %s", dict(nieznane.most_common(6)))
    return 0


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