"""Bada, ile martwych zrodel siedzi w kazdym tubie, ZANIM cokolwiek wygasimy.

Po sxyprn (34% martwych na 188 tys. zrodel) nasuwa sie, zeby przemielic tak samo cala
reszte. Metoda sxyprn NIE przenosi sie wprost, z dwoch powodow:

1. **Tylko 6 z 42 ekstraktorow ma twardy sygnal smierci** (`HosterDead`). Dla reszty
   "ekstraktor nic nie zwrocil" znaczy jednoczesnie "post skasowany" i "zmienil sie
   markup" albo "dostalismy rate limit". Wygaszanie na takiej podstawie schowaloby ZYWE
   sceny, dokladnie tak jak przy porntrex (6479 scen zniknelo po cichej zmianie markupu).
2. **Skala.** Zywych zrodel jest ~3,9 mln. W tempie sxyprn (~21 tys./h przy grzecznym
   odstepie) to ponad tydzien ciaglego pukania do naszej wlasnej podazy.

Dlatego najpierw pomiar, potem decyzja co w ogole warto mielic.

**Sygnal, ktory JEST generyczny:** `fetch_tube_html` podnosi `TubePageError` ze statusem
dla kazdego tuba. HTTP 404/410 na stronie sceny to twarda smierc, niezalezna od tego, czy
ekstraktor umie sparsowac strone. Zmiana markupu daje 200, wiec wpada do "nieznane" i nie
rusza niczego. To odwraca ryzyko we wlasciwa strone.

**Soft-404** (HTTP 200 ze strona "nie znaleziono", jak sxyprn) lapiemy przez wolanie
ekstraktora, bo to on podnosi `HosterDead`. Dla tubow BEZ ekstraktora zostaje sam status
HTTP i takie soft-404 pozostaja niewidoczne: raport pokazuje wtedy trafienie we frazy typu
"video not found", ale NIE UZYWA ich do werdyktu, bo taka fraza siedzi tez w zwyklej
nawigacji strony. Sluzy to wylacznie do wskazania tubow wartych recznego obejrzenia.

**Dwie kontrole, bo sam wynik proby niczego nie dowodzi.** "0% martwych" znaczy albo
"tub jest zdrowy", albo "nasza sonda nie widzi trupow". Bez rozroznienia tych dwoch
przypadkow raport jest bezuzyteczny, wiec kazdy tub dostaje obie kontrole:

* **Pozytywna (fals-pozytywy):** zrodla scen, ktore REALNIE ODEGRALY sie z tego tuba w
  ostatnich 14 dniach (`playback_events.status='success'`). Musza wrocic ZYWE. Gdyby tub
  zaczal oddawac 404 wszystkiemu, co nie ma sesji albo wlasciwego IP, przemial skasowalby
  caly katalog, a to wyjdzie wlasnie tutaj.
* **Negatywna (slepota):** prawdziwy adres z podmienionym numerem sceny na absurdalny.
  Taka strona NIE MOZE istniec, wiec musi wrocic MARTWA. Jesli wraca zywa, to tub oddaje
  200 na wszystko i o jego trupach ta metoda nie powie nic. Wynik proby dla takiego tuba
  trzeba zignorowac, a nie czytac jako "czysto".

Wygaszac wolno tylko tuby, ktore zdaly OBIE kontrole.

Uzycie (w kontenerze workera, przez `run -d --rm`, NIE `exec`, patrz nizej):
    python3 -u /srv/scripts/dead_source_survey.py --sample 150 --control 40
    python3 -u /srv/scripts/dead_source_survey.py --origins tube:porntrexcom

UWAGA NA RESTART: `docker compose restart worker` ubija zadania chodzace w kontenerze
workera. Odpalaj przez `docker compose run -d --rm worker`, to osobny kontener.
"""
from __future__ import annotations

import argparse
import collections
import json
import logging
import random
import re
import sys
import time
from concurrent.futures import ThreadPoolExecutor

from sqlalchemy import text

from app.db import session_scope
from app.extractors import is_vps_blocked_fallback, supported_sitetags, try_extract
from app.extractors._fetch import fetch_tube_html
from app.extractors._models import HosterDead, TubePageError

log = logging.getLogger("dead-survey")

_CONCURRENCY = 4
_SPACING_SEC = 0.4
_TIMEOUT = 25.0

# Frazy soft-404 tylko DO RAPORTU, nigdy do wygaszania. Wskazuja tuby warte obejrzenia
# recznie (jak sxyprn, gdzie skasowany post oddaje HTTP 200 ze strona "Post Not Found").
_SOFT404_HINTS = (
    "post not found",
    "video not found",
    "page not found",
    "has been deleted",
    "no longer available",
    "video was deleted",
    "video removed",
    "does not exist",
)


def _probe(item: tuple) -> tuple:
    """(origin, rola, sid, werdykt, szczegol). Sama siec, bez dotykania bazy.

    Idzie przez `try_extract`, a nie przez goly `fetch_tube_html`, i to jest istotne.
    Pierwsza wersja pytala tylko o status HTTP i przez to uznala sxyprn za tuba, na ktorym
    "sonda jest slepa" (0/6 w kontroli negatywnej), chociaz wiemy, ze ma tam 34% trupow.
    Powod: sxyprn oddaje skasowany post jako HTTP 200 ze strona "Post Not Found", a
    `HosterDead` leci dopiero z ekstraktora. Ta sama rzecz dotyczy porntrex, galaxyporn,
    paradisehill, watchporn i xhamster.

    Werdykt "martwe" pada TYLKO na twardych sygnalach: 404/410 albo `HosterDead`. Gdy
    ekstraktor zwroci None (zmiana markupu, rate limit, cokolwiek), jest "nieznane" i
    zrodla nikt nie rusza. Odwrotne ustawienie schowaloby zywe sceny.
    """
    origin, rola, sid, url = item
    sitetag = origin.split(":", 1)[-1]
    time.sleep(_SPACING_SEC)
    try:
        out = try_extract(sitetag, url)
    except TubePageError as e:
        if e.status_code in (404, 410):
            return (origin, rola, sid, "martwe", f"HTTP {e.status_code}")
        # 403, 5xx, rate limit: NIE jest dowodem na nic.
        return (origin, rola, sid, "nieznane", f"HTTP {e.status_code}")
    except HosterDead:
        return (origin, rola, sid, "martwe", "soft-404")
    except Exception as e:
        return (origin, rola, sid, "nieznane", type(e).__name__)
    if out:
        return (origin, rola, sid, "zywe", "")

    # Brak ekstraktora dla tuba: zostaje sam status HTTP.
    if sitetag not in supported_sitetags():
        try:
            html = fetch_tube_html(url, timeout=_TIMEOUT, max_retries=1)
        except TubePageError as e:
            werdykt = "martwe" if e.status_code in (404, 410) else "nieznane"
            return (origin, rola, sid, werdykt, f"HTTP {e.status_code}")
        except Exception as e:
            return (origin, rola, sid, "nieznane", type(e).__name__)
        low = html[:200000].lower()
        hint = next((h for h in _SOFT404_HINTS if h in low), "")
        # HTTP 200 to dla nas ZAWSZE zywe. Fraza idzie wylacznie do raportu.
        return (origin, rola, sid, "zywe", f"soft404?={hint}" if hint else "")

    return (origin, rola, sid, "nieznane", "ekstraktor zwrocil pusto")


_DIGITS_RE = re.compile(r"\d{3,}")


def _bogus(url: str) -> str | None:
    """Prawdziwy adres -> adres sceny, ktora na pewno nie istnieje.

    Podmieniamy OSTATNI dluzszy ciag cyfr (numer sceny) na absurdalnie wysoki, zamiast
    psuc slug. Powod: czesc tubow routuje po samym id i sluga w ogole nie sprawdza, wiec
    zepsuty slug oddalby 200 i falszywie wygladal na slepote sondy.
    """
    m = list(_DIGITS_RE.finditer(url))
    if not m:
        return None
    last = m[-1]
    return url[: last.start()] + "99899989" + url[last.end() :]


def _pick(origins: list[str], n_sample: int, n_control: int) -> list[tuple]:
    work: list[tuple] = []
    with session_scope() as s:
        for origin in origins:
            rows = s.execute(
                text(
                    """
                    SELECT id, 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": n_sample},
            ).fetchall()
            work += [(origin, "proba", sid, u) for sid, u in rows]

            # Kontrola: zrodla scen, ktore naprawde odegraly sie z tego tuba.
            ctrl = s.execute(
                text(
                    """
                    SELECT ps.id, 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() - interval '14 days'
                     GROUP BY ps.id, ps.page_url
                     ORDER BY random() LIMIT :n
                    """
                ),
                {"o": origin, "n": n_control},
            ).fetchall()
            work += [(origin, "kontrola", sid, u) for sid, u in ctrl]

            # Kontrola negatywna: adresy, ktore NIE MOGA istniec. Bierzemy je z proby,
            # zeby miec dokladnie ten sam ksztalt URL co realne strony tego tuba.
            for sid, u in rows[:8]:
                bad = _bogus(u)
                if bad:
                    work.append((origin, "negatyw", sid, bad))
    return work


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser()
    p.add_argument("--sample", type=int, default=150)
    p.add_argument("--control", type=int, default=60)
    p.add_argument("--origins", default="", help="lista po przecinku (domyslnie wszystkie)")
    p.add_argument("--min-sources", type=int, default=500, help="pomin tuby mniejsze niz to")
    p.add_argument("--out", default="/tmp/_survey_out.json")
    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)

    if a.origins:
        origins = [o.strip() for o in a.origins.split(",") if o.strip()]
    else:
        with session_scope() as s:
            origins = [
                r[0]
                for r in s.execute(
                    text(
                        """
                        SELECT origin FROM playback_sources WHERE dead_at IS NULL
                         GROUP BY origin HAVING count(*) >= :m ORDER BY count(*) DESC
                        """
                    ),
                    {"m": a.min_sources},
                ).fetchall()
            ]
    # Tuby resolvowane wylacznie w WebView na telefonie sa z VPS nieweryfikowalne
    # (CDN IP-bound / CAPTCHA), wiec kazda proba dalaby smieci zamiast pomiaru.
    slepe = [o for o in origins if is_vps_blocked_fallback(o.split(":", 1)[-1])]
    if slepe:
        log.info("pomijam (resolve tylko na telefonie): %s", ", ".join(slepe))
        origins = [o for o in origins if o not in slepe]
    log.info("tubow do zbadania: %d", len(origins))

    work = _pick(origins, a.sample, a.control)
    # Przeplatamy tuby, zeby zaden host nie dostal calej swojej porcji ciagiem.
    random.shuffle(work)
    log.info("prob do wykonania: %d", len(work))

    wynik: dict = collections.defaultdict(lambda: collections.defaultdict(collections.Counter))
    detale: dict = collections.defaultdict(collections.Counter)
    t0 = time.time()

    with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool:
        for i, (origin, rola, _sid, werdykt, szczegol) in enumerate(pool.map(_probe, work), 1):
            wynik[origin][rola][werdykt] += 1
            if szczegol:
                detale[origin][f"{rola}:{szczegol}"[:60]] += 1
            if i % 250 == 0:
                log.info("[%d/%d] %.0f min", i, len(work), (time.time() - t0) / 60)
                for h in logging.getLogger().handlers:
                    h.flush()

    log.info("=" * 100)
    log.info(
        "%-26s %6s %6s %6s %8s | %-14s %-12s %s",
        "tub", "zywe", "martwe", "nzn", "martwe%", "POZYTYWNA", "NEGATYWNA", "WERDYKT",
    )
    raport = {}
    for origin in sorted(wynik, key=lambda o: -wynik[o]["proba"]["martwe"]):
        pr = wynik[origin]["proba"]
        kt = wynik[origin]["kontrola"]
        ng = wynik[origin]["negatyw"]
        n, k, g = sum(pr.values()), sum(kt.values()), sum(ng.values())
        proc = 100 * pr["martwe"] / max(1, n)

        # Pozytywna: scena, ktora sie odegrala, a teraz wraca martwa, to fals-pozytyw.
        # Nie zadamy zera, bo w 14 dni czesc filmow znika naprawde. Zadamy, zeby ten
        # szum byl niski, inaczej przemial kasowalby zywy katalog.
        poz_proc = 100 * kt["martwe"] / max(1, k)
        poz_ok = k >= 10 and poz_proc <= 5
        poz = f"{kt['martwe']}/{k}" + ("" if poz_ok else (" !!" if k >= 10 else " za malo"))

        # Negatywna: adres, ktory nie moze istniec, MUSI wrocic martwy. Inaczej nie
        # odrozniamy "tub zdrowy" od "nic nie widzimy". Prog 80%, bo pojedynczy
        # wylosowany numer moze przypadkiem trafic w istniejaca scene.
        neg_ok = g >= 4 and ng["martwe"] >= 0.8 * g
        neg = f"{ng['martwe']}/{g}" + ("" if neg_ok else " SLEPA")

        # Odsetek z proby minus szum z kontroli, czyli ile trupow naprawde spodziewamy sie
        # znalezc. Bez tej korekty raportowalibysmy wlasne fals-pozytywy jako martwe zrodla.
        netto = max(0.0, proc - poz_proc)

        if not neg_ok:
            werdykt = "POMIN, sonda slepa"
        elif not poz_ok:
            werdykt = "POMIN, fals-pozytywy"
        elif netto < 2:
            werdykt = "zdrowy, nie ma co mielic"
        else:
            werdykt = f"MIELIC (~{netto:.0f}%)"

        log.info(
            "%-26s %6d %6d %6d %7.0f%% | %-14s %-12s %s",
            origin, pr["zywe"], pr["martwe"], pr["nieznane"], proc, poz, neg, werdykt,
        )
        raport[origin] = {
            "proba": dict(pr),
            "kontrola": dict(kt),
            "negatyw": dict(ng),
            "martwe_proc": round(proc, 1),
            "martwe_netto_proc": round(netto, 1),
            "kontrola_ok": poz_ok,
            "negatyw_ok": neg_ok,
            "mielic": neg_ok and poz_ok and pr["martwe"] > 0,
            "detale": dict(detale[origin].most_common(5)),
        }

    with open(a.out, "w", encoding="utf-8") as fh:
        json.dump(raport, fh, ensure_ascii=False, indent=1)
    log.info("zapisano %s | %.0f min", a.out, (time.time() - t0) / 60)
    return 0


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