"""Wygasza martwe zrodla dowolnego tuba, ale tylko jesli sam sobie udowodni, ze widzi smierc.

Uogolnienie `sxyprn_reap_dead.py` na pozostale tuby. Werdykt bierze sie z `try_extract`,
czyli z tego samego kodu, ktorego uzywa aplikacja przy odtwarzaniu, wiec skrypt nie moze
"widziec" czegos innego niz user.

**Martwe pada TYLKO na twardych sygnalach:** HTTP 404/410 (`TubePageError`) albo
`HosterDead` (soft-404, np. sxyprn "Post Not Found"). Pusty wynik ekstraktora to
"nieznane" i zrodla nikt nie rusza. Odwrotne ustawienie schowaloby zywe sceny po cichej
zmianie markupu, tak jak przy porntrex (6479 scen).

**Bramka wejsciowa, odpalana ZA KAZDYM RAZEM przed przemialem.** Nie polegamy na tym, ze
ktos pamieta wynik wczesniejszego pomiaru, bo tub moze zmienic zachowanie miedzy jednym
uruchomieniem a drugim. Przed wygaszeniem czegokolwiek skrypt sam sprawdza dwie rzeczy:

* **Kontrola negatywna:** adresy z numerem sceny podmienionym na absurdalny. Nie moga
  istniec, wiec musza wrocic martwe. Jesli wracaja zywe, sonda jest dla tego tuba slepa
  i przemial znalazlby zero trupow, mielac przy tym cala baze.
* **Kontrola pozytywna:** zrodla scen, ktore realnie odegraly sie w ostatnich 7 dniach.
  Musza wrocic zywe. Jesli wracaja martwe, sonda produkuje fals-pozytywy i przemial
  skasowalby ZYWY katalog. To jest ten blad, ktory boli, wiec prog jest ostry.

Nie zdana bramka konczy skrypt bez tkniecia bazy. `--skip-gate` istnieje wylacznie do
debugowania i wypisuje ostrzezenie.

Oznaczamy `dead_at` (odwracalne), a NIE kasujemy wiersza. Efekt w UI jest ten sam co
skasowanie (zrodlo znika z listy, scena bez zywego zrodla chowa sie za `has_playback`),
ale gdyby tub wskrzesil posty albo gdybysmy sie pomylili, wystarczy wyzerowac znacznik.

UWAGA NA RESTART: `docker compose restart worker` UBIJA zadania chodzace w kontenerze
workera, a restart workera to zwykly krok wdrozenia backendu. Odpalaj przez
`docker compose run -d --rm worker`, to osobny kontener.

Uzycie:
    python3 -u /srv/scripts/tube_reap_dead.py --origin tube:hqpornercom --limit 300 --dry-run
    python3 -u /srv/scripts/tube_reap_dead.py --origin tube:hqpornercom
    python3 -u /srv/scripts/tube_reap_dead.py --origin tube:X --after-id <uuid>   # wznowienie
"""
from __future__ import annotations

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

from sqlalchemy import text

from _reap_probe import ALIVE as _ALIVE_V
from _reap_probe import DEAD as _DEAD_V
from _reap_probe import UNKNOWN as _UNKNOWN_V
from _reap_probe import bogus_url, probe

from app.db import session_scope
from app.extractors import is_vps_blocked_fallback

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

# Grzecznie: to nasza wlasna podaz, nie chcemy bana ani rate-limitu na realnym resolve.
_CONCURRENCY = 3
_SPACING_SEC = 0.4
_TIMEOUT = 25.0
_PORCJA = 500

_ALIVE, _DEAD, _UNKNOWN = _ALIVE_V, _DEAD_V, _UNKNOWN_V

# Prog bramki. Negatywna: tyle prob musi wrocic martwych, zeby uznac, ze sonda widzi
# smierc. Pozytywna: tyle najwyzej moze wrocic martwych, bo czesc scen znika naprawde.
_NEG_MIN = 0.8
_POZ_MAX = 0.05
# Kontrola negatywna rozstrzyga sie od razu (adres nie istnieje, wiec albo widzimy 404,
# albo nie), wiec starcza jej malo prob. Pozytywna potrzebuje ich duzo wiecej, bo jeden
# trup na 12 to juz 8% i przewraca prog 5%, chociaz realny szum bywa ~3%. Przy 12 probach
# galaxyporn odpadal na pojedynczej scenie skasowanej naprawde.
_GATE_NEG_N = 12
_GATE_POZ_N = 40
# Okno dla kontroli pozytywnej. Krotkie celowo: im swiezsze odtworzenie, tym mniejsza
# szansa, ze scena zdazyla umrzec naprawde, wiec "martwe" mocniej wskazuje na nasz blad,
# a nie na kasowanie u zrodla.
_POZ_DNI = 7

def _gate(origin: str, sitetag: str) -> bool:
    """Czy sonda dla tego tuba jednoczesnie widzi smierc i nie zabija zywych."""
    with session_scope() as s:
        prawdziwe = [
            u
            for (u,) in s.execute(
                text(
                    """
                    SELECT page_url FROM playback_sources
                     WHERE origin = :o AND dead_at IS NULL AND page_url IS NOT NULL
                     ORDER BY random() LIMIT :n
                    """
                ),
                {"o": origin, "n": _GATE_NEG_N * 3},
            ).fetchall()
        ]
        odegrane = [
            u
            for (u,) in s.execute(
                text(
                    """
                    SELECT ps.page_url
                      FROM playback_sources ps
                      JOIN playback_events pe
                        ON pe.scene_id = ps.scene_id AND pe.origin = ps.origin
                     WHERE ps.origin = :o AND ps.dead_at IS NULL AND ps.page_url IS NOT NULL
                       AND pe.status = 'success'
                       AND pe.created_at > now() - make_interval(days => :dni)
                     GROUP BY ps.page_url
                     ORDER BY random() LIMIT :n
                    """
                ),
                {"o": origin, "n": _GATE_POZ_N, "dni": _POZ_DNI},
            ).fetchall()
        ]

    negatywy = [b for b in (bogus_url(u) for u in prawdziwe) if b][:_GATE_NEG_N]
    if len(negatywy) < 4:
        log.error(
            "BRAMKA: nie umiem zbudowac adresow nieistniejacych dla %s "
            "(adresy bez numeru sceny) -> nie da sie sprawdzic, czy sonda widzi smierc",
            origin,
        )
        return False
    if len(odegrane) < 20:
        log.error(
            "BRAMKA: %s nie ma dosc udanych odtworzen z ostatnich 7 dni (%d) "
            "-> nie da sie sprawdzic, czy sonda nie zabija zywych",
            origin,
            len(odegrane),
        )
        return False

    with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool:
        neg = [v for v, _ in pool.map(lambda u: probe(sitetag, u), negatywy)]
        poz = [v for v, _ in pool.map(lambda u: probe(sitetag, u), odegrane)]

    n_martwe = neg.count(_DEAD)
    p_martwe = poz.count(_DEAD)
    log.info(
        "BRAMKA %s: negatywna %d/%d martwych (min %.0f%%), pozytywna %d/%d martwych (max %.0f%%)",
        origin, n_martwe, len(neg), 100 * _NEG_MIN, p_martwe, len(poz), 100 * _POZ_MAX,
    )
    if n_martwe < _NEG_MIN * len(neg):
        log.error("BRAMKA NIE ZDANA: sonda nie widzi smierci na %s, przemial nic by nie dal", origin)
        return False
    if p_martwe > _POZ_MAX * len(poz):
        log.error(
            "BRAMKA NIE ZDANA: %s oddaje martwe na sceny, ktore sie odegraly "
            "-> przemial skasowalby zywy katalog",
            origin,
        )
        return False
    log.info("BRAMKA ZDANA dla %s", origin)
    return True


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser()
    p.add_argument("--origin", required=True, help="np. tube:hqpornercom")
    p.add_argument("--limit", type=int, default=0, help="ile zrodel sprawdzic (0 = wszystkie)")
    p.add_argument("--after-id", default=None, help="pomin zrodla o id <= podanego (wznowienie)")
    p.add_argument("--dry-run", action="store_true", help="nie oznaczaj, tylko policz")
    p.add_argument("--skip-gate", action="store_true", help="TYLKO do debugowania")
    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", "curl_cffi"):
        logging.getLogger(noisy).setLevel(logging.WARNING)

    origin = a.origin
    sitetag = origin.split(":", 1)[-1]
    if is_vps_blocked_fallback(sitetag):
        log.error("%s resolwuje sie tylko w WebView na telefonie, z VPS nic nie sprawdzimy", origin)
        return 2

    if a.skip_gate:
        log.warning("BRAMKA POMINIETA (--skip-gate). Wyniki moga byc smieciowe.")
    elif not _gate(origin, sitetag):
        return 3

    # Wznowienie: przebieg idzie ZAWSZE `ORDER BY id`, wiec wszystko ponizej podanego id
    # zostalo sprawdzone. Bez tego wznowienie mieli ponownie zrodla POTWIERDZONE jako zywe
    # (martwe wypadaja same, bo bierzemy tylko `dead_at IS NULL`). Granice bierzesz z bazy:
    #   SELECT max(id::text)::uuid FROM playback_sources
    #    WHERE origin = '<origin>' AND dead_reason LIKE 'tube-reap%';
    _po_id = "AND id > :aid" if a.after_id else ""
    sql = f"""
        SELECT id, page_url FROM playback_sources
         WHERE origin = :o 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)}"
    params: dict = {"o": origin}
    if a.after_id:
        params["aid"] = a.after_id
    with session_scope() as s:
        rows = s.execute(text(sql), params).fetchall()
    log.info("do sprawdzenia: %d zrodel %s", len(rows), origin)

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

    # Porcjami, ze swieza pula na porcje: `pool.map` na calej liscie kolejkuje wszystko
    # naraz, wiec postep widac dopiero na koncu, a przerwanie nie zostawia punktu
    # zaczepienia. Zapis jednym UPDATE na porcje.
    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, _det) in zip(
                [r[0] for r in porcja],
                pool.map(lambda r: probe(sitetag, r[1]), porcja),
            ):
                cnt[verdict] += 1
                if verdict == _DEAD:
                    do_ubicia.append(sid)
        if do_ubicia and not a.dry_run:
            with session_scope() as s:
                ubite += s.execute(
                    text(
                        "UPDATE playback_sources SET dead_at = now(), dead_reason = :r "
                        " WHERE id = ANY(:ids) AND dead_at IS NULL"
                    ),
                    {"r": f"tube-reap: {sitetag} 404/soft-404"[:512], "ids": do_ubicia},
                ).rowcount
        log.info(
            "[%d/%d] zywe=%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 milczy godzinami (buforowanie) i martwy przebieg
            # wyglada identycznie jak pracujacy

    total = sum(cnt.values())
    log.info(
        "KONIEC %s: zywe=%d (%.0f%%) martwe=%d (%.0f%%) nieznane=%d (%.0f%%) | ubite=%d | %.0f min",
        origin,
        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())