From bd1adc1f59d4e665cfe3d7debf023d29f2fc5f24 Mon Sep 17 00:00:00 2001 From: Light1YT Date: Sat, 27 Jun 2026 11:52:29 +0500 Subject: [PATCH] feat(freshness): zero-output check + kn_flats source (#1945) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The data-freshness monitor classified by run RECENCY only, so the domrf_kn FLATS loader running status=done but extracting 0 flats for ~5 weeks went undetected — and the kn source watched objects_count (healthy ~1548), not flats_count (the broken =0 metric). Add an opt-in zero-output check: an otherwise-fresh run-ledger source (recent success, would-be fresh by age) that produced 0 work-rows in the 7d window is downgraded to status="failed" (so scrape_freshness_check alerts), with an additive "reason". Guards: alert_on_zero_output flag, run-ledger only (timestamp_col is None), status=="ok" (age-stale/failed already covered), and upd_7d==0 (SUM of the source's own work_col over done-runs). Registry: new kn_flats source (kn_scrape_runs, work_col=flats_count, critical, flag on) — watches the column that was broken; existing kn (objects_count) unchanged. Flag also enabled on objective (rows_lots, critical). nspd/nspd_geo/ cadastre left unflagged (legitimate-0 / data-table). JSON additive only (new nullable "reason" key; endpoint is dict[str,Any], no frontend consumer / no codegen needed). 4 new tests (downgrade, no-false- positive, age-precedence, registry). code-reviewer APPROVE. Would have caught #1945 within ~8-14d instead of 5 weeks. --- backend/app/api/v1/admin_scrape.py | 69 ++++++- .../api/v1/test_admin_scrape_freshness.py | 168 +++++++++++++++++- 2 files changed, 229 insertions(+), 8 deletions(-) diff --git a/backend/app/api/v1/admin_scrape.py b/backend/app/api/v1/admin_scrape.py index 192b3286..6e56b194 100644 --- a/backend/app/api/v1/admin_scrape.py +++ b/backend/app/api/v1/admin_scrape.py @@ -1279,6 +1279,13 @@ class FreshnessSource(BaseModel): fresh_days: float = 7.0 # < fresh_days → fresh stale_days: float = 30.0 # < stale_days → stale, иначе critical critical: bool = False # критичный источник (для Sentry-алёртов в beat-таске) + # True → источник, который прогнался успешно И недавно (был бы fresh по возрасту), + # НО произвёл 0 work-строк в окне (SUM(work_col) FILTER(status='done') == 0), + # ДАУНГРЕЙДИТСЯ до "failed". Ловит "лоадер-прогнался-но-ничего-не-извлёк" + # (напр. WAF-break / тихий пустой extract: status=done, но flats_count=0). По + # умолчанию OFF — включаем только там, где 0 строк = поломка (не legitimate-0, + # как nspd/nspd_geo/cadastre). Применимо только к run-ledger (не data-table) режиму. + alert_on_zero_output: bool = False # Реестр источников. Run-ledger таблицы (kn/objective/nspd_geo/cadastre) проверены на @@ -1287,7 +1294,10 @@ class FreshnessSource(BaseModel): # 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 +# kn — beat ~еженедельно (scrape_kn cron, Mon) → fresh<8d, stale<14d +# kn_flats — та же run-таблица, work_col=flats_count + alert_on_zero_output (issue +# #1945: kn-run status=done но flats_count=0 ~5 недель — монитор молчал, +# т.к. kn смотрит objects_count ~1548, не flats_count) → fresh<8d, stale<14d # objective — beat ~еженедельно (objective_sync) + критичный → fresh<7d, stale<14d # nspd — data-table: harvest_stale_quarters (пн) + lazy-refresh → fresh<14d, stale<30d # nspd_geo — ручной bulk geo-job → fresh<7d, stale<30d @@ -1306,6 +1316,22 @@ _FRESHNESS_SOURCES: list[FreshnessSource] = [ stale_days=14.0, critical=True, ), + FreshnessSource( + # kn-run пишет ДВА счётчика: objects_count (ЖК, ~1548 — выше) и flats_count + # (квартиры). Источник истины для "лоадер прогнался, но ничего не извлёк" + # (issue #1945: domrf_kn FLATS-лоадер status=done, но flats_count=0 ~5 недель, + # монитор молчал — objects_count оставался здоровым). Та же run-таблица и пороги, + # что у kn, но work_col=flats_count + alert_on_zero_output → 0 квартир за неделю + # done-прогонов даунгрейдит otherwise-fresh источник до failed. + source="kn_flats", + label="DOM.РФ kn-API (квартиры)", + table="kn_scrape_runs", + work_col="flats_count", + fresh_days=8.0, + stale_days=14.0, + critical=True, + alert_on_zero_output=True, + ), FreshnessSource( source="objective", label="Объектив (лоты/цены)", @@ -1314,6 +1340,9 @@ _FRESHNESS_SOURCES: list[FreshnessSource] = [ fresh_days=7.0, stale_days=14.0, critical=True, + # Объектив-sync, прогнавшийся с 0 лотов за неделю — тоже поломка (тихий + # пустой extract), а не легитимный 0. Даунгрейд otherwise-fresh до failed. + alert_on_zero_output=True, ), FreshnessSource( source="nspd", @@ -1377,6 +1406,13 @@ def compute_freshness(db: Session) -> dict[str, Any]: - age_days — возраст last_success в днях (NULL если успехов не было) - freshness — fresh / stale / critical / unknown (по порогам источника) - status — ok / stale / failed (для UI-светофора) + - reason — пояснение к zero-output-даунгрейду (иначе None) + + Zero-output (issue #1945): для run-ledger источников с alert_on_zero_output=True + источник, который иначе был бы fresh→ok (недавний успешный прогон), но произвёл + 0 work-строк за 7d окно (objects_updated_7d == 0), ДАУНГРЕЙДИТСЯ до failed — ловит + "лоадер-прогнался-но-ничего-не-извлёк" (kn flats_count=0 ~5 недель при status=done). + Только когда возраст НЕ покрывает поломку (status=='ok'), чтобы не двойнить. Для data-table источников (src.timestamp_col задан, напр. nspd → nspd_quarter_dumps) нет run-ledger семантики (status/started/finished отсутствуют), поэтому: @@ -1481,6 +1517,9 @@ def compute_freshness(db: Session) -> dict[str, Any]: freshness = _classify_freshness(age_days, src.fresh_days, src.stale_days) + upd_24h = int(row["upd_24h"] or 0) if row else 0 + upd_7d = int(row["upd_7d"] or 0) if row else 0 + # status (UI-светофор): нет успехов вообще → failed; иначе мапим freshness. if last_success is None: status = "failed" @@ -1491,6 +1530,27 @@ def compute_freshness(db: Session) -> dict[str, Any]: else: # critical status = "failed" + # Zero-output downgrade (issue #1945): источник прогнался успешно И недавно + # (был бы fresh→ok по возрасту), НО за окно done-прогонов произвёл 0 work-строк + # (objects_updated_7d == SUM(work_col) FILTER(status='done') в 7d == 0) → + # ДАУНГРЕЙД до "failed", чтобы beat-таска (алёртит на stale/failed) выстрелила. + # Только для run-ledger (timestamp_col is None) с alert_on_zero_output=True и + # ТОЛЬКО когда источник иначе был бы fresh/ok — по возрасту stale/failed уже + # покрыт, не двойним. last_success is None уже отсеян (status=='ok' гарантирует + # недавний успех), поэтому это ДОПОЛНИТЕЛЬНЫЙ способ завалить otherwise-fresh. + reason: str | None = None + if ( + src.alert_on_zero_output + and src.timestamp_col is None + and status == "ok" + and upd_7d == 0 + ): + status = "failed" + reason = ( + f"zero-output: прогон успешен (age={age_days}d), но {src.work_col}=0 " + f"за 7d done-прогонов — лоадер прогнался, но ничего не извлёк (#1945)" + ) + if src.critical and severity[status] > severity[worst_critical]: worst_critical = status @@ -1502,12 +1562,15 @@ def compute_freshness(db: Session) -> dict[str, Any]: "last_success_at": last_success.isoformat() if last_success else None, "last_attempt_at": last_attempt.isoformat() if last_attempt else None, "last_status": last_status, - "objects_updated_24h": int(row["upd_24h"] or 0) if row else 0, - "objects_updated_7d": int(row["upd_7d"] or 0) if row else 0, + "objects_updated_24h": upd_24h, + "objects_updated_7d": upd_7d, "age_days": age_days, "thresholds": {"fresh_days": src.fresh_days, "stale_days": src.stale_days}, "freshness": freshness, "status": status, + # Доп. контекст при zero-output-даунгрейде (None в обычном случае). + # Additive — не ломает существующий UI/JSON-контракт. + "reason": reason, } ) diff --git a/backend/tests/api/v1/test_admin_scrape_freshness.py b/backend/tests/api/v1/test_admin_scrape_freshness.py index 442c9216..0e787e85 100644 --- a/backend/tests/api/v1/test_admin_scrape_freshness.py +++ b/backend/tests/api/v1/test_admin_scrape_freshness.py @@ -1,10 +1,11 @@ """Тесты для GET /api/v1/admin/scrape/freshness (#73 B5-3). -Эндпоинт читает 5 run-таблиц (kn / objective / nspd / nspd_geo / cadastre) и -для каждой считает свежесть. Здесь DB замокан: на каждый per-source SELECT +Эндпоинт читает 6 источников по 5 run-/data-таблицам (kn / kn_flats / objective / +nspd / nspd_geo / cadastre — kn и kn_flats делят kn_scrape_runs, разные work_col) +и для каждого считает свежесть. Здесь DB замокан: на каждый per-source SELECT возвращаем заранее заданную строку, чтобы проверить классификацию -fresh/stale/critical, маппинг status (ok/stale/failed) и агрегат overall_status -по критичным источникам. +fresh/stale/critical, маппинг status (ok/stale/failed), zero-output downgrade +(#1945) и агрегат overall_status по критичным источникам. """ from __future__ import annotations @@ -108,7 +109,8 @@ def test_freshness_mixed_statuses() -> None: assert resp.status_code == 200, resp.text body = resp.json() assert "generated_at" in body - assert len(body["sources"]) == 5 + # kn / kn_flats / objective / nspd / nspd_geo / cadastre. + assert len(body["sources"]) == len(admin_scrape._FRESHNESS_SOURCES) by_src = {s["source"]: s for s in body["sources"]} assert by_src["kn"]["status"] == "ok" @@ -279,6 +281,162 @@ def test_classify_freshness_boundaries() -> None: assert f(100.0, 2.0, 5.0) == "critical" +def test_zero_output_downgrades_otherwise_fresh_source() -> None: + """issue #1945: kn_flats прогнался успешно И недавно (был бы fresh→ok по возрасту), + но flats_count=0 за 7d done-прогонов → DOWNGRADE до failed (alertable). + + Та же run-таблица kn_scrape_runs у kn (alert_on_zero_output=False) и kn_flats + (True). Возвращаем строку с upd_7d=0 → kn остаётся ok (нет флага = нет ложного + срабатывания), kn_flats падает в failed (zero-output) с пояснением в reason. + """ + fresh_zero = { + "last_success_at": _ago(1.0), # < fresh_days=8 → был бы fresh + "last_attempt_at": _ago(1.0), + "upd_24h": 0, + "upd_7d": 0, # 0 work-строк за окно done-прогонов → zero-output + "last_status": "done", + } + # Остальные источники свежие+ненулевые, чтобы изолировать поведение kn/kn_flats. + nonzero_fresh = { + "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: + if s.timestamp_col is not None: + rows[s.table] = {**nonzero_fresh, "last_status": None} + elif s.table == "kn_scrape_runs": + rows[s.table] = dict(fresh_zero) + else: + rows[s.table] = dict(nonzero_fresh) + _override_db(_make_db(rows)) + try: + resp = TestClient(app).get(ENDPOINT) + finally: + app.dependency_overrides.clear() + + assert resp.status_code == 200, resp.text + body = resp.json() + by_src = {s["source"]: s for s in body["sources"]} + + # kn_flats: otherwise-fresh, но flats_count=0 → DOWNGRADE до failed (alertable). + assert by_src["kn_flats"]["freshness"] == "fresh" # по возрасту был бы fresh + assert by_src["kn_flats"]["status"] == "failed" # zero-output downgrade + assert by_src["kn_flats"]["objects_updated_7d"] == 0 + assert by_src["kn_flats"]["reason"] is not None + assert "flats_count" in by_src["kn_flats"]["reason"] + + # kn — та же таблица, upd_7d=0, но БЕЗ флага → остаётся ok (no false positive). + assert by_src["kn"]["status"] == "ok" + assert by_src["kn"]["reason"] is None + + # critical kn_flats упал → overall_status=failed. + assert by_src["kn_flats"]["critical"] is True + assert body["overall_status"] == "failed" + + +def test_zero_output_no_false_positive_when_nonzero() -> None: + """alert_on_zero_output=True источник с НЕнулевым выходом → остаётся ok (нет даунгрейда). + + Регрессия: zero-output даунгрейд НЕ должен бить по здоровому источнику, который + прогнался и реально извлёк строки. + """ + nonzero_fresh = { + "last_success_at": _ago(1.0), + "last_attempt_at": _ago(1.0), + "upd_24h": 12, + "upd_7d": 1234, # ненулевой выход + "last_status": "done", + } + rows: dict[str, dict[str, Any]] = {} + for s in admin_scrape._FRESHNESS_SOURCES: + rows[s.table] = ( + {**nonzero_fresh, "last_status": None} + if s.timestamp_col is not None + else dict(nonzero_fresh) + ) + _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"]} + # Все alert_on_zero_output-источники с ненулевым выходом → ok, reason=None. + for s in admin_scrape._FRESHNESS_SOURCES: + if s.alert_on_zero_output: + assert by_src[s.source]["status"] == "ok", s.source + assert by_src[s.source]["reason"] is None, s.source + + +def test_zero_output_not_applied_when_already_stale() -> None: + """Если источник уже stale/failed по возрасту — zero-output НЕ передоунгрейдит + (age уже покрывает), и не двойнит. objective: старый успех (10d → stale) + upd_7d=0 + → остаётся stale (НЕ перепрыгивает в failed через zero-output).""" + base_fresh = { + "last_success_at": _ago(0.5), + "last_attempt_at": _ago(0.5), + "upd_24h": 5, + "upd_7d": 50, + "last_status": "done", + } + rows: dict[str, dict[str, Any]] = {} + for s in admin_scrape._FRESHNESS_SOURCES: + rows[s.table] = ( + {**base_fresh, "last_status": None} + if s.timestamp_col is not None + else dict(base_fresh) + ) + # objective: stale по возрасту (10d: fresh<7, stale<14) + 0 выход. + rows["objective_scrape_runs"] = { + "last_success_at": _ago(10.0), + "last_attempt_at": _ago(10.0), + "upd_24h": 0, + "upd_7d": 0, + "last_status": "done", + } + _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"]} + # zero-output применяется ТОЛЬКО к otherwise-ok; stale остаётся stale, reason=None. + assert by_src["objective"]["status"] == "stale" + assert by_src["objective"]["reason"] is None + + +def test_kn_flats_registry_config() -> None: + """kn_flats: та же run-таблица что у kn, но work_col=flats_count + alert_on_zero_output, + критичный, недельные пороги. kn (objects_count) остаётся без флага (unchanged).""" + kn_flats = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "kn_flats") + assert kn_flats.table == "kn_scrape_runs" + assert kn_flats.work_col == "flats_count" + assert kn_flats.timestamp_col is None # run-ledger режим + assert kn_flats.alert_on_zero_output is True + assert kn_flats.critical is True + assert kn_flats.fresh_days == 8.0 + assert kn_flats.stale_days == 14.0 + + kn = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "kn") + assert kn.work_col == "objects_count" + assert kn.alert_on_zero_output is False # существующий источник unchanged + + objective = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == "objective") + assert objective.alert_on_zero_output is True + + # Legitimate-0 источники НЕ должны иметь флаг (избегаем ложных срабатываний). + for name in ("nspd", "nspd_geo", "cadastre"): + src = next(s for s in admin_scrape._FRESHNESS_SOURCES if s.source == name) + assert src.alert_on_zero_output is False, name + + def test_kn_thresholds_match_weekly_cadence() -> None: """kn шедулится еженедельно (Mon cron) → пороги должны покрывать полную неделю, иначе ложный 'stale' каждую среду-воскресенье. Регрессия на калибровку."""