"""Oznacza martwe źródła sxyprn (skasowane posty), których apka i tak nie odtworzy.

Obserwacja Jana (2026-08-26): scena sxyprn z widoczną miniaturą gra, a bez miniatury
w ok. 90% nie. Sprawdzone i potwierdzone mechanicznie: miniatura sxyprn NIE jest brana
z zapisanego `thumbnail_url` (te są podpisane czasowo, z TTL < 1h, i dawno wygasły),
tylko z `/proxy/sxyprn-thumb/<id>`, który NA ŻYWO pobiera stronę posta i wyłuskuje
poster. Odtwarzanie bierze `data-vnfo` z DOKŁADNIE tej samej strony. Skasowany post =
brak postera = pusty kafelek ORAZ brak odtwarzania. Stąd korelacja, którą widać w apce.

Sygnał śmierci jest twardy i jednoznaczny: sxyprn na usunięty post zwraca HTTP 200 ze
stroną „Post Not Found" (soft-404). Ekstraktor rzuca wtedy `HosterDead`. Tylko to
oznaczamy martwym. Strona bez `data-vnfo` (inny, niepewny przypadek) NIE jest ruszana,
bo to bywa przejściowe.

Oznaczamy `dead_at` (odwracalne), a NIE kasujemy wiersza: gdyby sxyprn kiedyś wskrzesił
post albo gdyby to był chwilowy błąd, wystarczy wyzerować znacznik. Efekt w UI jest ten
sam co skasowanie: źródło znika z listy, a scena tylko-sxyprn bez żywego źródła chowa
się za filtrem `has_playback`.

UWAGA na obciążenie: sxyprn to nasze NAJLEPSZE źródło (top origin, ~12 tys. odtworzeń
miesięcznie). Dlatego niska równoległość i odstęp między pobraniami, żeby nie oberwać
rate-limitem ani banem IP. To celowo wolne. Wznawialne za darmo: bierzemy tylko żywe
źródła, więc ponowny przebieg pomija już oznaczone.

UWAGA NA RESTART: `docker compose restart worker` UBIJA to zadanie, jesli chodzi w
kontenerze workera (2026-08-27: stracilem tak przebieg po 151 tys. sprawdzonych, log
urwal sie 4 minuty przed restartem). Odpalaj przez `docker compose run -d --rm worker`,
bo to OSOBNY kontener, ktorego restart workera nie dotyka. Wznawianie jest darmowe:
bierzemy tylko zywe zrodla, wiec ponowne odpalenie samo pomija juz oznaczone.

Użycie:
    python3 /srv/scripts/sxyprn_reap_dead.py --limit 300 --dry-run
    python3 /srv/scripts/sxyprn_reap_dead.py --limit 5000
"""
from __future__ import annotations

import argparse
import collections
import logging
import sys
import time
from concurrent.futures import ThreadPoolExecutor

from sqlalchemy import text

from app.db import session_scope
from app.extractors._models import HosterDead
from app.extractors.tubes import sxyprn

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

_ORIGIN = "tube:sxyprncom"
# Wolno i grzecznie: sxyprn to top źródło, nie chcemy bana. Trzy wątki, odstęp przed
# każdym pobraniem. Efektywnie kilka żądań na sekundę.
_CONCURRENCY = 3
_SPACING_SEC = 0.5
_TIMEOUT = 20.0

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


def _probe(page_url: str) -> str:
    """alive / dead / unknown. Tylko „Post Not Found" (HosterDead) to pewna śmierć."""
    time.sleep(_SPACING_SEC)
    try:
        res = sxyprn.extract(page_url, timeout=_TIMEOUT)
    except HosterDead:
        return _DEAD
    except Exception:
        # Sieć, timeout, rate-limit: NIEPEWNE, zostawiamy źródło nietknięte.
        return _UNKNOWN
    if res and getattr(res[0], "link", None):
        return _ALIVE
    # Brak data-vnfo: niepewne (mogła się zmienić struktura strony), nie ruszamy.
    return _UNKNOWN


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser()
    p.add_argument("--limit", type=int, default=0, help="ile źródeł sprawdzić (0 = wszystkie żywe)")
    p.add_argument("--after-id", default=None,
                   help="pomin zrodla o id <= podanego (wznowienie przerwanego przebiegu)")
    p.add_argument("--dry-run", action="store_true", help="nie oznaczaj, 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)

    # Wznowienie bez mielenia od nowa: przebieg idzie ZAWSZE `ORDER BY id`, więc wszystko
    # poniżej podanego id zostało już sprawdzone. Bez tego wznowienie po przerwaniu
    # sprawdzało ponownie 100 tys. źródeł POTWIERDZONYCH jako żywe, czyli 73% pracy i
    # tyle samo zbędnych żądań do sxyprn (2026-08-27). Granicę bierzesz z bazy:
    #   SELECT max(id::text)::uuid FROM playback_sources
    #    WHERE dead_reason = 'sxyprn-reap: Post Not Found';
    _po_id = f"AND id > '{a.after_id}'" if a.after_id else ""
    sql = f"""
        SELECT id, page_url FROM playback_sources
         WHERE origin = '{_ORIGIN}' AND dead_at IS NULL AND page_url IS NOT NULL
           {_po_id}
         ORDER BY id
    """
    if a.limit:
        sql += f" LIMIT {int(a.limit)}"
    with session_scope() as s:
        rows = s.execute(text(sql)).fetchall()
    log.info("do sprawdzenia: %d źródeł sxyprn", len(rows))

    cnt: collections.Counter = collections.Counter()
    ubite = 0
    t0 = time.time()

    def work(row):
        sid, page_url = row
        return sid, _probe(page_url)

    # Porcjami, ze SWIEZA pula na kazda porcje. `pool.map` na calej liscie kolejkowal
    # 188 tys. zadan naraz, wiec postep byl widoczny dopiero na koncu, a przerwanie w
    # polowie nie zostawialo zadnego punktu zaczepienia.
    _PORCJA = 500
    for start in range(0, len(rows), _PORCJA):
        porcja = rows[start : start + _PORCJA]
        do_ubicia: list = []
        with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool:
            for sid, verdict in pool.map(work, porcja):
                cnt[verdict] += 1
                if verdict == _DEAD:
                    do_ubicia.append(sid)
        if do_ubicia and not a.dry_run:
            # Jeden UPDATE na porcje zamiast jednego na wiersz.
            with session_scope() as s:
                ubite += s.execute(
                    text(
                        "UPDATE playback_sources SET dead_at = now(), "
                        "dead_reason = 'sxyprn-reap: Post Not Found' "
                        "WHERE id = ANY(:ids) AND dead_at IS NULL"
                    ),
                    {"ids": do_ubicia},
                ).rowcount
        log.info(
            "[%d/%d] żywe=%d martwe=%d nieznane=%d, ubite=%d (%.0f min)",
            min(start + _PORCJA, len(rows)), len(rows),
            cnt[_ALIVE], cnt[_DEAD], cnt[_UNKNOWN], ubite, (time.time() - t0) / 60,
        )
        for h in logging.getLogger().handlers:
            h.flush()   # bez tego log do pliku milczal godzinami (buforowanie)

    total = sum(cnt.values())
    log.info(
        "KONIEC: żywe=%d (%.0f%%) martwe=%d (%.0f%%) nieznane=%d (%.0f%%) | ubite=%d | %.0f min",
        cnt[_ALIVE], 100 * cnt[_ALIVE] / max(1, total),
        cnt[_DEAD], 100 * cnt[_DEAD] / max(1, total),
        cnt[_UNKNOWN], 100 * cnt[_UNKNOWN] / max(1, total),
        ubite, (time.time() - t0) / 60,
    )
    return 0


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