feat(reap): generyczny przemial martwych zrodel z bramka, ktora tub musi zdac

Uogolnienie `sxyprn_reap_dead.py` na pozostale tuby. Werdykt bierze sie z `try_extract`,
czyli z kodu, ktorego uzywa aplikacja przy odtwarzaniu, wiec sonda nie widzi czegos
innego niz user. Martwe pada TYLKO na 404/410 albo `HosterDead`; pusty wynik ekstraktora
to "nieznane" i zrodla nikt nie rusza.

Bramka odpala sie ZA KAZDYM URUCHOMIENIEM, zamiast polegac na tym, ze ktos pamieta wynik
wczesniejszego pomiaru. Sprawdza jednoczesnie, czy sonda widzi smierc (adresy scen, ktore
nie moga istniec, musza wrocic martwe) i czy nie zabija zywych (sceny odegrane w ostatnich
7 dniach musza wrocic zywe). Nie zdana bramka konczy skrypt bez tkniecia bazy.

Wynik pierwszego pomiaru: przemial ma sens dla hqporner (~29%) i galaxyporn (~14%).
porntrex, xvideos, neporn, supjav i reszta soft-404 odpadaja, bo sonda ich smierci nie
widzi. youperv odpada, bo oddaje martwe na 25% scen, ktore realnie sie odegraly.

Trzy rzeczy, ktore wyszly dopiero przez kontrole, nie przez czytanie kodu:

* Pierwsza wersja sondy pytala tylko o status HTTP i uznala SXYPRN za tuba, na ktorym nic
  nie widac (0/6), chociaz wlasnie wygasilismy tam 34% zrodel. Sxyprn oddaje skasowany
  post jako HTTP 200, a `HosterDead` leci dopiero z ekstraktora. Po przejsciu na
  `try_extract` negatywna daje 7/7.
* Kontrola pozytywna przy 12 probach byla za ciasna: jeden trup to 8% i przewracal prog
  5%, chociaz realny szum galaxyporn to ~3%. Stad 40 prob i okno 7 dni zamiast 14, bo im
  swiezsze odtworzenie, tym mocniej "martwe" wskazuje na nasz blad, a nie na kasowanie.
* Budowanie adresow nieistniejacych przez podmiane cyfr nie dzialalo dla tubow BEZ numeru
  sceny w adresie (hentaigasm, mypornerleak, freshporno). Wychodzily jako "0/0", czyli
  niezmierzone, a to okolo 950 tys. zrodel. Teraz kolejno: numer w query, segment-
  identyfikator, a dopiero na koncu slug. Kolejnosc jest istotna, bo psucie segmentu
  trasy (`videos`) sprawdzaloby, czy tub obsluguje adres, a nie czy istnieje scena, wiec
  tub oddajacy 404 na zla trase i 200 na nieistniejaca scene falszywie zdalby bramke.

Sonda siedzi w `_reap_probe.py` wspolnym dla pomiaru i przemialu, zeby bramka testowala
dokladnie ten kod, ktory potem mieli baze.
This commit is contained in:
goon-foss 2026-08-27 15:30:54 +02:00
parent bbdff6bc90
commit f10d999410
2 changed files with 279 additions and 60 deletions

View file

@ -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

256
scripts/tube_reap_dead.py Normal file
View file

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