"""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}