diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index bb094440..192b3286 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -1264,24 +1264,32 @@ class FreshnessSource(BaseModel): source: str # стабильный ключ (kn / nspd / objective / nspd_geo / cadastre) label: str # человекочитаемое имя для UI - table: str # run-таблица + table: str # run-таблица ЛИБО data-таблица (если задан timestamp_col) # Колонка-счётчик "сколько объектов обновил done-прогон". Агрегируется как # COALESCE(SUM() FILTER (...), 0) — FILTER применяется к SUM, не к COALESCE. + # В timestamp-режиме (timestamp_col задан) work_col игнорируется. work_col: str started_col: str = "started_at" # Колонка-fallback для last_attempt когда started_at NULL (queued-but-not-run). attempt_fallback_col: str | None = None + # Data-table режим: если задан, источник — это ПЛОСКАЯ data-таблица (не run-ledger, + # без status/started/finished). Свежесть = MAX(timestamp_col); upd_24h/_7d = COUNT(*) + # по строкам в окне. work_col/started_col/attempt_fallback_col при этом игнорируются. + timestamp_col: str | None = None fresh_days: float = 7.0 # < fresh_days → fresh stale_days: float = 30.0 # < stale_days → stale, иначе critical critical: bool = False # критичный источник (для Sentry-алёртов в beat-таске) -# Реестр источников. Все 5 таблиц проверены на проде (information_schema) — -# статусы done присутствуют у kn/nspd_geo/objective/cadastre; nspd только failed. +# Реестр источников. Run-ledger таблицы (kn/objective/nspd_geo/cadastre) проверены на +# проде (information_schema) — статусы done присутствуют. nspd — data-table режим: +# смотрит nspd_quarter_dumps.fetched_at_utc (живой harvest harvest_stale_quarters + +# write-back /analyze-резолвера #1891), а НЕ defunct ledger nspd_scrape_runs (3 WAF- +# забаненных manual-прогона 2026-04-30, никогда не шедулился → ложный "no successful runs"). # Пороги: # kn — beat ~ежедневно (scrape_kn cron) → fresh<2d, stale<5d # objective — beat ~еженедельно (objective_sync) + критичный → fresh<7d, stale<14d -# nspd — harvest_stale_quarters (пн) → fresh<8d, stale<30d +# nspd — data-table: harvest_stale_quarters (пн) + lazy-refresh → fresh<14d, stale<30d # nspd_geo — ручной bulk geo-job → fresh<7d, stale<30d # cadastre — ручной cadastre-job → fresh<14d, stale<45d _FRESHNESS_SOURCES: list[FreshnessSource] = [ @@ -1309,10 +1317,17 @@ _FRESHNESS_SOURCES: list[FreshnessSource] = [ ), FreshnessSource( source="nspd", - label="NSPD кварталы/здания (harvest)", - table="nspd_scrape_runs", - work_col="quarters_ok", - fresh_days=8.0, + label="NSPD quarter dumps (harvest+resolver)", + # Data-table режим: nspd_quarter_dumps — это контент-таблица квартальных + # дампов (NO status/started/finished). Живой harvest пишет fetched_at_utc; + # defunct nspd_scrape_runs (manual WAF-ban 2026-04-30) больше НЕ источник истины. + table="nspd_quarter_dumps", + timestamp_col="fetched_at_utc", + # В timestamp-режиме не используется — оставляем валидное имя колонки. + work_col="total_features", + # Медленный кадастровый + lazy-refresh источник: дампы освежаются по мере + # устаревания кварталов / on-demand из /analyze. Текущий возраст ~4d → fresh. + fresh_days=14.0, stale_days=30.0, ), FreshnessSource( @@ -1354,7 +1369,7 @@ def compute_freshness(db: Session) -> dict[str, Any]: scrape_freshness_check). Принимает открытую сессию, НЕ коммитит/закрывает её — жизненным циклом сессии управляет вызывающий (FastAPI-зависимость или таска). - Для каждого источника один проход по его run-таблице агрегирует: + Для run-ledger источников один проход по run-таблице агрегирует: - last_success_at — max(COALESCE(finished_at, started_at)) WHERE status='done' - last_attempt_at — max(started_at) (или created_at fallback) по всем прогонам - last_status — status последнего прогона @@ -1363,6 +1378,13 @@ def compute_freshness(db: Session) -> dict[str, Any]: - freshness — fresh / stale / critical / unknown (по порогам источника) - status — ok / stale / failed (для UI-светофора) + Для data-table источников (src.timestamp_col задан, напр. nspd → nspd_quarter_dumps) + нет run-ledger семантики (status/started/finished отсутствуют), поэтому: + - last_success_at = last_attempt_at = MAX(timestamp_col) + - objects_updated_24h / _7d = COUNT(*) строк, обновлённых в окне + - last_status = NULL (косметика только для run-ledger'ов) + Остальной downstream (age_days / _classify_freshness / status-маппинг) — общий. + `overall_status` агрегирует худший статус по критичным источникам. """ sources_out: list[dict[str, Any]] = [] @@ -1370,54 +1392,83 @@ def compute_freshness(db: Session) -> dict[str, Any]: severity = {"ok": 0, "stale": 1, "failed": 2} for src in _FRESHNESS_SOURCES: - started = src.started_col - attempt_expr = ( - f"MAX(COALESCE({started}, {src.attempt_fallback_col}))" - if src.attempt_fallback_col - else f"MAX({started})" - ) - # last_status — отдельным скаляром: status последнего по времени прогона. - order_col = ( - f"COALESCE({started}, {src.attempt_fallback_col})" - if src.attempt_fallback_col - else started - ) - # work_col / table / column-имена — из доверенного серверного реестра - # (НЕ из пользовательского ввода), поэтому f-string в SQL здесь безопасен; + # table / column-имена — из доверенного серверного реестра (НЕ из + # пользовательского ввода), поэтому f-string в SQL здесь безопасен; # все временные границы передаются параметрами (:d1/:d7). - # FILTER применяется к SUM(...), а COALESCE(...,0) оборачивает результат — - # COALESCE(...) FILTER(...) — синтаксическая ошибка в PostgreSQL. - row = ( - db.execute( - text( - f""" - SELECT - MAX(COALESCE(finished_at, {started})) - FILTER (WHERE status = 'done') AS last_success_at, - {attempt_expr} AS last_attempt_at, - COALESCE(SUM({src.work_col}) FILTER ( - WHERE status = 'done' - AND COALESCE(finished_at, {started}) - > NOW() - CAST(:d1 AS interval) - ), 0) AS upd_24h, - COALESCE(SUM({src.work_col}) FILTER ( - WHERE status = 'done' - AND COALESCE(finished_at, {started}) - > NOW() - CAST(:d7 AS interval) - ), 0) AS upd_7d, - ( - SELECT status FROM {src.table} - ORDER BY {order_col} DESC NULLS LAST - LIMIT 1 - ) AS last_status - FROM {src.table} - """ - ), - {"d1": "24 hours", "d7": "7 days"}, + if src.timestamp_col is not None: + # Data-table режим: плоская контент-таблица без run-ledger семантики + # (нет status/started/finished). Свежесть = MAX(timestamp_col), + # upd_24h/_7d = COUNT(*) строк, обновлённых в окне. last_status=NULL + # (косметика только для run-ledger'ов). + ts = src.timestamp_col + row = ( + db.execute( + text( + f""" + SELECT + MAX({ts}) AS last_success_at, + MAX({ts}) AS last_attempt_at, + COALESCE(COUNT(*) FILTER ( + WHERE {ts} > NOW() - CAST(:d1 AS interval) + ), 0) AS upd_24h, + COALESCE(COUNT(*) FILTER ( + WHERE {ts} > NOW() - CAST(:d7 AS interval) + ), 0) AS upd_7d, + NULL AS last_status + FROM {src.table} + """ + ), + {"d1": "24 hours", "d7": "7 days"}, + ) + .mappings() + .first() + ) + else: + started = src.started_col + attempt_expr = ( + f"MAX(COALESCE({started}, {src.attempt_fallback_col}))" + if src.attempt_fallback_col + else f"MAX({started})" + ) + # last_status — отдельным скаляром: status последнего по времени прогона. + order_col = ( + f"COALESCE({started}, {src.attempt_fallback_col})" + if src.attempt_fallback_col + else started + ) + # FILTER применяется к SUM(...), а COALESCE(...,0) оборачивает результат — + # COALESCE(...) FILTER(...) — синтаксическая ошибка в PostgreSQL. + row = ( + db.execute( + text( + f""" + SELECT + MAX(COALESCE(finished_at, {started})) + FILTER (WHERE status = 'done') AS last_success_at, + {attempt_expr} AS last_attempt_at, + COALESCE(SUM({src.work_col}) FILTER ( + WHERE status = 'done' + AND COALESCE(finished_at, {started}) + > NOW() - CAST(:d1 AS interval) + ), 0) AS upd_24h, + COALESCE(SUM({src.work_col}) FILTER ( + WHERE status = 'done' + AND COALESCE(finished_at, {started}) + > NOW() - CAST(:d7 AS interval) + ), 0) AS upd_7d, + ( + SELECT status FROM {src.table} + ORDER BY {order_col} DESC NULLS LAST + LIMIT 1 + ) AS last_status + FROM {src.table} + """ + ), + {"d1": "24 hours", "d7": "7 days"}, + ) + .mappings() + .first() ) - .mappings() - .first() - ) last_success = row["last_success_at"] if row else None last_attempt = row["last_attempt_at"] if row else None diff --git a/backend/tests/api/v1/test_admin_scrape_freshness.py b/backend/tests/api/v1/test_admin_scrape_freshness.py index b62f07f1..442c9216 100644 --- a/backend/tests/api/v1/test_admin_scrape_freshness.py +++ b/backend/tests/api/v1/test_admin_scrape_freshness.py @@ -58,8 +58,8 @@ def _override_db(db: MagicMock) -> None: def test_freshness_mixed_statuses() -> None: - """kn свежий (ok), objective stale, nspd без успехов (failed), nspd_geo critical, - cadastre fresh. overall_status = failed (по критичному nspd_geo→failed).""" + """kn свежий (ok), objective stale, nspd (data-table) свежий (ok), nspd_geo critical, + cadastre fresh. overall_status = stale (по критичному objective→stale).""" rows = { "kn_scrape_runs": { "last_success_at": _ago(1.0), # < fresh_days=2 → fresh → ok @@ -75,12 +75,14 @@ def test_freshness_mixed_statuses() -> None: "upd_7d": 325000, "last_status": "done", }, - "nspd_scrape_runs": { - "last_success_at": None, # нет успехов → failed - "last_attempt_at": _ago(20.0), - "upd_24h": 0, - "upd_7d": 0, - "last_status": "failed", + # nspd теперь data-table режим (nspd_quarter_dumps.fetched_at_utc): + # last_success_at = MAX(timestamp_col), last_status = NULL (нет run-ledger). + "nspd_quarter_dumps": { + "last_success_at": _ago(4.0), # < fresh_days=14 → fresh → ok + "last_attempt_at": _ago(4.0), + "upd_24h": 5, + "upd_7d": 56, + "last_status": None, }, "nspd_geo_jobs": { "last_success_at": _ago(40.0), # >= stale_days=30 → critical → failed @@ -117,9 +119,11 @@ def test_freshness_mixed_statuses() -> None: assert by_src["objective"]["status"] == "stale" assert by_src["objective"]["freshness"] == "stale" - assert by_src["nspd"]["status"] == "failed" - assert by_src["nspd"]["last_success_at"] is None - assert by_src["nspd"]["age_days"] is None + # nspd — data-table режим: свежий dump → ok, last_status=NULL не ломает маппинг. + assert by_src["nspd"]["status"] == "ok" + assert by_src["nspd"]["freshness"] == "fresh" + assert by_src["nspd"]["last_status"] is None + assert by_src["nspd"]["objects_updated_7d"] == 56 assert by_src["nspd_geo"]["status"] == "failed" assert by_src["nspd_geo"]["freshness"] == "critical" @@ -132,6 +136,84 @@ def test_freshness_mixed_statuses() -> None: assert body["overall_status"] == "stale" +def _healthy_run_rows_except_nspd(nspd_row: dict[str, Any]) -> dict[str, dict[str, Any]]: + """Все run-ledger источники свежие, nspd — отдельной data-table строкой. + + Используется для изоляции поведения nspd-источника в data-table режиме, не + затрагивая остальные источники. + """ + fresh_run = { + "last_success_at": _ago(0.5), + "last_attempt_at": _ago(0.5), + "upd_24h": 10, + "upd_7d": 100, + "last_status": "done", + } + rows: dict[str, dict[str, Any]] = {} + for s in admin_scrape._FRESHNESS_SOURCES: + rows[s.table] = nspd_row if s.source == "nspd" else dict(fresh_run) + return rows + + +def test_nspd_uses_timestamp_col_mode_fresh() -> None: + """nspd теперь читает data-таблицу nspd_quarter_dumps по timestamp_col=fetched_at_utc. + + Свежий dump (~4d < fresh_days=14) + last_status=NULL → freshness=fresh, status=ok + (НЕ failed). Регрессия на repoint defunct nspd_scrape_runs → живой harvest-dump. + """ + nspd = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "nspd") + # Подтверждаем что источник в data-table режиме и указывает на живую таблицу. + assert nspd.timestamp_col == "fetched_at_utc" + assert nspd.table == "nspd_quarter_dumps" + assert nspd.critical is False # остаётся некритичным + + rows = _healthy_run_rows_except_nspd( + { + "last_success_at": _ago(4.0), # < fresh_days=14 → fresh + "last_attempt_at": _ago(4.0), + "upd_24h": 5, + "upd_7d": 56, + "last_status": None, # data-table → NULL, маппинг не должен падать + } + ) + _override_db(_make_db(rows)) + try: + resp = TestClient(app).get(ENDPOINT) + finally: + app.dependency_overrides.clear() + + assert resp.status_code == 200, resp.text + by_src = {s["source"]: s for s in resp.json()["sources"]} + assert by_src["nspd"]["freshness"] == "fresh" + assert by_src["nspd"]["status"] == "ok" + assert by_src["nspd"]["last_status"] is None + assert by_src["nspd"]["objects_updated_7d"] == 56 + assert by_src["nspd"]["age_days"] is not None and by_src["nspd"]["age_days"] < 14.0 + + +def test_nspd_timestamp_col_stale_dump_is_critical() -> None: + """Старый dump (40d >= stale_days=30) в data-table режиме → critical → status=failed.""" + rows = _healthy_run_rows_except_nspd( + { + "last_success_at": _ago(40.0), # >= stale_days=30 → critical + "last_attempt_at": _ago(40.0), + "upd_24h": 0, + "upd_7d": 0, + "last_status": None, + } + ) + _override_db(_make_db(rows)) + try: + resp = TestClient(app).get(ENDPOINT) + finally: + app.dependency_overrides.clear() + + assert resp.status_code == 200, resp.text + by_src = {s["source"]: s for s in resp.json()["sources"]} + assert by_src["nspd"]["freshness"] == "critical" + assert by_src["nspd"]["status"] == "failed" + + def test_freshness_all_healthy() -> None: """Все источники свежие → overall_status=ok.""" fresh_row = {