goon/app/scheduler/dead_reap.py

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