"""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 losowych źródeł, 2026-08-24): 26 razy HTTP 404, 18 razy odpowiedź bez
strumienia, 1 raz działający strumień. Czyli około 2% katalogu na tym silniku jeszcze
gra, reszty hoster już nie ma. Sam protokół jest bez zmian, dlatego NIE kasujemy tych
źródeł hurtem, tylko pytamy o każde z osobna i zostawiamy te, które odpowiadają.

Bezpieczeństwo: filmów zależnych WYŁĄCZNIE od tej rodziny jest w bazie dokładnie
jeden, pozostałe 26 223 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.

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 json
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._fetch import _DEFAULT_UA, browser_get
from app.extractors.hosters.seekplayer_engine import _HOST_RE, _decrypt

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"
)
# Pola, w których silnik podaje adres strumienia. Nazwy wzięte z ich własnego bundla
# (`{Cloudflare: cf, Tiktok: hlsVideoTiktok, Google: hlsVideoGoogle, "In-House": source}`),
# a warunek „nie ma żadnego z nich" to dokładnie ten sam warunek, po którym ich player
# pokazuje „Video is not ready yet".
_STREAM_FIELDS = ("source", "cf", "hlsVideoGoogle", "hlsVideoTiktok", "cfNative", "p2pStream")

_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."""
    key, url = item
    time.sleep(_SPACING_SEC)
    p = urlparse(url)
    host = f"{p.scheme}://{p.hostname}"
    api = f"{host}/api/v1/video?id={p.fragment}&w=1920&h=1080&r="
    headers = {"User-Agent": _DEFAULT_UA, "Accept": "*/*", "Referer": host + "/"}
    for attempt in (1, 2):
        try:
            r = browser_get(api, headers=headers, timeout=_TIMEOUT)
        except Exception as e:
            return (key, _UNKNOWN, f"{type(e).__name__}")
        if r.status_code in (404, 410):
            return (key, _DEAD, f"HTTP {r.status_code}")
        if r.status_code == 429:
            # Rate limit nie jest dowodem na cokolwiek, więc jedna przerwa i koniec.
            if attempt == 1:
                time.sleep(5)
                continue
            return (key, _UNKNOWN, "HTTP 429")
        if r.status_code != 200 or not r.text:
            return (key, _UNKNOWN, f"HTTP {r.status_code}")
        try:
            data = json.loads(_decrypt(r.text))
        except Exception as e:
            return (key, _UNKNOWN, f"dekod {type(e).__name__}")
        if data.get("error"):
            return (key, _DEAD, str(data["error"])[:80])
        if any(data.get(k) for k in _STREAM_FIELDS):
            return (key, _ALIVE, "")
        return (key, _DEAD, "brak pola source (Video is not ready yet)")
    return (key, _UNKNOWN, "429")


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())