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