goon/scripts/dead_source_survey.py

"""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 sys
import time
from concurrent.futures import ThreadPoolExecutor

from sqlalchemy import text

from _reap_probe import bogus_url, probe

from app.db import session_scope
from app.extractors import is_vps_blocked_fallback, supported_sitetags
from app.extractors._fetch import fetch_tube_html
from app.extractors._models import 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). Cienka nakladka na wspolna sonde.

    Werdykt liczy `_reap_probe.probe`, czyli DOKLADNIE ten sam kod, ktorego uzywa bramka
    przemialu. Tutaj dokladamy tylko dwie rzeczy: obsluge tubow bez zarejestrowanego
    ekstraktora (zostaje sam status HTTP) i wykrywanie fraz soft-404 do raportu.
    """
    origin, rola, sid, url = item
    sitetag = origin.split(":", 1)[-1]
    werdykt, szczegol = probe(sitetag, url)
    if werdykt != "unknown" or sitetag in supported_sitetags():
        return (origin, rola, sid, {"alive": "zywe", "dead": "martwe"}.get(werdykt, "nieznane"), szczegol)

    # Brak ekstraktora dla tuba: zostaje sam status HTTP.
    try:
        html = fetch_tube_html(url, timeout=_TIMEOUT, max_retries=1)
    except TubePageError as e:
        w = "martwe" if e.status_code in (404, 410) else "nieznane"
        return (origin, rola, sid, w, 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 "")


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