goon/app/scheduler/quality_watchdog.py

"""Watchdogi jakosci zrodel, uzupelnienie freshness z ingest_watchdog.py.

Freshness widzi tylko "brak nowych pozycji". Cztery klasy awarii, ktore przez to przechodzily
(wszystkie z 10.2026): watchporn zmienil motyw i parser dopinal 12 performerow na scene
z kart "related" (ingest "zdrowy", new>0); galaxyporn, latestpornvideo i porndish oddaja
sceny bez dlugosci (apka filtruje min 60 s, wiec sa niewidoczne, a licznik new rosnie);
porndoe oddawal 31 tych samych scen (listing pelny, new=0, nie wiadomo czy to serwis stoi,
czy czytamy zla strone); perverzija grala w 50% przy "zywych" zrodlach.

Trzy sprawdzenia, kazde z wlasnym fingerprintem w Sentry (jedno issue na zrodlo i rodzaj,
bez spamu co 6 h):

- **listing** (browse_runs): ostatnie przebiegi scrapera z seen=0 albo wyjatkiem = listing
  pusty lub blad; dlugi ciag seen>0 i new=0 u zrodla publikujacego codziennie = czytamy
  zla strone (rotacja karuzeli, endpoint nie sortowany po dacie).
- **parser** (playback_sources + scenes z 24 h wobec 14 dni wczesniej): skok lub zanik
  performerow na scene, udzial scen bez dlugosci, bez miniatury. Lapie zmiane motywu
  strony w pierwszej dobie, bez czekania na zgloszenie z apki.
- **playback** (playback_events 24 h wobec 7 dni): spadek udanych odtworzen per origin.

`to_sentry=False` daje sam odczyt (reczne sprawdzenie bez zasmiecania issue).
"""
from __future__ import annotations

import logging
import statistics
from collections import defaultdict
from typing import Any

from sqlalchemy import text

from app.db import session_scope

log = logging.getLogger(__name__)

# listing
_LISTING_EMPTY_STREAK = 3        # tyle ostatnich przebiegow z seen=0/bledem z rzedu
_LISTING_ALLKNOWN_STREAK = 8     # tyle przebiegow (48 h przy co 6 h) z seen>0 i new=0
_LISTING_ALLKNOWN_MAX_GAP_DAYS = 2.0  # tylko zrodla publikujace co najmniej co 2 dni
# parser
_PARSER_MIN_RECENT = 10
_PARSER_MIN_BASE = 30
# playback
_PLAYBACK_MIN_ATTEMPTS = 30
_PLAYBACK_MIN_BASE_ATTEMPTS = 50


def _cadence_gap_days(session, origin: str, lookback_days: int = 60) -> float | None:
    """Mediana odstepu (w dobach) miedzy dniami, w ktorych zrodlo dalo nowe pozycje."""
    rows = session.execute(
        text(
            "SELECT DISTINCT created_at::date AS d FROM playback_sources "
            "WHERE origin = :o AND created_at > now() - make_interval(days => :lb) ORDER BY d"
        ),
        {"o": origin, "lb": lookback_days},
    ).all()
    days = [r[0] for r in rows]
    if len(days) < 5:
        return None
    gaps = [(b - a).days for a, b in zip(days, days[1:])]
    return float(statistics.median(gaps)) if gaps else None


def check_listing_health(session) -> list[dict[str, Any]]:
    rows = session.execute(
        text(
            "SELECT sitetag, started_at, seen, new, skipped, failed, note FROM browse_runs "
            "WHERE started_at > now() - interval '60 hours' ORDER BY sitetag, started_at DESC"
        )
    ).all()
    by_tag: dict[str, list] = defaultdict(list)
    for r in rows:
        by_tag[r.sitetag].append(r)
    out: list[dict[str, Any]] = []
    for sitetag, runs in by_tag.items():
        last = runs[:_LISTING_EMPTY_STREAK]
        if len(last) >= _LISTING_EMPTY_STREAK and all(r.failed or r.seen == 0 for r in last):
            note = next((r.note for r in last if r.note), None)
            out.append(
                {
                    "sitetag": sitetag,
                    "problem": "empty",
                    "runs": len(last),
                    "note": note,
                    "level": "warning",
                    "message": (
                        f"listing-watchdog: {sitetag} listing pusty albo blad w {len(last)} "
                        f"przebiegach z rzedu" + (f" ({note[:120]})" if note else "")
                    ),
                }
            )
            continue
        last = runs[:_LISTING_ALLKNOWN_STREAK]
        if len(last) >= _LISTING_ALLKNOWN_STREAK and all(
            (not r.failed) and r.seen > 0 and r.new == 0 for r in last
        ):
            gap = _cadence_gap_days(session, f"tube:{sitetag}")
            if gap is not None and gap <= _LISTING_ALLKNOWN_MAX_GAP_DAYS:
                seen = sum(r.seen for r in last)
                out.append(
                    {
                        "sitetag": sitetag,
                        "problem": "all_known",
                        "runs": len(last),
                        "seen": seen,
                        "gap_days": gap,
                        "level": "warning",
                        "message": (
                            f"listing-watchdog: {sitetag} {len(last)} przebiegi z rzedu, {seen} pozycji "
                            f"i 0 nowych, a zrodlo publikuje co {gap:.0f} d: zla strona listingu "
                            f"(rotacja, brak sortu po dacie) albo serwis stanal"
                        ),
                    }
                )
    return out


def check_parser_quality(session) -> list[dict[str, Any]]:
    rows = session.execute(
        text(
            """
            WITH w AS (
              SELECT ps.origin, ps.scene_id, ps.thumbnail_url,
                     (ps.created_at > now() - interval '24 hours') AS recent
              FROM playback_sources ps
              WHERE ps.origin LIKE 'tube:%' AND ps.created_at > now() - interval '15 days'
            )
            SELECT w.origin, w.recent, count(*) AS n,
                   avg((SELECT count(*) FROM scene_performers sp WHERE sp.scene_id = w.scene_id)) AS perf,
                   avg(CASE WHEN x.duration_sec IS NULL THEN 1.0 ELSE 0.0 END) AS dur_null,
                   avg(CASE WHEN w.thumbnail_url IS NULL THEN 1.0 ELSE 0.0 END) AS thumb_null
            FROM w JOIN scenes x ON x.id = w.scene_id
            GROUP BY 1, 2
            """
        )
    ).all()
    stats: dict[str, dict[bool, Any]] = defaultdict(dict)
    for r in rows:
        stats[r.origin][bool(r.recent)] = r
    out: list[dict[str, Any]] = []
    for origin, parts in stats.items():
        rec, base = parts.get(True), parts.get(False)
        if rec is None or base is None or rec.n < _PARSER_MIN_RECENT or base.n < _PARSER_MIN_BASE:
            continue
        pr, pb = float(rec.perf or 0), float(base.perf or 0)
        dr, db = float(rec.dur_null or 0), float(base.dur_null or 0)
        tr, tb = float(rec.thumb_null or 0), float(base.thumb_null or 0)
        probs: list[tuple[str, str]] = []
        if pb >= 0 and pr >= 2.5 * max(pb, 0.4) and pr - pb >= 3:
            probs.append(("perf_inflated", f"obsada rozdeta: {pr:.1f} performerow na scene (wczesniej {pb:.1f}), karty related?"))
        if pb >= 0.8 and pr <= 0.25 * pb:
            probs.append(("perf_lost", f"obsada zniknela: {pr:.2f} na scene (wczesniej {pb:.2f})"))
        if dr >= 0.5 and db <= 0.25:
            probs.append(("duration_lost", f"{dr*100:.0f}% scen bez dlugosci (wczesniej {db*100:.0f}%), apka je ukrywa"))
        if tr >= 0.5 and tb <= 0.25:
            probs.append(("thumb_lost", f"{tr*100:.0f}% zrodel bez miniatury (wczesniej {tb*100:.0f}%)"))
        for metric, desc in probs:
            out.append(
                {
                    "origin": origin,
                    "metric": metric,
                    "recent_n": int(rec.n),
                    "base_n": int(base.n),
                    "level": "warning",
                    "message": f"parser-watchdog: {origin} {desc} (24 h: {int(rec.n)} scen, baza {int(base.n)})",
                }
            )
    return out


def check_playback_health(session) -> list[dict[str, Any]]:
    rows = session.execute(
        text(
            """
            SELECT origin,
                   count(*) FILTER (WHERE created_at > now() - interval '24 hours') AS att24,
                   avg(CASE WHEN status = 'success' THEN 1.0 ELSE 0.0 END)
                       FILTER (WHERE created_at > now() - interval '24 hours') AS ok24,
                   count(*) FILTER (WHERE created_at <= now() - interval '24 hours') AS attb,
                   avg(CASE WHEN status = 'success' THEN 1.0 ELSE 0.0 END)
                       FILTER (WHERE created_at <= now() - interval '24 hours') AS okb
            FROM playback_events
            WHERE created_at > now() - interval '8 days'
            GROUP BY 1
            """
        )
    ).all()
    out: list[dict[str, Any]] = []
    for r in rows:
        att24 = int(r.att24 or 0)
        if att24 < _PLAYBACK_MIN_ATTEMPTS or r.ok24 is None:
            continue
        ok24 = float(r.ok24)
        okb = float(r.okb) if r.okb is not None else None
        attb = int(r.attb or 0)
        bucket = None
        if ok24 < 0.3:
            bucket, level = "lt30", "error"
        elif ok24 < 0.6 and okb is not None and attb >= _PLAYBACK_MIN_BASE_ATTEMPTS and okb - ok24 >= 0.25:
            bucket, level = "lt60", "warning"
        if not bucket:
            continue
        base_txt = f", tydzien wczesniej {okb*100:.0f}%" if okb is not None else ""
        out.append(
            {
                "origin": r.origin,
                "bucket": bucket,
                "ok24": round(ok24, 3),
                "att24": att24,
                "level": level,
                "message": f"playback-watchdog: {r.origin} {ok24*100:.0f}% udanych z {att24} prob w 24 h{base_txt}",
            }
        )
    return out


def _emit(items: list[dict[str, Any]], *, kind: str, key: str, sub: str) -> None:
    try:
        import sentry_sdk
    except Exception:  # pragma: no cover
        return
    for x in items:
        try:
            with sentry_sdk.push_scope() as scope:
                scope.level = x["level"]
                scope.set_tag("watchdog_kind", kind)
                scope.set_tag("watchdog_origin", str(x.get(key)))
                for k, v in x.items():
                    if k not in ("message", "level"):
                        scope.set_extra(k, v)
                scope.fingerprint = [kind, str(x.get(key)), str(x.get(sub))]
                sentry_sdk.capture_message(x["message"])
        except Exception:  # pragma: no cover
            log.exception("quality-watchdog: Sentry capture failed")


def run_quality_watchdogs(*, to_sentry: bool = True) -> dict[str, Any]:
    with session_scope() as s:
        listing = check_listing_health(s)
        parser = check_parser_quality(s)
        playback = check_playback_health(s)
    for group, name in ((listing, "listing"), (parser, "parser"), (playback, "playback")):
        for x in group:
            log.warning("quality-watchdog [%s]: %s", name, x["message"])
    if to_sentry:
        _emit(listing, kind="listing-watchdog", key="sitetag", sub="problem")
        _emit(parser, kind="parser-watchdog", key="origin", sub="metric")
        _emit(playback, kind="playback-watchdog", key="origin", sub="bucket")
    summary = {"listing": len(listing), "parser": len(parser), "playback": len(playback)}
    if not any(summary.values()):
        log.info("quality-watchdog: bez uwag")
    return {"summary": summary, "listing": listing, "parser": parser, "playback": playback}