fix(tradein): persist scrape_runs.total_seen/new_count from sweep counters (#1926) #1928
2 changed files with 175 additions and 10 deletions
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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) ─────────────────────────────────────────────
|
||||
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue