feat(workers): data-freshness monitor (Celery beat → GlitchTip)

Extract compute_freshness(db) from the /admin/scrape/freshness handler (endpoint unchanged) + daily beat task scrape_freshness_check that alerts via sentry_sdk.capture_message when a source is stale/failed (error on critical-failed, else warning). Registered in celery include + beat (09:00 MSK). Refs #73

backend/app/api/v1/admin_scrape.py

@@ -1343,12 +1343,13 @@ def _classify_freshness(age_days: float | None, fresh_days: float, stale_days: f
     return "critical"
 
 
-@router.get("/freshness")
+def compute_freshness(db: Session) -> dict[str, Any]:
-def scrape_freshness(
-    db: Annotated[Session, Depends(get_db)],
-) -> dict[str, Any]:
     """Свежесть данных по каждому источнику (для dashboard + alerting).
 
+    Чистая функция (один источник истины для HTTP-эндпоинта и beat-таски
+    scrape_freshness_check). Принимает открытую сессию, НЕ коммитит/закрывает её —
+    жизненным циклом сессии управляет вызывающий (FastAPI-зависимость или таска).
+
     Для каждого источника один проход по его run-таблице агрегирует:
       - last_success_at  — max(COALESCE(finished_at, started_at)) WHERE status='done'
       - last_attempt_at  — max(started_at) (или created_at fallback) по всем прогонам
@@ -1462,6 +1463,19 @@ def scrape_freshness(
     }
 
 
+@router.get("/freshness")
+def scrape_freshness(
+    db: Annotated[Session, Depends(get_db)],
+) -> dict[str, Any]:
+    """GET-обёртка над compute_freshness — только читает, не алертит.
+
+    Вся агрегация живёт в compute_freshness (один источник истины); эндпоинт
+    лишь прокидывает сессию и возвращает результат без изменений формы payload.
+    Sentry-алёрты по критичным источникам шлёт beat-таска scrape_freshness_check.
+    """
+    return compute_freshness(db)
+
+
 # ── DomRF (kn) — runs list ───────────────────────────────────────────────────
 
 

backend/app/workers/beat_schedule.py

@@ -557,4 +557,16 @@ def build_beat_schedule() -> dict:
         "options": {"queue": "celery"},
     }
 
+    # Data-freshness monitor: ежедневно 09:00 МСК (Celery conf.timezone=Europe/Moscow →
+    # crontab в МСК, #1233). Запускает compute_freshness по 5 run-таблицам и шлёт
+    # GlitchTip-алёрт, если критичный источник устарел/упал. 09:00 — после ночного
+    # кластера скрейперов (03:00–07:00) и в начале рабочего дня, чтобы оператор увидел
+    # деградацию свежести данных утром. Только читает + алертит (не пишет), дёшево.
+    # Техническая, не в job_settings (как refresh-analytics / mv-sales-tracker).
+    schedule["scrape-freshness-check-daily"] = {
+        "task": "tasks.scrape_freshness_check.scrape_freshness_check",
+        "schedule": _parse_cron("0 9 * * *"),  # 09:00 MSK ежедневно
+        "options": {"queue": "celery"},
+    }
+
     return schedule

backend/app/workers/celery_app.py

@@ -85,6 +85,7 @@ celery_app = Celery(
         "app.workers.tasks.refresh_layout_velocity",
         "app.workers.tasks.riasurt_sverdl_harvest",
         "app.workers.tasks.mv_sales_tracker_refresh",
+        "app.workers.tasks.scrape_freshness_check",
     ],
 )
 celery_app.conf.timezone = "Europe/Moscow"

backend/app/workers/tasks/scrape_freshness_check.py

@@ -0,0 +1,107 @@
+"""Celery-beat data-freshness monitor: алёрт в GlitchTip при устаревших/упавших источниках.
+
+GET /api/v1/admin/scrape/freshness считает свежесть по 5 run-таблицам, но только ЧИТАЕТ
+(чтобы UI-поллинг не спамил issue). Эта beat-таска периодически запускает ту же агрегацию
+(compute_freshness — единый источник истины) и пушит Sentry/GlitchTip-сообщение, когда есть
+проблемные источники.
+
+Какие источники считаем алёртабельными:
+  - status == "failed" (нет успехов вообще / возраст >= stale_days) — серьёзно;
+  - status == "stale"  (возраст между fresh_days и stale_days) — предупреждение.
+Severity сообщения: "error", если хотя бы один КРИТИЧНЫЙ источник (registry.critical=True)
+полностью упал (failed); иначе "warning". Если sentry в worker-процессе не инициализирован
+(нет glitchtip_dsn) — capture_message молча no-op, поэтому дополнительно логируем WARNING.
+
+Mirror conventions (zone_regulation_refresh.py / backend.md): SessionLocal()+try/finally;
+broad except не валит beat-loop (любая ошибка → log + возврат summary-dict).
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any, Literal
+
+import sentry_sdk
+
+from app.api.v1.admin_scrape import compute_freshness
+from app.core.db import SessionLocal
+from app.workers.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+# Статусы из compute_freshness, которые считаем поводом для алёрта.
+_ALERTABLE = {"stale", "failed"}
+
+
+def _format_age(source: dict[str, Any]) -> str:
+    """Человекочитаемый возраст источника для тела сообщения."""
+    age = source.get("age_days")
+    if age is None:
+        return "нет успешных прогонов"
+    return f"{age}d"
+
+
+@celery_app.task(name="tasks.scrape_freshness_check.scrape_freshness_check")
+def scrape_freshness_check() -> dict[str, Any]:
+    """Проверяет свежесть источников и шлёт GlitchTip-алёрт при stale/failed.
+
+    Returns:
+        Summary-dict для observability/тестов:
+        {'checked': N, 'alertable': M, 'stale': S, 'failed': F,
+         'overall_status': str, 'alerted': bool, 'level': str|None}
+        При внутренней ошибке: {'error': '<msg>'} — исключение НЕ пробрасывается,
+        чтобы не сорвать beat-loop.
+    """
+    db = SessionLocal()
+    try:
+        payload = compute_freshness(db)
+    except Exception as exc:  # любая ошибка агрегации не должна валить beat
+        logger.exception("scrape_freshness_check: compute_freshness упал: %s", exc)
+        return {"error": str(exc)}
+    finally:
+        db.close()
+
+    sources: list[dict[str, Any]] = payload.get("sources", [])
+    overall = payload.get("overall_status", "unknown")
+
+    alertable = [s for s in sources if s.get("status") in _ALERTABLE]
+    stale = [s for s in alertable if s.get("status") == "stale"]
+    failed = [s for s in alertable if s.get("status") == "failed"]
+
+    summary: dict[str, Any] = {
+        "checked": len(sources),
+        "alertable": len(alertable),
+        "stale": len(stale),
+        "failed": len(failed),
+        "overall_status": overall,
+        "alerted": False,
+        "level": None,
+    }
+
+    if not alertable:
+        logger.info(
+            "scrape_freshness_check: все источники свежие (checked=%d, overall=%s)",
+            len(sources),
+            overall,
+        )
+        return summary
+
+    # Severity: критичный источник упал полностью → error, иначе warning.
+    critical_failed = [s for s in failed if s.get("critical")]
+    level: Literal["error", "warning"] = "error" if critical_failed else "warning"
+
+    parts = [f"{s['source']} [{s['status']}, {_format_age(s)}]" for s in alertable]
+    message = (
+        f"Data freshness alert (overall={overall}): "
+        f"{len(failed)} failed, {len(stale)} stale — " + ", ".join(parts)
+    )
+
+    # capture_message — no-op, если SDK не инициализирован (нет glitchtip_dsn);
+    # поэтому всегда дублируем в лог (WARNING/ERROR по severity).
+    sentry_sdk.capture_message(message, level=level)
+    log = logger.error if level == "error" else logger.warning
+    log("scrape_freshness_check: %s", message)
+
+    summary["alerted"] = True
+    summary["level"] = level
+    return summary

backend/tests/workers/test_scrape_freshness_check.py

@@ -0,0 +1,145 @@
+"""Тесты beat-таски scrape_freshness_check — алёрт в GlitchTip при stale/failed источниках.
+
+Сеть/БД не дёргаются: мокаем SessionLocal + compute_freshness (единый источник истины) +
+sentry_sdk.capture_message. Проверяем: severity-маппинг (error при failed критичного,
+warning при stale), сообщение называет проблемные источники, all-ok → нет алёрта, ошибка
+агрегации → graceful (исключение не пробрасывается).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+
+def _source(
+    name: str,
+    status: str,
+    *,
+    critical: bool = False,
+    age_days: float | None = 1.0,
+) -> dict[str, Any]:
+    """Минимальный source-dict в форме compute_freshness (нужные таске поля)."""
+    return {"source": name, "status": status, "critical": critical, "age_days": age_days}
+
+
+def _payload(sources: list[dict[str, Any]], overall: str) -> dict[str, Any]:
+    return {"generated_at": "2026-06-25T09:00:00+00:00", "overall_status": overall,
+            "sources": sources}
+
+
+def _run_with(payload_or_exc: Any) -> tuple[dict[str, Any], MagicMock]:
+    """Запускает таску с замоканными SessionLocal/compute_freshness/capture_message.
+
+    payload_or_exc: dict → возвращается из compute_freshness; Exception → бросается.
+    Возвращает (summary, capture_message_mock).
+    """
+    from app.workers.tasks import scrape_freshness_check as task_mod
+
+    db = MagicMock()
+    if isinstance(payload_or_exc, Exception):
+        compute = MagicMock(side_effect=payload_or_exc)
+    else:
+        compute = MagicMock(return_value=payload_or_exc)
+
+    with (
+        patch.object(task_mod, "SessionLocal", return_value=db),
+        patch.object(task_mod, "compute_freshness", compute),
+        patch.object(task_mod.sentry_sdk, "capture_message") as cap,
+    ):
+        summary = task_mod.scrape_freshness_check.run()
+
+    # Сессию всегда закрываем (try/finally).
+    db.close.assert_called_once()
+    return summary, cap
+
+
+def test_stale_and_failed_emit_single_alert() -> None:
+    """Один stale + один failed (некритичный) → ровно один capture_message, level=warning,
+    сообщение называет оба источника."""
+    payload = _payload(
+        [
+            _source("kn", "ok", critical=True),
+            _source("objective", "stale", critical=True, age_days=10.0),
+            _source("nspd", "failed", critical=False, age_days=None),
+        ],
+        overall="stale",
+    )
+    summary, cap = _run_with(payload)
+
+    cap.assert_called_once()
+    args, kwargs = cap.call_args
+    message = args[0]
+    assert kwargs["level"] == "warning"  # упал НЕкритичный → warning
+    assert "objective" in message
+    assert "nspd" in message
+    assert "kn" not in message  # ok-источник не упоминается
+
+    assert summary["alerted"] is True
+    assert summary["level"] == "warning"
+    assert summary["stale"] == 1
+    assert summary["failed"] == 1
+    assert summary["alertable"] == 2
+    assert summary["checked"] == 3
+
+
+def test_critical_failed_escalates_to_error() -> None:
+    """Критичный источник полностью упал (failed) → level=error."""
+    payload = _payload(
+        [
+            _source("kn", "failed", critical=True, age_days=None),
+            _source("nspd", "stale", critical=False, age_days=12.0),
+        ],
+        overall="failed",
+    )
+    summary, cap = _run_with(payload)
+
+    cap.assert_called_once()
+    _, kwargs = cap.call_args
+    assert kwargs["level"] == "error"
+    assert summary["level"] == "error"
+    assert summary["alerted"] is True
+
+
+def test_all_ok_emits_no_alert() -> None:
+    """Все источники ok → capture_message НЕ вызывается."""
+    payload = _payload(
+        [
+            _source("kn", "ok", critical=True),
+            _source("objective", "ok", critical=True),
+            _source("nspd", "ok"),
+        ],
+        overall="ok",
+    )
+    summary, cap = _run_with(payload)
+
+    cap.assert_not_called()
+    assert summary["alerted"] is False
+    assert summary["level"] is None
+    assert summary["alertable"] == 0
+    assert summary["overall_status"] == "ok"
+
+
+def test_compute_freshness_raising_is_graceful() -> None:
+    """compute_freshness бросает → таска логирует и возвращает {'error': ...}, не падает."""
+    summary, cap = _run_with(RuntimeError("db boom"))
+
+    cap.assert_not_called()
+    assert summary == {"error": "db boom"}
+
+
+def test_endpoint_delegates_to_compute_freshness() -> None:
+    """Рефактор: HTTP-эндпоинт scrape_freshness просто возвращает compute_freshness(db).
+
+    Без живой БД: monkeypatch compute_freshness → проверяем, что handler отдаёт его
+    результат байт-в-байт и прокидывает ту же сессию.
+    """
+    from app.api.v1 import admin_scrape
+
+    sentinel = {"generated_at": "x", "overall_status": "ok", "sources": []}
+    db = MagicMock()
+    with patch.object(admin_scrape, "compute_freshness", return_value=sentinel) as comp:
+        result = admin_scrape.scrape_freshness(db)
+
+    comp.assert_called_once_with(db)
+    assert result is sentinel

frontend/src/lib/api-types.ts

@@ -1560,18 +1560,11 @@ export interface paths {
     };
     /**
      * Scrape Freshness
-     * @description Свежесть данных по каждому источнику (для dashboard + alerting).
+     * @description GET-обёртка над compute_freshness — только читает, не алертит.
      *
-     *     Для каждого источника один проход по его run-таблице агрегирует:
+     *     Вся агрегация живёт в compute_freshness (один источник истины); эндпоинт
-     *       - last_success_at  — max(COALESCE(finished_at, started_at)) WHERE status='done'
+     *     лишь прокидывает сессию и возвращает результат без изменений формы payload.
-     *       - last_attempt_at  — max(started_at) (или created_at fallback) по всем прогонам
+     *     Sentry-алёрты по критичным источникам шлёт beat-таска scrape_freshness_check.
-     *       - last_status      — status последнего прогона
-     *       - objects_updated_24h / _7d — сумма run-счётчика по done-прогонам в окне
-     *       - age_days         — возраст last_success в днях (NULL если успехов не было)
-     *       - freshness        — fresh / stale / critical / unknown (по порогам источника)
-     *       - status           — ok / stale / failed (для UI-светофора)
-     *
-     *     `overall_status` агрегирует худший статус по критичным источникам.
      */
     get: operations["scrape_freshness_api_v1_admin_scrape_freshness_get"];
     put?: never;