feat(playback): zgloszenie 404/410 z telefonu sprawdzane z serwera

Zrodlo znikalo tylko po dlugim przytrzymaniu przez uzytkownika. Teraz
zgloszenie "Response code: 404/410" uruchamia w tle ponowny resolve z serwera;
martwe dopiero, gdy ekstraktor mowi "usuniete" albo plik z serwera tez daje
404/410 (dwie kontrole). Na sucho na 84 parach z 14 dni: 0 martwych, 83 zywe,
czyli 410 z telefonu prawie zawsze znaczy zly adres, nie skasowany plik.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

app/api/playback_events.py

@@ -17,6 +17,7 @@ from app.api.device import get_device_id
 from app.auth import require_api_key
 from app.db import get_session
 from app.models.playback_event import PlaybackEvent
+from app.playback_gone import is_gone_report, verify_gone_async
 
 router = APIRouter(prefix="/playback-events", tags=["telemetry"], dependencies=[Depends(require_api_key)])
 
@@ -51,3 +52,7 @@ def post_playback_event(
         )
     )
     session.commit()
+    # Plik zniknal (404/410 w odtwarzaczu): serwer sprawdza zrodlo sam i oznacza martwe
+    # dopiero po wlasnym potwierdzeniu (app/playback_gone.py).
+    if is_gone_report(body.status, body.error_kind, body.error_detail):
+        verify_gone_async(body.scene_id, body.origin)

app/playback_gone.py

@@ -0,0 +1,136 @@
+"""Zgloszenie „plik zniknal" z telefonu -> sprawdzenie z serwera -> zrodlo martwe.
+
+Skad: telemetria 7 dni (2026-10-02) miala ~230 bledow „Response code: 410" z fullmovies.xxx
+i hdporn.gg, a zrodlo znikalo tylko po dlugim przytrzymaniu przez uzytkownika. Kolejne osoby
+trafialy w ten sam skasowany plik.
+
+Dwie kontrole, jak przy kazdym przemiale zrodel (patrz notatka o reaperach): telefon dostal
+404/410 na pliku, a serwer rozwiazuje strone tube na nowo i sam dostaje 404/410 na pliku albo
+ekstraktor mowi wprost, ze film usuniety (HosterDead, TubePageError 404/410). 503, bledy sieci,
+pusty ekstraktor i hostery rozwiazywane na telefonie niczego nie oznaczaja.
+"""
+from __future__ import annotations
+
+import logging
+import re
+import threading
+import time
+import uuid
+from datetime import UTC, datetime
+
+import httpx
+from sqlalchemy import select
+
+from app.db import session_scope
+from app.models.playback_source import PlaybackSource
+
+log = logging.getLogger(__name__)
+
+#: Komunikat ExoPlayera o pliku, ktorego nie ma (expo-video: "... Source error Response code: 410").
+_GONE_RE = re.compile(r"response code: (404|410)\b", re.IGNORECASE)
+_UA = "Mozilla/5.0 (Linux; Android 14) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0.0.0 Mobile Safari/537.36"
+#: Ta sama para (scena, origin) sprawdzana najwyzej raz na tyle sekund (wiele telefonow naraz).
+_RECHECK_SEC = 3600
+_checked: dict[tuple, float] = {}
+_lock = threading.Lock()
+
+
+def is_gone_report(status: str, error_kind: str | None, error_detail: str | None) -> bool:
+    return status == "error" and bool(_GONE_RE.search(error_detail or ""))
+
+
+def _file_status(url: str, referer: str | None) -> int | None:
+    headers = {"User-Agent": _UA, "Range": "bytes=0-1023"}
+    if referer:
+        headers["Referer"] = referer
+    try:
+        with httpx.Client(timeout=20, follow_redirects=True) as http:
+            return http.get(url, headers=headers).status_code
+    except httpx.HTTPError:
+        return None
+
+
+def _mark_dead(pb_id: uuid.UUID, reason: str) -> None:
+    with session_scope() as s:
+        pb = s.get(PlaybackSource, pb_id)
+        if pb is not None and pb.dead_at is None:
+            pb.dead_at = datetime.now(UTC)
+            pb.dead_reason = reason[:512]
+            s.commit()
+            log.info("gone: playback %s (%s) martwy: %s", pb_id, pb.origin, reason)
+
+
+def verify_gone(scene_id: uuid.UUID, origin: str, *, dry_run: bool = False) -> dict[str, int]:
+    """Sprawdza zywe zrodla (scena, origin) zgloszone jako 404/410. Zwraca liczniki.
+    `dry_run`: tylko liczy, nic nie oznacza (do sprawdzenia na historii przed backfillem)."""
+    mark = (lambda *_a: None) if dry_run else _mark_dead
+    from app.extractors import HosterDead, TubePageError, is_vps_blocked_fallback, try_extract
+
+    out = {"checked": 0, "dead": 0, "alive": 0, "unknown": 0}
+    if not origin.startswith("tube:"):
+        return out
+    sitetag = origin.split(":", 1)[1]
+    if is_vps_blocked_fallback(sitetag):
+        return out  # rozwiazywane tylko na telefonie, serwer nie ma jak sprawdzic
+    with session_scope() as s:
+        rows = s.execute(
+            select(PlaybackSource.id, PlaybackSource.page_url).where(
+                PlaybackSource.scene_id == scene_id,
+                PlaybackSource.origin == origin,
+                PlaybackSource.dead_at.is_(None),
+            )
+        ).all()
+    for pb_id, page_url in rows:
+        out["checked"] += 1
+        try:
+            sources = try_extract(sitetag, page_url)
+        except HosterDead as e:
+            mark(pb_id, f"{e} (zgloszenie 410 z telefonu)")
+            out["dead"] += 1
+            continue
+        except TubePageError as e:
+            if e.status_code in (404, 410):
+                mark(pb_id, f"tube page {e.status_code} (zgloszenie 410 z telefonu)")
+                out["dead"] += 1
+            else:
+                out["unknown"] += 1
+            continue
+        except Exception:  # noqa: BLE001
+            out["unknown"] += 1
+            continue
+        direct = [x for x in (sources or []) if (x.type or "mp4") in ("mp4", "m3u8")]
+        if not direct:
+            out["unknown"] += 1
+            continue
+        codes = [_file_status(x.link, x.referer or page_url) for x in direct[:3]]
+        if codes and all(c in (404, 410) for c in codes):
+            mark(pb_id, f"plik {codes[0]} z telefonu i z serwera")
+            out["dead"] += 1
+        elif any(c in (200, 206) for c in codes):
+            out["alive"] += 1
+        else:
+            out["unknown"] += 1
+    return out
+
+
+def verify_gone_async(scene_id: uuid.UUID | None, origin: str) -> None:
+    """Fire-and-forget z endpointu telemetrii; ta sama para najwyzej raz na godzine."""
+    if scene_id is None:
+        return
+    key = (scene_id, origin)
+    now = time.monotonic()
+    with _lock:
+        if now - _checked.get(key, -1e9) < _RECHECK_SEC:
+            return
+        _checked[key] = now
+        if len(_checked) > 20000:
+            for k in [k for k, t in _checked.items() if now - t > _RECHECK_SEC]:
+                _checked.pop(k, None)
+
+    def run() -> None:
+        try:
+            verify_gone(scene_id, origin)
+        except Exception:  # noqa: BLE001
+            log.exception("gone: sprawdzenie %s %s nie wyszlo", scene_id, origin)
+
+    threading.Thread(target=run, name="gone-check", daemon=True).start()

scripts/verify_gone_backfill.py

@@ -0,0 +1,38 @@
+"""Jednorazowo: zrodla, ktore telefony zglosily jako 404/410, sprawdzone z serwera.
+
+Ta sama logika co na zywo (app/playback_gone.py), dla zgloszen sprzed wdrozenia.
+  docker compose exec -T worker python3 /srv/scripts/verify_gone_backfill.py [dni] [--commit] [--limit N]
+Bez --commit tylko liczy.
+"""
+from __future__ import annotations
+
+import sys
+from collections import Counter
+
+from sqlalchemy import text
+
+from app.db import session_scope
+from app.playback_gone import verify_gone
+
+days = int(next((a for a in sys.argv[1:] if a.isdigit()), "14"))
+commit = "--commit" in sys.argv
+limit = int(sys.argv[sys.argv.index("--limit") + 1]) if "--limit" in sys.argv else 100000
+
+with session_scope() as s:
+    pairs = s.execute(text("""
+        SELECT DISTINCT e.scene_id, e.origin FROM playback_events e
+        WHERE e.created_at > now() - make_interval(days => :d) AND e.status = 'error'
+          AND e.error_detail ~* 'response code: (404|410)' AND e.scene_id IS NOT NULL
+          AND EXISTS (SELECT 1 FROM playback_sources p WHERE p.scene_id = e.scene_id
+                      AND p.origin = e.origin AND p.dead_at IS NULL)
+        LIMIT :lim"""), {"d": days, "lim": limit}).all()
+print(f"par (scena, serwis) do sprawdzenia: {len(pairs)}, {'NA SERIO' if commit else 'na sucho'}", flush=True)
+total: Counter = Counter()
+per_origin: dict[str, Counter] = {}
+for scene_id, origin in pairs:
+    r = verify_gone(scene_id, origin, dry_run=not commit)
+    total.update(r)
+    per_origin.setdefault(origin, Counter()).update(r)
+for o, c in sorted(per_origin.items(), key=lambda kv: -kv[1]["checked"]):
+    print(f"  {o:24s} {dict(c)}")
+print("razem:", dict(total))