"""Oznacza martwe zrodla KVS (yespornvip i inne) ze strona sceny 404/410 (2026-10-15).

resolve_kvs do 15.10 zamienial 404 na "chwilowy blad", wiec usuniete filmy wisialy jako
zywe (yespornvip: ok. polowa z 39 tys. zrodel, zgloszenie 5e81eb76). Od 15.10 resolve sam
oznacza 404 przy kliknieciu, ten skrypt czysci zaleglosci bez czekania na klikniecia.

Dwie kontrole (patrz notatka o przemiale zrodel): strona daje 404/410 DWA razy, w odstepie
co najmniej --recheck-after sekund, i w tresci nie ma odtwarzacza (license_code). Przekierowanie
na nowy adres z odtwarzaczem to zywe zrodlo i zostaje.

    python scripts/reap_kvs_404.py --origin tube:yespornvip --limit 100          # podglad
    python scripts/reap_kvs_404.py --origin tube:yespornvip --limit 50000 --apply
"""
from __future__ import annotations

import argparse
import logging
import time
from datetime import UTC, datetime

import sqlalchemy as sa

from app.db import session_scope
from app.extractors._fetch import browser_get

log = logging.getLogger(__name__)


def _status(url: str) -> tuple[int, bool]:
    try:
        r = browser_get(url, timeout=20.0, follow_redirects=True)
    except Exception:
        return 0, False
    return r.status_code, "license_code" in (r.text or "")


def run(origin: str, limit: int, apply: bool, rps: float, recheck_after: float) -> dict[str, int]:
    c = {"checked": 0, "alive": 0, "first_404": 0, "dead": 0, "other": 0}
    with session_scope() as s:
        rows = s.execute(
            sa.text(
                "SELECT id, page_url FROM playback_sources WHERE origin = :o AND dead_at IS NULL "
                "ORDER BY id LIMIT :lim"
            ),
            {"o": origin, "lim": limit},
        ).all()
    suspects: list[tuple] = []
    for pb_id, url in rows:
        c["checked"] += 1
        st, player = _status(url)
        if st in (404, 410) and not player:
            c["first_404"] += 1
            suspects.append((pb_id, url, time.time()))
        elif st and st < 400:
            c["alive"] += 1
        else:
            c["other"] += 1
        if c["checked"] % 500 == 0:
            log.info("kvs reap pass1 %s", c)
        time.sleep(1.0 / rps)
    dead: list = []
    for pb_id, url, t0 in suspects:
        wait = recheck_after - (time.time() - t0)
        if wait > 0:
            time.sleep(wait)
        st, player = _status(url)
        if st in (404, 410) and not player:
            dead.append(pb_id)
        time.sleep(1.0 / rps)
    c["dead"] = len(dead)
    print(c)
    if apply and dead:
        reason = f"tube page 404 (reap_kvs_404 {datetime.now(UTC):%Y-%m-%d})"
        for i in range(0, len(dead), 500):
            with session_scope() as s:
                s.execute(
                    sa.text("UPDATE playback_sources SET dead_at = now(), dead_reason = :r WHERE id = ANY(:ids)"),
                    {"r": reason, "ids": dead[i:i + 500]},
                )
                s.commit()
    return c


def main() -> int:
    logging.basicConfig(level=logging.INFO)
    p = argparse.ArgumentParser()
    p.add_argument("--origin", required=True)
    p.add_argument("--limit", type=int, default=100)
    p.add_argument("--apply", action="store_true")
    p.add_argument("--rps", type=float, default=3.0)
    p.add_argument("--recheck-after", type=float, default=60.0)
    args = p.parse_args()
    run(args.origin, args.limit, args.apply, args.rps, args.recheck_after)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())