diff --git a/scripts/dead_source_survey.py b/scripts/dead_source_survey.py index dee2a23..96c8415 100644 --- a/scripts/dead_source_survey.py +++ b/scripts/dead_source_survey.py @@ -52,17 +52,18 @@ import collections import json import logging import random -import re 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, try_extract +from app.extractors import is_vps_blocked_fallback, supported_sitetags from app.extractors._fetch import fetch_tube_html -from app.extractors._models import HosterDead, TubePageError +from app.extractors._models import TubePageError log = logging.getLogger("dead-survey") @@ -85,68 +86,30 @@ _SOFT404_HINTS = ( def _probe(item: tuple) -> tuple: - """(origin, rola, sid, werdykt, szczegol). Sama siec, bez dotykania bazy. + """(origin, rola, sid, werdykt, szczegol). Cienka nakladka na wspolna sonde. - 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. + 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] - 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", "") + 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. - 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() :] + 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]: @@ -187,7 +150,7 @@ def _pick(origins: list[str], n_sample: int, n_control: int) -> list[tuple]: # 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) + bad = bogus_url(u) if bad: work.append((origin, "negatyw", sid, bad)) return work diff --git a/scripts/tube_reap_dead.py b/scripts/tube_reap_dead.py new file mode 100644 index 0000000..2e6935e --- /dev/null +++ b/scripts/tube_reap_dead.py @@ -0,0 +1,256 @@ +"""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 # 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 = '' 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())