From 89d05bbfecf129d30f551200f094a8653ed61e02 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Fri, 26 Jun 2026 18:09:17 +0000 Subject: [PATCH] fix(tradein): persist scrape_runs.total_seen/new_count from sweep counters (#1926) (#1928) --- .../backend/app/services/scrape_runs.py | 94 +++++++++++++++++-- tradein-mvp/backend/tests/test_city_sweep.py | 91 ++++++++++++++++++ 2 files changed, 175 insertions(+), 10 deletions(-) diff --git a/tradein-mvp/backend/app/services/scrape_runs.py b/tradein-mvp/backend/app/services/scrape_runs.py index 62b3ed63..d38ddcc0 100644 --- a/tradein-mvp/backend/app/services/scrape_runs.py +++ b/tradein-mvp/backend/app/services/scrape_runs.py @@ -22,6 +22,36 @@ logger = logging.getLogger(__name__) CONSECUTIVE_FAILURE_ALERT_THRESHOLD = 3 +def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]: + """Извлечь значения для dedicated-колонок total_seen / new_count из jsonb-counters. + + Все sweep-counters (YandexCitySweepCounters / CitySweepCounters / …) пишут число + объявлений в выдаче как ``lots_fetched`` и число впервые вставленных как + ``lots_inserted``. Раньше эти значения попадали ТОЛЬКО в counters jsonb, а + выделенные колонки total_seen/new_count оставались 0 → admin/observability + показывала total_seen=0 при реально сохранённых строках (audit #1871/#1926). + + Приоритет ключей: + - total_seen ← 'total_seen' (если уже есть в counters) иначе 'lots_fetched' + - new_count ← 'new_count' (если уже есть) иначе 'lots_inserted' + + Возвращает (total_seen, new_count); None для ключа, которого нет в counters — + тогда соответствующая колонка не перезаписывается (COALESCE-семантика в UPDATE). + """ + + def _pick(*keys: str) -> int | None: + for key in keys: + val = counters.get(key) + if val is not None: + try: + return int(val) + except (TypeError, ValueError): + return None + return None + + return _pick("total_seen", "lots_fetched"), _pick("new_count", "lots_inserted") + + def _alert_if_consecutive_failures(db: Session, source: str) -> None: """Отправить Sentry alert если последние CONSECUTIVE_FAILURE_ALERT_THRESHOLD завершённых запусков для данного source имеют статус 'failed' или 'banned'. @@ -115,16 +145,30 @@ def create_run(db: Session, *, source: str, params: dict[str, Any]) -> int: def update_heartbeat(db: Session, run_id: int, counters: dict[str, int]) -> None: - """UPDATE heartbeat_at=NOW(), counters=:counters.""" + """UPDATE heartbeat_at=NOW(), counters=:counters + total_seen/new_count колонки. + + total_seen/new_count извлекаются из counters (lots_fetched/lots_inserted) и + пишутся в выделенные колонки, чтобы observability не показывала 0 (audit #1926). + COALESCE: если ключа нет в counters — старое значение колонки сохраняется. + """ + total_seen, new_count = _column_counts(counters) db.execute( text( """ UPDATE scrape_runs - SET heartbeat_at = NOW(), counters = CAST(:counters AS jsonb) + SET heartbeat_at = NOW(), + counters = CAST(:counters AS jsonb), + total_seen = COALESCE(CAST(:total_seen AS int), total_seen), + new_count = COALESCE(CAST(:new_count AS int), new_count) WHERE id = :run_id """ ), - {"run_id": run_id, "counters": json.dumps(counters)}, + { + "run_id": run_id, + "counters": json.dumps(counters), + "total_seen": total_seen, + "new_count": new_count, + }, ) db.commit() @@ -139,18 +183,30 @@ def is_cancelled(db: Session, run_id: int) -> bool: def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None: - """Финализация run: status='done', finished_at=NOW(), counters.""" + """Финализация run: status='done', finished_at=NOW(), counters + total_seen/new_count. + + total_seen/new_count извлекаются из counters (lots_fetched/lots_inserted) и пишутся + в выделенные колонки — иначе admin/observability показывает 0 (audit #1926). + """ + total_seen, new_count = _column_counts(counters) row = db.execute( text( """ UPDATE scrape_runs SET status = 'done', finished_at = NOW(), heartbeat_at = NOW(), - counters = CAST(:counters AS jsonb) + counters = CAST(:counters AS jsonb), + total_seen = COALESCE(CAST(:total_seen AS int), total_seen), + new_count = COALESCE(CAST(:new_count AS int), new_count) WHERE id = :run_id AND status = 'running' RETURNING id """ ), - {"run_id": run_id, "counters": json.dumps(counters)}, + { + "run_id": run_id, + "counters": json.dumps(counters), + "total_seen": total_seen, + "new_count": new_count, + }, ).first() if row is None: logger.warning("mark_done no-op: run_id=%d not in 'running' state", run_id) @@ -167,17 +223,26 @@ def mark_failed(db: Session, run_id: int, error: str, counters: dict[str, int]) db.rollback() # defensive: cascade-safe if prior UPDATE left txn in error state except Exception: pass + total_seen, new_count = _column_counts(counters) row = db.execute( text( """ UPDATE scrape_runs SET status = 'failed', finished_at = NOW(), heartbeat_at = NOW(), - error = :error, counters = CAST(:counters AS jsonb) + error = :error, counters = CAST(:counters AS jsonb), + total_seen = COALESCE(CAST(:total_seen AS int), total_seen), + new_count = COALESCE(CAST(:new_count AS int), new_count) WHERE id = :run_id AND status = 'running' RETURNING id """ ), - {"run_id": run_id, "error": error[:1000], "counters": json.dumps(counters)}, + { + "run_id": run_id, + "error": error[:1000], + "counters": json.dumps(counters), + "total_seen": total_seen, + "new_count": new_count, + }, ).first() if row is None: logger.warning("mark_failed no-op: run_id=%d not in 'running' state", run_id) @@ -199,17 +264,26 @@ def mark_banned(db: Session, run_id: int, error: str, counters: dict[str, int]) db.rollback() # defensive: cascade-safe if prior UPDATE left txn in error state except Exception: pass + total_seen, new_count = _column_counts(counters) row = db.execute( text( """ UPDATE scrape_runs SET status = 'banned', finished_at = NOW(), heartbeat_at = NOW(), - error = :error, counters = CAST(:counters AS jsonb) + error = :error, counters = CAST(:counters AS jsonb), + total_seen = COALESCE(CAST(:total_seen AS int), total_seen), + new_count = COALESCE(CAST(:new_count AS int), new_count) WHERE id = :run_id AND status = 'running' RETURNING id """ ), - {"run_id": run_id, "error": error[:1000], "counters": json.dumps(counters)}, + { + "run_id": run_id, + "error": error[:1000], + "counters": json.dumps(counters), + "total_seen": total_seen, + "new_count": new_count, + }, ).first() if row is None: logger.warning("mark_banned no-op: run_id=%d not in 'running' state", run_id) diff --git a/tradein-mvp/backend/tests/test_city_sweep.py b/tradein-mvp/backend/tests/test_city_sweep.py index 7df1dd9c..8d5774b1 100644 --- a/tradein-mvp/backend/tests/test_city_sweep.py +++ b/tradein-mvp/backend/tests/test_city_sweep.py @@ -132,6 +132,97 @@ def test_scrape_runs_mark_cancelled_returns_false_when_not_running() -> None: assert result is False +# ── total_seen / new_count column population (audit #1926) ────────────────── + + +def _captured_params(mock_db: MagicMock) -> dict: + """Извлечь dict bind-параметров из последнего db.execute(text(...), params).""" + assert mock_db.execute.call_args is not None, "db.execute was not called" + return mock_db.execute.call_args.args[1] + + +def test_column_counts_maps_lots_fetched_inserted() -> None: + """_column_counts: lots_fetched→total_seen, lots_inserted→new_count.""" + from app.services.scrape_runs import _column_counts + + total_seen, new_count = _column_counts( + {"lots_fetched": 150, "lots_inserted": 12, "anchors_done": 5} + ) + assert total_seen == 150 + assert new_count == 12 + + +def test_column_counts_prefers_explicit_total_seen_new_count() -> None: + """Если в counters уже есть total_seen/new_count — они приоритетнее lots_*.""" + from app.services.scrape_runs import _column_counts + + total_seen, new_count = _column_counts( + {"total_seen": 99, "new_count": 7, "lots_fetched": 1, "lots_inserted": 1} + ) + assert total_seen == 99 + assert new_count == 7 + + +def test_column_counts_returns_none_when_absent() -> None: + """Нет ни одного из ключей → None (COALESCE сохранит старое значение колонки).""" + from app.services.scrape_runs import _column_counts + + total_seen, new_count = _column_counts({"anchors_done": 3}) + assert total_seen is None + assert new_count is None + + +def test_mark_done_persists_total_seen_new_count_columns() -> None: + """mark_done пишет total_seen/new_count из lots_fetched/lots_inserted (audit #1926).""" + from app.services.scrape_runs import mark_done + + mock_db = MagicMock() + mock_db.execute.return_value.first.return_value = MagicMock(id=1) # row updated + + mark_done(mock_db, 1, {"lots_fetched": 200, "lots_inserted": 18, "anchors_done": 5}) + + params = _captured_params(mock_db) + assert params["total_seen"] == 200 + assert params["new_count"] == 18 + # SQL must SET the dedicated columns, not only the jsonb blob + sql = str(mock_db.execute.call_args.args[0]) + assert "total_seen" in sql + assert "new_count" in sql + mock_db.commit.assert_called() + + +def test_update_heartbeat_persists_total_seen_new_count_columns() -> None: + """update_heartbeat также обновляет колонки (live progress, не только финал).""" + from app.services.scrape_runs import update_heartbeat + + mock_db = MagicMock() + + update_heartbeat(mock_db, 1, {"lots_fetched": 75, "lots_inserted": 4}) + + params = _captured_params(mock_db) + assert params["total_seen"] == 75 + assert params["new_count"] == 4 + mock_db.commit.assert_called() + + +def test_mark_done_yandex_counters_populate_columns() -> None: + """End-to-end shape: YandexCitySweepCounters.to_dict() → mark_done → columns set.""" + from app.services.scrape_pipeline import YandexCitySweepCounters + from app.services.scrape_runs import mark_done + + counters = YandexCitySweepCounters( + anchors_total=1, anchors_done=1, lots_fetched=540, lots_inserted=42 + ) + mock_db = MagicMock() + mock_db.execute.return_value.first.return_value = MagicMock(id=1) + + mark_done(mock_db, 1, counters.to_dict()) + + params = _captured_params(mock_db) + assert params["total_seen"] == 540 + assert params["new_count"] == 42 + + # ── FastAPI endpoints (offline) ─────────────────────────────────────────────