diff --git a/alembic/versions/20260919_0044_playback_eporner_thumb_idx.py b/alembic/versions/20260919_0044_playback_eporner_thumb_idx.py new file mode 100644 index 0000000..9adf14b --- /dev/null +++ b/alembic/versions/20260919_0044_playback_eporner_thumb_idx.py @@ -0,0 +1,37 @@ +"""Indeks do szukania zrodla eporner po adresie miniatury (app/dead_sources.py). + +Proxy miniatur dostaje od CDN-u eporner 404 i zna tylko adres obrazka; zeby sprawdzic i +oznaczyc film, trzeba znalezc jego zrodlo. Miniatura ma inny identyfikator niz strona +filmu, wiec bez indeksu to przeszukanie 4 mln wierszy przy kazdym pustym kafelku. + +Hash i czesciowy (tylko eporner): wyszukujemy wylacznie po rownosci, a eporner to ~350 tys. +wierszy, wiec indeks ma kilkanascie MB. + +CONCURRENTLY w autocommit_block, jak w 0035, 0038, 0039, 0041 i 0043. + +Revision ID: 0044_playback_eporner_thumb_idx +Revises: 0043_scene_tags_tag_sort_idx +Create Date: 2026-09-19 +""" +from __future__ import annotations + +from alembic import op + +revision = "0044_playback_eporner_thumb_idx" +down_revision: str | None = "0043_scene_tags_tag_sort_idx" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + with op.get_context().autocommit_block(): + op.execute( + "CREATE INDEX CONCURRENTLY IF NOT EXISTS ix_playback_sources_eporner_thumb " + "ON playback_sources USING hash (thumbnail_url) " + "WHERE origin = 'tube:epornercom'" + ) + + +def downgrade() -> None: + with op.get_context().autocommit_block(): + op.execute("DROP INDEX CONCURRENTLY IF EXISTS ix_playback_sources_eporner_thumb") diff --git a/app/api/stream_proxy.py b/app/api/stream_proxy.py index 2bef574..ccc83d3 100644 --- a/app/api/stream_proxy.py +++ b/app/api/stream_proxy.py @@ -433,6 +433,12 @@ async def proxy_image( # GOON-5 (Cloudflare 523) i GOON-D, bezsensowny noise w Sentry, lepiej # info log + 502 pass-through bez exception. log.info("img proxy upstream %d for %s", r.status_code, target) + if r.status_code == 404 and ".eporner.com/" in target: + # 404 na miniaturze eporner prawie zawsze znaczy usuniety film; strona + # jest sprawdzana w tle przed oznaczeniem (app/dead_sources.py). + from app import dead_sources + + dead_sources.eporner_thumb_404(target) return Response( content=b"", status_code=502 if r.status_code >= 500 else r.status_code, @@ -548,6 +554,11 @@ async def sxyprn_thumb(post_id: str) -> Response: return Response(content=b"", status_code=502, media_type="image/jpeg") html = r.text if "Post Not Found" in html: + # Post usuniety, czyli strumien tez (ta sama strona). Oznaczamy zrodlo w tle, + # zeby scena zniknela z list, zamiast czekac na klikniecie (2026-09-19). + from app import dead_sources + + dead_sources.sxyprn_post_not_found(pid) return Response(content=b"", status_code=404, media_type="image/jpeg") m = _OG_IMG_RE.search(html) or _OG_IMG_RE2.search(html) or _VID_POSTER_RE.search(html) if not m: diff --git a/app/config.py b/app/config.py index 88f5d46..55f671f 100644 --- a/app/config.py +++ b/app/config.py @@ -122,6 +122,10 @@ class Settings(BaseSettings): sched_studio_title_dedup_hours: int = Field( default=12, validation_alias="GOON_SCHED_STUDIO_TITLE_DEDUP_HOURS" ) + # Sprzatanie martwych zrodel sxyprn i eporner (app/scheduler/dead_reap.py): co N godzin + # porcja zywych zrodel na serwis, polowa z ostatnich 30 dni. 0 = off. + sched_dead_reap_hours: int = Field(default=6, validation_alias="GOON_SCHED_DEAD_REAP_HOURS") + dead_reap_per_origin: int = Field(default=1500, validation_alias="GOON_DEAD_REAP_PER_ORIGIN") # Blacklista zdegenerowanych phashy, wartości dzielonych przez ≥10 scen (zaślepka, # czarna klatka, intro studia). Audyt 2026-07-27: 73 takie wartości przy 4375 # scenach, rekordzistka przy 892 scenach o 892 różnych tytułach. Tube może zacząć diff --git a/app/dead_sources.py b/app/dead_sources.py new file mode 100644 index 0000000..cf84d97 --- /dev/null +++ b/app/dead_sources.py @@ -0,0 +1,116 @@ +"""Oznaczanie martwych zrodel na podstawie sygnalow z proxy miniatur (2026-09-19). + +Pusty kafelek to zwykle martwy film: user widzi scene bez miniatury, klika i dostaje +"the tube removed this video". Dotad zrodlo oznaczal dopiero resolve po kliknieciu, a sceny +bez klikniec wisialy w listach tygodniami. Proxy miniatur i tak pobiera to, co mowi o smierci +filmu, wiec tu zapisujemy wniosek: + +- sxyprn: `/proxy/sxyprn-thumb/` pobiera strone posta; "Post Not Found" to pewny + sygnal (ta sama strona daje miniature i strumien), oznaczamy od razu. +- eporner: miniatura z 404 w 28 na 30 przypadkow nalezala do usunietego filmu, w 2 strona + miala 404 (probka 2026-09-19). Mimo to sprawdzamy strone filmu przed oznaczeniem, bo 404 + na miniaturze bywa tez chwilowe. + +Wszystko w tle, na malej puli watkow: lista 50 scen potrafi dac kilkadziesiat 404 naraz, +a proxy miniatur nie moze na to czekac. Kazdy klucz obslugujemy raz na godzine. +""" +from __future__ import annotations + +import logging +import threading +import time +from concurrent.futures import ThreadPoolExecutor + +from sqlalchemy import text + +from app.db import session_scope + +log = logging.getLogger(__name__) + +_POOL = ThreadPoolExecutor(max_workers=2, thread_name_prefix="dead-mark") +_SEEN: dict[str, float] = {} +_SEEN_LOCK = threading.Lock() +_SEEN_TTL = 3600.0 +_EPORNER_ORIGIN = "tube:epornercom" +_SXYPRN_ORIGIN = "tube:sxyprncom" + + +def _first_time(key: str) -> bool: + now = time.time() + with _SEEN_LOCK: + if _SEEN.get(key, 0.0) > now: + return False + _SEEN[key] = now + _SEEN_TTL + if len(_SEEN) > 50_000: + for k in [k for k, v in _SEEN.items() if v <= now]: + _SEEN.pop(k, None) + return True + + +def _mark(where_sql: str, params: dict, reason: str) -> int: + with session_scope() as s: + n = s.execute( + text( + "UPDATE playback_sources SET dead_at = now(), dead_reason = :reason " + f"WHERE dead_at IS NULL AND {where_sql}" + ), + {**params, "reason": reason[:512]}, + ).rowcount + return n or 0 + + +def sxyprn_post_not_found(post_id: str) -> None: + """Strona posta sxyprn mowi "Post Not Found". Nie blokuje wolajacego.""" + if not _first_time(f"sx:{post_id}"): + return + _POOL.submit(_mark_sxyprn, post_id) + + +def _mark_sxyprn(post_id: str) -> None: + try: + n = _mark( + "origin = :o AND page_url = :p", + {"o": _SXYPRN_ORIGIN, "p": f"https://sxyprn.com/post/{post_id}.html"}, + "sxyprn: Post Not Found (proxy miniatury)", + ) + if n: + log.info("dead-mark sxyprn %s: %d zrodel", post_id, n) + except Exception as e: # pragma: no cover - obronnie, to tylko sprzatanie + log.warning("dead-mark sxyprn %s padl: %s", post_id, e) + + +def eporner_thumb_404(thumb_url: str) -> None: + """Miniatura eporner oddala 404. Sprawdza strone filmu w tle i oznacza, jesli usuniety.""" + if not _first_time(f"ep:{thumb_url}"): + return + _POOL.submit(_verify_eporner, thumb_url) + + +def _verify_eporner(thumb_url: str) -> None: + from app.extractors._fetch import fetch_tube_html + from app.extractors._models import TubePageError + from app.extractors.tubes.eporner import is_deleted_page + + try: + with session_scope() as s: + rows = s.execute( + text( + "SELECT page_url FROM playback_sources " + "WHERE origin = :o AND thumbnail_url = :t AND dead_at IS NULL" + ), + {"o": _EPORNER_ORIGIN, "t": thumb_url}, + ).all() + for (page,) in rows: + reason = None + try: + if is_deleted_page(fetch_tube_html(page, timeout=30)): + reason = "eporner: video has been deleted (proxy miniatury)" + except TubePageError as e: + if e.status_code in (404, 410): + reason = f"eporner: strona {e.status_code} (proxy miniatury)" + if reason: + n = _mark("origin = :o AND page_url = :p", {"o": _EPORNER_ORIGIN, "p": page}, reason) + if n: + log.info("dead-mark eporner %s: %s", page, reason) + except Exception as e: # pragma: no cover - obronnie, to tylko sprzatanie + log.warning("dead-mark eporner %s padl: %s", thumb_url, e) diff --git a/app/extractors/tubes/eporner.py b/app/extractors/tubes/eporner.py index 9308027..8b0ebcd 100644 --- a/app/extractors/tubes/eporner.py +++ b/app/extractors/tubes/eporner.py @@ -19,7 +19,7 @@ import logging import re from app.extractors._fetch import _DEFAULT_UA, browser_get, fetch_tube_html -from app.extractors._models import StreamSource +from app.extractors._models import HosterDead, StreamSource log = logging.getLogger(__name__) @@ -46,8 +46,19 @@ def _hash_to_b36(hex_hash: str) -> str: return "".join(parts) +def is_deleted_page(html: str) -> bool: + """Strona usunietego filmu odpowiada 200 z tytulem, ale bez playera i z tekstem + "Video has been deleted" (np. "File has been removed due to copyright owner request"). + Zywa strona tego tekstu nie ma (sprawdzone 2026-09-19).""" + return "Video has been deleted" in html and not _VID_RE.search(html) + + def extract(page_url: str, *, timeout: float = 60.0) -> list[StreamSource] | None: html = fetch_tube_html(page_url, timeout=timeout) + if is_deleted_page(html): + # Bez tego resolve zwracal "blad chwilowy" (503) i zrodlo nigdy nie bylo oznaczane + # jako martwe: 82 takie wejscia w 48 h, te same martwe sceny wisialy w feedzie. + raise HosterDead(f"eporner {page_url}: video has been deleted") m_vid = _VID_RE.search(html) m_hash = _HASH_RE.search(html) if not m_vid or not m_hash: diff --git a/app/scheduler/dead_reap.py b/app/scheduler/dead_reap.py new file mode 100644 index 0000000..747cef6 --- /dev/null +++ b/app/scheduler/dead_reap.py @@ -0,0 +1,126 @@ +"""Cykliczne sprzatanie martwych zrodel sxyprn i eporner (2026-09-19). + +Oba serwisy kasuja filmy szybko i po cichu: strona dalej odpowiada 200, tylko z +komunikatem ("Post Not Found" na sxyprn, "Video has been deleted" na eporner). Probka +2026-09-19: 57% zywych w bazie zrodel sxyprn w wieku 10-40 dni i 22% eporner bylo juz +usunietych. W apce to puste kafelki i "the tube removed this video" po kliknieciu. + +Dwie drogi, ktore sie uzupelniaja: +- app/dead_sources.py oznacza to, co ktos wlasnie przewija (sygnal z proxy miniatur), +- ten job co kilka godzin sprawdza porcje reszty: polowe z ostatnich 30 dni, bo tam + kasowanie jest najczestsze, polowe losowo ze starszych. Bez kursora i bez stanu: kazdy + przebieg bierze tylko zywe zrodla, wiec oznaczone same wypadaja z puli. + +Oznaczamy tylko przy pewnym sygnale (komunikat o usunieciu albo 404 strony). Siec, timeout, +rate-limit i strona bez playera to "nie wiadomo" i zrodlo zostaje. Wolno i grzecznie +(3 watki, odstep przed kazdym pobraniem), bo sxyprn to nasze najczesciej ogladane zrodlo i +ban IP bylby gorszy niz martwe kafelki. Pelny przebieg po zaleglosciach: +scripts/sxyprn_reap_dead.py. +""" +from __future__ import annotations + +import collections +import logging +import time +from concurrent.futures import ThreadPoolExecutor + +from sqlalchemy import text + +from app.db import session_scope + +log = logging.getLogger(__name__) + +_CONCURRENCY = 3 +_SPACING_SEC = 0.5 +_TIMEOUT = 20.0 +_ALIVE, _DEAD, _UNKNOWN = "alive", "dead", "unknown" + + +def _probe_sxyprn(page_url: str) -> str: + from app.extractors._models import HosterDead + from app.extractors.tubes import sxyprn + + try: + res = sxyprn.extract(page_url, timeout=_TIMEOUT) + except HosterDead: + return _DEAD + except Exception: + return _UNKNOWN + return _ALIVE if res and getattr(res[0], "link", None) else _UNKNOWN + + +def _probe_eporner(page_url: str) -> str: + from app.extractors._fetch import fetch_tube_html + from app.extractors._models import TubePageError + from app.extractors.tubes.eporner import is_deleted_page + + try: + html = fetch_tube_html(page_url, timeout=_TIMEOUT) + except TubePageError as e: + return _DEAD if e.status_code in (404, 410) else _UNKNOWN + except Exception: + return _UNKNOWN + if is_deleted_page(html): + return _DEAD + return _ALIVE if "EP.video.player.vid" in html else _UNKNOWN + + +_PROBES = { + "tube:sxyprncom": ("sxyprn-reap: Post Not Found", _probe_sxyprn), + "tube:epornercom": ("eporner-reap: video has been deleted", _probe_eporner), +} + + +def _pick(origin: str, per_run: int) -> list[tuple[str, str]]: + half = per_run // 2 + with session_scope() as s: + recent = s.execute( + text( + "SELECT id::text, page_url FROM playback_sources " + "WHERE origin = :o AND dead_at IS NULL AND page_url IS NOT NULL " + "AND created_at > now() - interval '30 days' ORDER BY random() LIMIT :n" + ), + {"o": origin, "n": half}, + ).all() + older = s.execute( + text( + "SELECT id::text, page_url FROM playback_sources " + "WHERE origin = :o AND dead_at IS NULL AND page_url IS NOT NULL " + "AND created_at <= now() - interval '30 days' ORDER BY random() LIMIT :n" + ), + {"o": origin, "n": per_run - len(recent)}, + ).all() + return [(r[0], r[1]) for r in recent + older] + + +def run_dead_reap(*, per_origin: int = 2000) -> dict[str, dict[str, int]]: + """Sprawdza po `per_origin` zywych zrodel sxyprn i eporner, oznacza usuniete.""" + out: dict[str, dict[str, int]] = {} + for origin, (reason, probe) in _PROBES.items(): + t0 = time.time() + rows = _pick(origin, per_origin) + cnt: collections.Counter = collections.Counter() + dead_ids: list[str] = [] + + def work(row: tuple[str, str], _probe=probe) -> tuple[str, str]: + time.sleep(_SPACING_SEC) + return row[0], _probe(row[1]) + + with ThreadPoolExecutor(max_workers=_CONCURRENCY) as pool: + for sid, verdict in pool.map(work, rows): + cnt[verdict] += 1 + if verdict == _DEAD: + dead_ids.append(sid) + marked = 0 + for i in range(0, len(dead_ids), 500): + with session_scope() as s: + marked += s.execute( + text( + "UPDATE playback_sources SET dead_at = now(), dead_reason = :r " + "WHERE id = ANY(CAST(:ids AS uuid[])) AND dead_at IS NULL" + ), + {"r": reason, "ids": dead_ids[i : i + 500]}, + ).rowcount + out[origin] = {"checked": len(rows), **cnt, "marked": marked} + log.info("dead-reap %s: %s w %.0f s", origin, out[origin], time.time() - t0) + return out diff --git a/app/scheduler/jobs.py b/app/scheduler/jobs.py index 0040aed..2adf723 100644 --- a/app/scheduler/jobs.py +++ b/app/scheduler/jobs.py @@ -416,6 +416,22 @@ def _job_studio_title_dedup() -> None: log.exception("[scheduler] studio-title dedup failed") +def _job_dead_reap(per_origin: int) -> None: + """Oznacz martwe zrodla sxyprn i eporner (strona mowi, ze film usuniety). Porcja na + przebieg, polowa z ostatnich 30 dni; szczegoly w app/scheduler/dead_reap.py.""" + log.info("[scheduler] dead-reap starting (per_origin=%d)", per_origin) + try: + from app.scheduler.dead_reap import run_dead_reap + + # 2 x 1500 sprawdzen po ~1,2 s na 3 watkach to ~20 min; limit z zapasem. + _run_with_timeout( + lambda: run_dead_reap(per_origin=per_origin), label="dead-reap", timeout_sec=3600 + ) + log.info("[scheduler] dead-reap done") + except Exception: + log.exception("[scheduler] dead-reap failed") + + def _job_ingest_watchdog( max_age_hours: int, search_max_age_hours: int, movie_max_age_hours: int ) -> None: @@ -756,6 +772,19 @@ def build_scheduler(cfg: dict[str, Any]) -> BlockingScheduler: ) log.info("scheduler: studio-title dedup every %dh", cfg["studio_title_dedup_hours"]) + if cfg.get("dead_reap_hours"): + _dr_per = cfg.get("dead_reap_per_origin") or 1500 + sched.add_job( + _job_dead_reap, + IntervalTrigger(hours=cfg["dead_reap_hours"], start_date=INTERVAL_ANCHOR), + args=[_dr_per], + id="dead_reap", + replace_existing=True, + max_instances=1, + coalesce=True, + ) + log.info("scheduler: dead-reap every %dh (per_origin=%d)", cfg["dead_reap_hours"], _dr_per) + if cfg.get("ingest_watchdog_hours"): wd_max_age = cfg.get("ingest_watchdog_max_age_hours") or 48 wd_search_max_age = cfg.get("ingest_watchdog_search_max_age_hours") or 168 diff --git a/app/scheduler/worker.py b/app/scheduler/worker.py index 6876c27..765fcba 100644 --- a/app/scheduler/worker.py +++ b/app/scheduler/worker.py @@ -218,6 +218,9 @@ def run_forever() -> int: "title_dedup_hours": getattr(settings, "sched_title_dedup_hours", 12) or None, # Studio + rdzen tytulu, kopie z tubow o roznym formacie tytulu (report bb1a357d). "studio_title_dedup_hours": getattr(settings, "sched_studio_title_dedup_hours", 12) or None, + # Martwe zrodla sxyprn/eporner (strona mowi "usuniety"), porcja co 6 h. + "dead_reap_hours": getattr(settings, "sched_dead_reap_hours", 6) or None, + "dead_reap_per_origin": getattr(settings, "dead_reap_per_origin", 1500), # Blacklista zdegenerowanych phashy, zaślepki dzielone przez setki scen (audyt # 2026-07-27). Bez tego zaślepka jest zawsze najbliższym trafieniem w Path 3. "phash_blacklist_hours": getattr(settings, "sched_phash_blacklist_hours", 24) or None,