diff --git a/app/config.py b/app/config.py index 009a721..f3650e5 100644 --- a/app/config.py +++ b/app/config.py @@ -123,6 +123,20 @@ class Settings(BaseSettings): sched_phash_blacklist_hours: int = Field( default=24, validation_alias="GOON_SCHED_PHASH_BLACKLIST_HOURS" ) + # Próg ciszy dla konektorów FILMOWYCH. Wcześniej filmy nie były pokryte watchdogiem + # w ogóle — streamporn.vip stał 23 dni i nic nie krzyknęło. Filmy ingestujemy + # codziennie, więc 72h to bezpieczny margines na jeden nieudany cykl. + ingest_watchdog_movie_max_age_hours: int = Field( + default=72, validation_alias="GOON_INGEST_WATCHDOG_MOVIE_MAX_AGE_HOURS" + ) + # Dobowy digest zamrożonych źródeł na Slacka. Sentry dostaje alert co 6h, ale + # tam sygnał ginął (patrz docstring ingest_watchdog). 0 = off. + sched_watchdog_digest_hours: int = Field( + default=24, validation_alias="GOON_SCHED_WATCHDOG_DIGEST_HOURS" + ) + # Slack — bez OBU wartości powiadomienia są no-opem. Kanału nie zgadujemy. + slack_bot_token: str = Field(default="", validation_alias="GOON_SLACK_BOT_TOKEN") + slack_channel: str = Field(default="", validation_alias="GOON_SLACK_CHANNEL") # Śmieciowi performerzy — kategorie i studia podszywające się pod osoby („Creampie", # „Natural tits", „Brazzers"). Audyt 2026-07-27: 277 rekordów, 33 784 przypisania. # Odrastają, bo tube-search dokleja performera po jednym tokenie. 24h. 0 = off. diff --git a/app/notify/__init__.py b/app/notify/__init__.py new file mode 100644 index 0000000..95f98c5 --- /dev/null +++ b/app/notify/__init__.py @@ -0,0 +1 @@ +"""Powiadomienia wychodzace (Slack).""" diff --git a/app/notify/slack.py b/app/notify/slack.py new file mode 100644 index 0000000..c7397eb --- /dev/null +++ b/app/notify/slack.py @@ -0,0 +1,50 @@ +"""Wysyłka powiadomień na Slacka — kanał, który człowiek realnie czyta. + +Powód powstania (2026-08-03): watchdog świeżości poprawnie wykrywał zamrożone źródła +(vjav stał 12 dni, streamporn.vip 23), ale sygnał ginął. Szedł wyłącznie do Sentry ze +stabilnym fingerprintem i poziomem `warning`, więc powstawało JEDNO issue przy +pierwszym wystąpieniu i potem tylko rósł mu licznik — Sentry powiadamia o nowych +i regresjach, nie o kolejnych wystąpieniach otwartego issue. + +Domyślnie WYŁĄCZONE: bez `GOON_SLACK_BOT_TOKEN` i `GOON_SLACK_CHANNEL` funkcja jest +no-opem i nic nie próbuje wysyłać. Nie zgadujemy kanału. +""" +from __future__ import annotations + +import logging + +import httpx + +from app.config import get_settings + +log = logging.getLogger(__name__) + +_API = "https://slack.com/api/chat.postMessage" + + +def send_slack(text: str, *, timeout: float = 15.0) -> bool: + """Wyślij wiadomość na skonfigurowany kanał. Zwraca True gdy poszła. + + Cicho zwraca False gdy brak konfiguracji — to normalny stan, nie błąd. + """ + settings = get_settings() + token = getattr(settings, "slack_bot_token", "") or "" + channel = getattr(settings, "slack_channel", "") or "" + if not token or not channel: + return False + try: + r = httpx.post( + _API, + headers={"Authorization": f"Bearer {token}"}, + json={"channel": channel, "text": text}, + timeout=timeout, + ) + data = r.json() + if not data.get("ok"): + # Slack zwraca 200 nawet przy błędzie logicznym — sprawdzamy `ok`. + log.warning("slack: wysyłka odrzucona: %s", data.get("error")) + return False + return True + except Exception as e: + log.warning("slack: wysyłka nieudana: %s", e) + return False diff --git a/app/scheduler/ingest_watchdog.py b/app/scheduler/ingest_watchdog.py index 600ea42..f634fed 100644 --- a/app/scheduler/ingest_watchdog.py +++ b/app/scheduler/ingest_watchdog.py @@ -1,20 +1,32 @@ -"""Per-sitetag freshness watchdog — alert gdy aktywny tube przestał dawać nowe sceny. +"""Per-origin freshness watchdog — alert gdy aktywne źródło przestało dawać nowe treści. Globalny monitor źródeł (ingest_runs per `Source`) tego NIE łapie, bo wszystkie tube scrapery dzielą jeden `Source` = "tube-scraper" — pojedynczy origin może zamarznąć (np. freshporno: scraper browsował z roota `/`, który KVS rotuje → cold-session dostawała stary zestaw → new=0/skipped=N przez 2 dni), a zagregowany run nadal -raportuje success. Sygnał per-origin: `max(created_at)` na playback_sources danego -`tube:`. Jak zamrożony > próg → alert (report 14f3a655 2026-06-15). +raportuje success. Sygnał per-origin: `max(created_at)` na źródłach playbacku. -Pokrywamy DWIE klasy scraperów, każda z własnym progiem: +Pokrywamy TRZY klasy, każda z własnym progiem: - **browse** (`ALL_BROWSE_SCRAPERS`) — crawlowane codziennie z listingu, próg 48h. - **search** (`ALL_DIRECT_SCRAPERS`) — performer-driven, nierówna kadencja (~30d - refresh per performer), próg wyższy (domyślnie 7d). Bez tego pokrycia kilka - search-tubów (sxyland, latestpornvideo, perverzija, fpoxxx, mypornerleak, porndish) - zamarzło cicho 2026-05-07/06-07/06-13 i nic nie krzyknęło do Sentry. -Tag obecny w obu listach (xvideoscom, epornercom — i browse i search) liczymy jako -browse (ostrzejszy próg). + refresh per performer), próg wyższy (domyślnie 7d). + - **movies** (`_MOVIE_CONNECTORS`) — dodane 2026-08-03. Wcześniej filmy NIE BYŁY + pokryte w ogóle: streamporn.vip stał 23 dni (czytał listing nieposortowany po + dacie) i żaden automat nie miał jak tego zauważyć. + +Tag obecny w kilku listach liczymy wg najostrzejszego progu. + +**Eskalacja zamiast jednego cichego issue.** Do 2026-08-03 zdarzenie szło zawsze jako +`warning` ze stabilnym fingerprintem per origin. Fingerprint jest tam po to, żeby nie +tworzyć nowego issue co 6h — ale skutkiem ubocznym było JEDNO issue przy pierwszym +wystąpieniu, potem tylko rosnący licznik. Sentry powiadamia o nowych i regresjach, nie +o kolejnych wystąpieniach otwartego issue, a reguły alertów zwykle celują w `error`, +nie `warning`. Efekt: vjav stał 12 dni, watchdog go poprawnie wypisywał co cykl, i nikt +się nie dowiedział. + +Teraz fingerprint zawiera KUBEŁEK WIEKU, więc przekroczenie każdego kolejnego progu +zakłada nowe issue (= nowe powiadomienie), a w obrębie kubełka dalej nie ma spamu. +Od 7 dni ciszy poziom idzie na `error`, żeby wpaść w standardowe reguły alertów. Patrz [[reference_kvs_root_rotates_use_latest_updates]] dla klasy błędu, którą to łapie. """ @@ -30,58 +42,87 @@ from app.db import session_scope log = logging.getLogger(__name__) +#: Kubełki wieku ciszy (godziny, malejąco) → etykieta + poziom Sentry. Wejście w +#: kolejny kubełek zmienia fingerprint, czyli zakłada NOWE issue i wysyła alert. +_BUCKETS: tuple[tuple[int, str, str], ...] = ( + (720, "30d+", "error"), + (168, "7d+", "error"), + (48, "2d+", "warning"), + (0, "swiezo", "warning"), +) + + +def _bucket(age_h: float) -> tuple[str, str]: + for floor, label, level in _BUCKETS: + if age_h >= floor: + return label, level + return "swiezo", "warning" + def run_ingest_freshness_watchdog( *, max_age_hours: int = 48, search_max_age_hours: int = 168, + movie_max_age_hours: int = 72, min_history: int = 100, + to_sentry: bool = True, + to_slack: bool = False, ) -> dict[str, Any]: - """Sprawdź każdy aktywny scraper: czy origin dostał nową scenę < próg dla swojej klasy. + """Sprawdź każde aktywne źródło: czy dostało nową treść < próg dla swojej klasy. - Skanujemy sitetagi z `ALL_BROWSE_SCRAPERS` (próg `max_age_hours`, domyślnie 48h) oraz - z `ALL_DIRECT_SCRAPERS` (performer-driven search, próg `search_max_age_hours`, domyślnie - 7d — nierówna kadencja, 48h dawałoby false-positivy). Tag w obu listach liczymy jako - browse (ostrzejszy próg). Nie skanujemy wszystkich origin-ów, żeby nie alarmować o - legacy/jednorazowych źródłach. `min_history` odsiewa świeżo dodane tuby bez ustalonej - kadencji (za mało scen, by wiedzieć czy cisza to anomalia). + `min_history` odsiewa świeżo dodane źródła bez ustalonej kadencji (za mało pozycji, + by wiedzieć, czy cisza to anomalia). - Stale origin → Sentry `capture_message` ze stabilnym fingerprintem per origin - (wiek + próg w extra, nie w tytule — inaczej każdy run = nowe issue). Zwraca - {checked, stale:[{origin, age_hours, total, kind, max_age_hours}]}. + `to_sentry` / `to_slack` sterują kanałami — job co 6h woła Sentry, a osobny job + dobowy woła Slacka, żeby nie wysyłać tej samej listy cztery razy dziennie. """ - from app.connectors.direct_scrapers import ( + from app.connectors import get_movie_connectors # noqa: PLC0415 + from app.connectors.direct_scrapers import ( # noqa: PLC0415 ALL_BROWSE_SCRAPERS, ALL_DIRECT_SCRAPERS, ) browse_tags = {cls.sitetag for cls in ALL_BROWSE_SCRAPERS} - # Search-tuby dzielące tag z browse (xvideoscom, epornercom) idą pod browse-próg. search_tags = {cls.sitetag for cls in ALL_DIRECT_SCRAPERS} - browse_tags - # (sitetag, próg_h, klasa) — klasa tylko do logów/extra w Sentry. - checks: list[tuple[str, int, str]] = [ - (tag, max_age_hours, "browse") for tag in sorted(browse_tags) - ] + [(tag, search_max_age_hours, "search") for tag in sorted(search_tags)] + # (origin, tabela, próg_h, klasa) + checks: list[tuple[str, str, int, str]] = ( + [(f"tube:{t}", "playback_sources", max_age_hours, "browse") for t in sorted(browse_tags)] + + [(f"tube:{t}", "playback_sources", search_max_age_hours, "search") + for t in sorted(search_tags)] + ) + try: + for name, _cls in get_movie_connectors(): + checks.append((name, "movie_playback_sources", movie_max_age_hours, "movies")) + except Exception as e: # pragma: no cover - rejestr filmów niedostępny + log.warning("ingest-watchdog: nie udało się pobrać konektorów filmowych: %s", e) now = datetime.now(UTC) stale: list[dict[str, Any]] = [] with session_scope() as s: - for tag, threshold, kind in checks: - origin = f"tube:{tag}" - row = s.execute( - text( - "SELECT max(created_at) AS newest, count(*) AS total " - "FROM playback_sources WHERE origin = :o" - ), - {"o": origin}, - ).one() + for origin, table, threshold, kind in checks: + # Filmy trzymają origin jako `:`, sceny jako dokładne + # `tube:` — stąd LIKE dla filmów, równość dla scen. + if table == "movie_playback_sources": + sql = ( + f"SELECT max(created_at) AS newest, count(*) AS total FROM {table} " + "WHERE origin = :o OR origin LIKE :p" + ) + params = {"o": origin, "p": f"{origin}:%"} + else: + sql = ( + f"SELECT max(created_at) AS newest, count(*) AS total FROM {table} " + "WHERE origin = :o" + ) + params = {"o": origin} + row = s.execute(text(sql), params).one() newest, total = row.newest, row.total if total < min_history or newest is None: continue age_h = (now - newest).total_seconds() / 3600.0 if age_h >= threshold: + label, level = _bucket(age_h) stale.append( { "origin": origin, @@ -89,39 +130,55 @@ def run_ingest_freshness_watchdog( "total": total, "kind": kind, "max_age_hours": threshold, + "bucket": label, + "level": level, } ) - if stale: - log.warning( - "ingest-watchdog: %d/%d origin(s) bez nowych scen (browse>%dh / search>%dh): %s", - len(stale), len(checks), max_age_hours, search_max_age_hours, - ", ".join(f"{x['origin']}({x['age_hours']}h,{x['kind']})" for x in stale), - ) + if not stale: + log.info("ingest-watchdog: wszystkie %d źródła świeże", len(checks)) + return {"checked": len(checks), "stale": []} + + stale.sort(key=lambda x: -x["age_hours"]) + log.warning( + "ingest-watchdog: %d/%d źródeł bez nowych treści: %s", + len(stale), len(checks), + ", ".join(f"{x['origin']}({x['age_hours']}h,{x['kind']})" for x in stale), + ) + + if to_sentry: try: import sentry_sdk for x in stale: with sentry_sdk.push_scope() as scope: - scope.level = "warning" + scope.level = x["level"] scope.set_tag("ingest_origin", x["origin"]) scope.set_tag("ingest_scraper_kind", x["kind"]) + scope.set_tag("ingest_stale_bucket", x["bucket"]) scope.set_extra("age_hours", x["age_hours"]) - scope.set_extra("total_scenes", x["total"]) + scope.set_extra("total_items", x["total"]) scope.set_extra("max_age_hours", x["max_age_hours"]) - # Fingerprint per origin → jedno trwałe issue na zamrożony tube, - # nie fragmentowane przez zmienny wiek. - scope.fingerprint = ["ingest-stale-origin", x["origin"]] + # Kubełek W fingerprincie: nowe issue (= nowy alert) przy każdym + # kolejnym progu, ale bez spamu co 6h w obrębie tego samego progu. + scope.fingerprint = ["ingest-stale-origin", x["origin"], x["bucket"]] sentry_sdk.capture_message( - f"ingest-watchdog: {x['origin']} ({x['kind']}) bez nowych scen " + f"ingest-watchdog: {x['origin']} ({x['kind']}) bez nowych treści " f"({x['age_hours']:.0f}h, próg {x['max_age_hours']}h)" ) except Exception: # pragma: no cover - Sentry off / brak DSN log.exception("ingest-watchdog: Sentry capture failed") - else: - log.info( - "ingest-watchdog: wszystkie %d origin świeże (browse<%dh / search<%dh)", - len(checks), max_age_hours, search_max_age_hours, - ) + + if to_slack: + from app.notify.slack import send_slack # noqa: PLC0415 + + lines = [f"*Zamrożone źródła ingestu* ({len(stale)}/{len(checks)})"] + for x in stale: + days = x["age_hours"] / 24.0 + lines.append( + f"• `{x['origin']}` ({x['kind']}) — {days:.1f} dnia bez nowych treści " + f"(próg {x['max_age_hours']}h, {x['total']} pozycji w bazie)" + ) + send_slack("\n".join(lines)) return {"checked": len(checks), "stale": stale} diff --git a/app/scheduler/jobs.py b/app/scheduler/jobs.py index e858c2f..4273d53 100644 --- a/app/scheduler/jobs.py +++ b/app/scheduler/jobs.py @@ -324,17 +324,21 @@ def _job_title_duration_dedup() -> None: log.exception("[scheduler] title-duration dedup failed") -def _job_ingest_watchdog(max_age_hours: int, search_max_age_hours: int) -> None: - """Per-origin freshness watchdog — alert do Sentry gdy aktywny tube przestał dawać - nowe sceny > próg (browse: max_age_hours, search: search_max_age_hours). Globalny - monitor (jeden Source 'tube-scraper') tego nie łapie; pojedynczy origin może zamarznąć - przy success-runie (report 14f3a655).""" +def _job_ingest_watchdog( + max_age_hours: int, search_max_age_hours: int, movie_max_age_hours: int +) -> None: + """Per-origin freshness watchdog → Sentry. Obejmuje sceny (browse/search) ORAZ + filmy (od 2026-08-03; wcześniej filmy nie były pokryte i streamporn.vip stał 23 dni + niezauważony). Globalny monitor (jeden Source 'tube-scraper') tego nie łapie; + pojedynczy origin może zamarznąć przy success-runie (report 14f3a655).""" try: from app.scheduler.ingest_watchdog import run_ingest_freshness_watchdog res = run_ingest_freshness_watchdog( max_age_hours=max_age_hours, search_max_age_hours=search_max_age_hours, + movie_max_age_hours=movie_max_age_hours, + to_sentry=True, ) if res["stale"]: log.warning("[scheduler] ingest-watchdog: %d stale origin(s)", len(res["stale"])) @@ -342,6 +346,26 @@ def _job_ingest_watchdog(max_age_hours: int, search_max_age_hours: int) -> None: log.exception("[scheduler] ingest-watchdog failed") +def _job_watchdog_digest( + max_age_hours: int, search_max_age_hours: int, movie_max_age_hours: int +) -> None: + """Dobowy digest zamrożonych źródeł na Slacka. Osobno od joba co 6h, żeby nie + wysyłać tej samej listy cztery razy dziennie; Sentry zostaje kanałem maszynowym, + Slack ludzkim. Bez GOON_SLACK_BOT_TOKEN i GOON_SLACK_CHANNEL to no-op.""" + try: + from app.scheduler.ingest_watchdog import run_ingest_freshness_watchdog + + run_ingest_freshness_watchdog( + max_age_hours=max_age_hours, + search_max_age_hours=search_max_age_hours, + movie_max_age_hours=movie_max_age_hours, + to_sentry=False, + to_slack=True, + ) + except Exception: + log.exception("[scheduler] watchdog-digest failed") + + def _job_hetzner_monitor() -> None: """Hetzner Cloud bandwidth monitor — alert do Sentry przy progach % included_traffic. No-op gdy brak HETZNER_API_TOKEN/SERVER_ID w env (loguje że wyłączony).""" @@ -538,8 +562,9 @@ def build_scheduler(cfg: dict[str, Any]) -> BlockingScheduler: if cfg.get("ingest_watchdog_hours"): wd_max_age = cfg.get("ingest_watchdog_max_age_hours") or 48 wd_search_max_age = cfg.get("ingest_watchdog_search_max_age_hours") or 168 + wd_movie_max_age = cfg.get("ingest_watchdog_movie_max_age_hours") or 72 sched.add_job( - lambda: _job_ingest_watchdog(wd_max_age, wd_search_max_age), + lambda: _job_ingest_watchdog(wd_max_age, wd_search_max_age, wd_movie_max_age), IntervalTrigger(hours=cfg["ingest_watchdog_hours"], start_date=INTERVAL_ANCHOR), id="ingest_watchdog", replace_existing=True, @@ -547,10 +572,23 @@ def build_scheduler(cfg: dict[str, Any]) -> BlockingScheduler: coalesce=True, ) log.info( - "scheduler: ingest-watchdog every %dh (browse max_age=%dh, search max_age=%dh)", - cfg["ingest_watchdog_hours"], wd_max_age, wd_search_max_age, + "scheduler: ingest-watchdog every %dh (browse=%dh, search=%dh, movies=%dh)", + cfg["ingest_watchdog_hours"], wd_max_age, wd_search_max_age, wd_movie_max_age, ) + if cfg.get("watchdog_digest_hours"): + sched.add_job( + lambda: _job_watchdog_digest(wd_max_age, wd_search_max_age, wd_movie_max_age), + IntervalTrigger( + hours=cfg["watchdog_digest_hours"], start_date=INTERVAL_ANCHOR + ), + id="watchdog_digest", + replace_existing=True, + max_instances=1, + coalesce=True, + ) + log.info("scheduler: watchdog-digest (Slack) every %dh", cfg["watchdog_digest_hours"]) + if cfg.get("hetzner_monitor_hours"): sched.add_job( _job_hetzner_monitor, diff --git a/app/scheduler/worker.py b/app/scheduler/worker.py index 5db04b4..9663c38 100644 --- a/app/scheduler/worker.py +++ b/app/scheduler/worker.py @@ -233,6 +233,12 @@ def run_forever() -> int: "ingest_watchdog_search_max_age_hours": getattr( settings, "ingest_watchdog_search_max_age_hours", 168 ), + # Filmy — wcześniej BEZ pokrycia watchdogiem; streamporn.vip stał 23 dni. + "ingest_watchdog_movie_max_age_hours": getattr( + settings, "ingest_watchdog_movie_max_age_hours", 72 + ), + # Dobowy digest na Slacka (Sentry gubił sygnał na stabilnym fingerprincie). + "watchdog_digest_hours": getattr(settings, "sched_watchdog_digest_hours", 24) or None, # Hetzner Cloud bandwidth monitor — alert do Sentry przy progach % included. # No-op gdy brak HETZNER_API_TOKEN/SERVER_ID (sam job może być on). "hetzner_monitor_hours": getattr(settings, "sched_hetzner_monitor_hours", 6) or None,