fix(tradein): persist scrape_runs.total_seen/new_count columns from sweep counters (#1926)

Root: city-sweep counters (yandex/avito/cian/domclick/newbuilding) wrote
lots_fetched/lots_inserted only into scrape_runs.counters jsonb; the dedicated
columns total_seen/new_count (schema 015) were never SET, so admin/observability
reported total_seen=0 despite real rows saved (audit #1871/#1926).

Fix (Option A): single-place extraction in scrape_runs.py via _column_counts()
maps counters lots_fetched->total_seen and lots_inserted->new_count (with
explicit total_seen/new_count keys taking priority). update_heartbeat, mark_done,
mark_failed and mark_banned now SET both columns with COALESCE(CAST(:x AS int),
col) so a missing key preserves the existing value. Covers all sweep sources at
once since every pipeline routes counters through these finalizers.

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)

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) ─────────────────────────────────────────────