"""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/<post>` 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)