fix(tradein/scrapers): прогон больше не рапортует «done» поверх провала и нуля #2892
4 changed files with 578 additions and 11 deletions
|
|
@ -143,6 +143,28 @@ def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
|
|||
# unique_fetched — full-load'ы avito/cian/yandex (4 источника, 133 прогона) — раньше
|
||||
# сторож их не видел, хотя у cian_full_load 6 из 38 успешных прогонов
|
||||
# реально дали ноль.
|
||||
# succeeded — yandex_newbuilding_sweep (42 прогона/90д) и newbuilding_enrich
|
||||
# (65 прогонов/90д, единственные два писателя ключа на проде,
|
||||
# проверено 2026-08-15). НЕ 'rows_inserted': тот ключ пишет ЕЩЁ и
|
||||
# rosreestr_dkp_import (67 прогонов/90д) — у него rows_inserted=0 в
|
||||
# 66 из 67 это ЗДОРОВЫЙ ответ догнавшего инкрементального импорта
|
||||
# (rows_fetched=rows_skipped=96974, last_id не двигается неделями),
|
||||
# а не отказ; если бы 'rows_inserted' попал в этот список, сторож
|
||||
# зачитывал бы этот здоровый ноль как измеренный провал и копил бы
|
||||
# практически непрерываемый стрик (rosreestr_dkp_import не
|
||||
# прерывается другим статусом — импорт либо 'done', либо не бежал).
|
||||
# НЕ 'processed' по той же причине с другой стороны: это счётчик
|
||||
# ПОПЫТОК (у newbuilding_enrich processed==attempted==limit даже
|
||||
# когда succeeded меньше — прод-факт 09.08: processed=25 succeeded=14,
|
||||
# 44% отказов замаскировались бы под measured-25) — сторож нулевого
|
||||
# результата на нём молчал бы ровно там, где должен сработать, а на
|
||||
# будущем опустении очереди домов (cian_houses_pending) создал бы
|
||||
# свой вечный ложный zero-стрик. 'succeeded' у yandex_newbuilding_sweep
|
||||
# численно совпадает с 'rows_inserted' на всех 42/42 прод-прогонах —
|
||||
# замена не теряет исходную цель (десять прогонов подряд 26.07-10.08,
|
||||
# все 'done', succeeded=0 rows_inserted=0 failed_resolve=4-5 — раньше
|
||||
# ни total_seen/lots_fetched/unique_fetched не было, и
|
||||
# _run_result_count всегда возвращал None (honest-run-status)).
|
||||
# Сводить сюда счётчики ОСТАЛЬНЫХ задач бессмысленно: на проде 28 источников (2650
|
||||
# прогонов) не имеют общего результатного ключа вовсе — у каждого свой словарь
|
||||
# (deactivated / rows_written / poi_loaded / snapshotted / upserted / listings_matched
|
||||
|
|
@ -150,7 +172,12 @@ def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
|
|||
# трёх мониторов результата нет по смыслу. Ноль у них — часто ЗДОРОВЫЙ ответ
|
||||
# (deactivate_stale_* без протухших объявлений). Поэтому сторож не угадывает их
|
||||
# словарь, а честно признаёт, что мерить нечем — см. _run_result_count.
|
||||
_RESULT_COUNTER_KEYS = ("total_seen", "lots_fetched", "unique_fetched")
|
||||
_RESULT_COUNTER_KEYS = (
|
||||
"total_seen",
|
||||
"lots_fetched",
|
||||
"unique_fetched",
|
||||
"succeeded",
|
||||
)
|
||||
|
||||
|
||||
def _run_result_count(counters: Mapping[str, Any] | None) -> int | None:
|
||||
|
|
@ -282,6 +309,63 @@ def _phase_totally_failed(counters: Mapping[str, Any]) -> str | None:
|
|||
return None
|
||||
|
||||
|
||||
# honest-run-status (2026-08-15): доля отказов, которая обесценивает формально ненулевой
|
||||
# сбор. Прод-факт avito_detail_backfill 15.08: {"attempted":64,"failed":57,"enriched":6,
|
||||
# "blocked":1} — 89% попыток отказали, а mark_backfill_finished всё равно звал mark_done,
|
||||
# потому что "produced != 0" (6 обогащено). Ни _sweep_run_did_nothing (нужны
|
||||
# anchors_total/errors_count, у backfill'ов их нет), ни _phase_totally_failed (нужна пара
|
||||
# "<phase>_attempted"/"<phase>_failed" — здесь голые "attempted"/"failed" без фазового
|
||||
# префикса, `"attempted".endswith("_attempted")` не матчит) эту форму counters не ловят —
|
||||
# обе проверки написаны под СВОИ формы, а не под backfill'овскую.
|
||||
#
|
||||
# Порог 'failed' — половина и больше отказов: сбор для практических целей провалился,
|
||||
# даже если несколько записей всё же обогатились. Порог 'partial' НЕ заведён отдельным
|
||||
# статусом scrape_runs.status — это потребовало бы миграции (DROP+ADD CHECK constraint,
|
||||
# 051_scrape_runs_extend.sql) и обучило бы новому значению ещё 4 места (Literal-фильтр
|
||||
# admin API, хардкод статусов фронта, оба IN-списка сторожей) — тот же класс "оборванной
|
||||
# проводки", из-за которого заведён #2686/ban_kind. Вместо статуса — тот же диагноз, что и
|
||||
# у ban_kind: causa в тексте `error`, терминальный статус один ('failed'). 0.15..0.5 —
|
||||
# та же 'failed', но с другой формулировкой причины ("деградировал", не "провалился"), чтобы
|
||||
# оператор видел разницу читая error, не только status.
|
||||
FAILED_RATIO_FAILED_THRESHOLD = 0.5
|
||||
FAILED_RATIO_DEGRADED_THRESHOLD = 0.15
|
||||
# Минимум попыток, при котором доля вообще что-то значит — иначе 1 отказ из 2 (=0.5)
|
||||
# палит статус на шуме единичного случая. То же рассуждение и то же число, что у
|
||||
# _PHASE_MIN_ATTEMPTS (см. выше).
|
||||
_FAILED_RATIO_MIN_ATTEMPTS = _PHASE_MIN_ATTEMPTS
|
||||
|
||||
|
||||
def _failed_ratio_too_high(counters: Mapping[str, Any]) -> str | None:
|
||||
"""Прогон, у которого доля отказов слишком велика, даже если что-то собрано.
|
||||
|
||||
Возвращает текст причины (для error) либо None. Читает ГОЛЫЕ ключи "attempted"/
|
||||
"failed" (без фазового префикса) — сейчас это словарь только у четырёх
|
||||
detail-backfill'ов (avito/yandex/domclick/newbuilding_enrich), все идут через
|
||||
mark_backfill_finished → mark_done. `attempted < _FAILED_RATIO_MIN_ATTEMPTS` или
|
||||
отсутствие любого из ключей → None (нечем/не о чём судить — счётчики либо не
|
||||
заполнены, либо принадлежат другому источнику со своим словарём).
|
||||
|
||||
Что признак НЕ доказывает: КТО виноват (площадка, наш прокси, наш парсер) — поэтому
|
||||
'failed' без диагноза, как и у #2625/#2700/#2764.
|
||||
"""
|
||||
attempted = _pick_int(counters, "attempted")
|
||||
failed = _pick_int(counters, "failed")
|
||||
if attempted is None or failed is None or attempted < _FAILED_RATIO_MIN_ATTEMPTS:
|
||||
return None
|
||||
ratio = failed / max(attempted, 1)
|
||||
if ratio >= FAILED_RATIO_FAILED_THRESHOLD:
|
||||
verb = "провалился"
|
||||
elif ratio >= FAILED_RATIO_DEGRADED_THRESHOLD:
|
||||
verb = "деградировал"
|
||||
else:
|
||||
return None
|
||||
return (
|
||||
f"failed-ratio-honest-status: сбор {verb} — {failed} из {attempted} попыток "
|
||||
f"отказали (доля {ratio:.0%}); формально ненулевой результат этого не искупает. "
|
||||
f"Причина НЕ установлена — статус 'failed' без диагноза"
|
||||
)
|
||||
|
||||
|
||||
def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
|
||||
"""Извлечь значения для dedicated-колонок total_seen / new_count из jsonb-counters.
|
||||
|
||||
|
|
@ -292,13 +376,24 @@ def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
|
|||
показывала total_seen=0 при реально сохранённых строках (audit #1871/#1926).
|
||||
|
||||
Приоритет ключей:
|
||||
- total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched)
|
||||
- new_count ← 'new_count' (если уже есть) иначе 'lots_inserted'
|
||||
- total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched /
|
||||
succeeded)
|
||||
- new_count ← 'new_count' / 'lots_inserted' / 'saved_inserted' / 'rows_inserted'
|
||||
(первый присутствующий). 'saved_inserted' — full-load'ы (cian/avito/yandex,
|
||||
CianFullLoadCounters и аналоги в pipeline.py): на проде витрина показывала
|
||||
new_count=0 у трёх подряд cian_full_load при реально сохранённых
|
||||
saved_inserted=482/214/239 (honest-run-status) — ключ 'new_count'/'lots_inserted'
|
||||
у full-load'ов в counters не пишется вовсе. 'rows_inserted' — тот же ключ,
|
||||
которым yandex_newbuilding_sweep и rosreestr_dkp_import сообщают число upsert'ов;
|
||||
здесь (для витринной колонки new_count) это безопасно — в отличие от
|
||||
_RESULT_COUNTER_KEYS этот список не участвует в подсчёте zero-result-стрика.
|
||||
|
||||
Возвращает (total_seen, new_count); None для ключа, которого нет в counters —
|
||||
тогда соответствующая колонка не перезаписывается (COALESCE-семантика в UPDATE).
|
||||
"""
|
||||
return _run_result_count(counters), _pick_int(counters, "new_count", "lots_inserted")
|
||||
return _run_result_count(counters), _pick_int(
|
||||
counters, "new_count", "lots_inserted", "saved_inserted", "rows_inserted"
|
||||
)
|
||||
|
||||
|
||||
def _alert_if_consecutive_failures(db: Session, source: str) -> None:
|
||||
|
|
@ -558,6 +653,11 @@ def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
|
|||
#2700: там же — отказ называть успехом прогон, у которого отказала КАЖДАЯ попытка
|
||||
целой фазы (см. _phase_totally_failed). Отличие от #2625: тот случай про «не сделано
|
||||
ничего», этот — про «одно направление работы мертво, а суммарный сбор это прячет».
|
||||
|
||||
honest-run-status: там же — отказ называть успехом прогон с высокой долей отказов,
|
||||
даже если собрано > 0 (см. _failed_ratio_too_high). Отличие от #2625/#2700: те два
|
||||
смотрят на «всё или ничего» (все якоря / вся фаза), этот — на ДОЛЮ отказов у
|
||||
detail-backfill'ов, где ни один из первых двух признаков не матчит форму counters.
|
||||
"""
|
||||
did_nothing = _sweep_run_did_nothing(counters)
|
||||
if did_nothing is not None:
|
||||
|
|
@ -569,6 +669,11 @@ def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
|
|||
logger.error("%s run_id=%d", phase_dead, run_id)
|
||||
mark_failed(db, run_id, phase_dead, counters)
|
||||
return
|
||||
ratio_bad = _failed_ratio_too_high(counters)
|
||||
if ratio_bad is not None:
|
||||
logger.error("%s run_id=%d", ratio_bad, run_id)
|
||||
mark_failed(db, run_id, ratio_bad, counters)
|
||||
return
|
||||
total_seen, new_count = _column_counts(counters)
|
||||
row = db.execute(
|
||||
text(
|
||||
|
|
|
|||
|
|
@ -5,7 +5,17 @@
|
|||
1500-1600 попыток без единого обогащения), yandex 31/52, domclick 24/30
|
||||
(494 попытки → 0 обогащено, 63 блока, 431 fail — и все 30 'done').
|
||||
|
||||
Проверяем ровно ветвление mark_backfill_finished — БД замокана.
|
||||
Проверяем ровно ветвление mark_backfill_finished — БД замокана (mark_done/mark_failed/
|
||||
mark_banned здесь fake-заглушки, регистрирующие ТОЛЬКО факт вызова). Это значит: кейсы
|
||||
ниже с высокой долей отказов (attempted=50, failed=38 или 36 — 76%/72%), ожидающие
|
||||
'done', проверяют лишь то, КАКОЙ финализатор ВЫБРАЛ mark_backfill_finished (#2674:
|
||||
"обогатили хоть что-то — успех"), а НЕ то, что реально запишет в БД mark_done. С
|
||||
honest-run-status (2026-08-15) mark_done САМ переквалифицирует такой прогон в 'failed'
|
||||
через _failed_ratio_too_high (доля отказов >= 0.5) — реальный терминальный статус
|
||||
для этих двух кейсов на проде теперь 'failed', не 'done'. Это намеренно проверяется
|
||||
отдельно, БЕЗ мока mark_done, в tests/test_honest_run_status_failed_ratio.py
|
||||
(test_prod_fact_avito_15_08_no_longer_done и соседние) — не читай эти два кейса как
|
||||
"76%/72% отказов = 'done' в проде".
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
|
@ -59,9 +69,15 @@ def _finish(counters: dict[str, int], *, aborted: bool = False) -> tuple[str, st
|
|||
({"attempted": 5, "enriched": 0, "failed": 5}, False, "failed"),
|
||||
# Кандидатов не было — честная пустота, это успех.
|
||||
({"attempted": 0, "enriched": 0, "blocked": 0, "failed": 0}, False, "done"),
|
||||
# Частичный прогон: обогатили хоть что-то → успех.
|
||||
# Частичный прогон: обогатили хоть что-то → mark_backfill_finished ВЫБИРАЕТ
|
||||
# mark_done как финализатор (#2674). 76% отказов (38 из 50) — здесь mark_done
|
||||
# замокан, поэтому статус остаётся 'done'; в реальном mark_done с
|
||||
# honest-run-status (2026-08-15) это переквалифицируется в 'failed'
|
||||
# (_failed_ratio_too_high, доля >= 0.5) — см. докстринг модуля.
|
||||
({"attempted": 50, "enriched": 12, "blocked": 0, "failed": 38}, False, "done"),
|
||||
# Блоки были, но прогон доработал и обогатил — не бан.
|
||||
# Блоки были, но прогон доработал и обогатил — mark_backfill_finished выбирает
|
||||
# НЕ 'banned'. 72% отказов (36 из 50) — та же оговорка: реальный mark_done
|
||||
# переквалифицирует в 'failed', см. докстринг модуля выше.
|
||||
({"attempted": 50, "enriched": 12, "blocked": 2, "failed": 36}, False, "done"),
|
||||
# Блок оборвал прогон, хотя часть успели обогатить — работа не доделана.
|
||||
({"attempted": 50, "enriched": 12, "blocked": 5, "failed": 33}, True, "banned"),
|
||||
|
|
|
|||
336
tradein-mvp/backend/tests/test_honest_run_status_failed_ratio.py
Normal file
336
tradein-mvp/backend/tests/test_honest_run_status_failed_ratio.py
Normal file
|
|
@ -0,0 +1,336 @@
|
|||
"""honest-run-status (2026-08-15): статус прогона не должен рапортовать 'done' поверх
|
||||
провала или нуля. Три прод-факта закрыты этой правкой:
|
||||
|
||||
(a) avito_detail_backfill 15.08: {"attempted":64,"failed":57,"enriched":6,"blocked":1}
|
||||
-> status='done' — 89% отказов, статус зелёный. mark_backfill_finished звал
|
||||
mark_done, потому что produced=6 (>0); ни _sweep_run_did_nothing (нет
|
||||
anchors_total/errors_count у backfill'ов), ни _phase_totally_failed (ключи
|
||||
"attempted"/"failed" без фазового префикса) эту форму counters не ловили.
|
||||
Фикс: _failed_ratio_too_high внутри mark_done.
|
||||
|
||||
(b) yandex_newbuilding_sweep 26.07-10.08: десять прогонов подряд 'done' при
|
||||
processed=5, succeeded=0, rows_inserted=0, failed_resolve=4-5 — сторож нулевого
|
||||
результата (_alert_if_consecutive_zero_results) слеп, т.к. _RESULT_COUNTER_KEYS
|
||||
не знал ни одного ключа этого sweep'а (total_seen/lots_fetched/unique_fetched).
|
||||
Фикс: _RESULT_COUNTER_KEYS дополнен 'succeeded'. Первая версия правки добавляла
|
||||
голые 'rows_inserted'/'processed' — ревью нашло, что 'rows_inserted' пишет ЕЩЁ
|
||||
rosreestr_dkp_import (66/67 прод-прогонов, здоровый ноль догнавшего импорта, а не
|
||||
отказ) и завёл бы непрерываемый ложный zero-стрик, а 'processed' — счётчик
|
||||
попыток (==limit даже при частичном провале у newbuilding_enrich) и маскирует
|
||||
реальные отказы. 'succeeded' пишут только yandex_newbuilding_sweep и
|
||||
newbuilding_enrich, численно совпадает с прежним 'rows_inserted' на всех
|
||||
прод-прогонах sweep'а — см. test_rosreestr_dkp_import_healthy_zero_stays_unmeasured
|
||||
и test_newbuilding_enrich_partial_failure_not_masked_by_processed ниже.
|
||||
|
||||
(c) admin-витрина показывала new_count=0 у трёх подряд cian_full_load, хотя реально
|
||||
сохранено saved_inserted=482/214/239 — full-load'ы не пишут ни 'new_count', ни
|
||||
'lots_inserted'. Фикс: _column_counts дополнен saved_inserted/rows_inserted.
|
||||
|
||||
Проверяем на обоих модулях (kit-копия и app-копия — байт-эквивалентны по докстрингу
|
||||
runs.py), тем же паттерном, что test_2625_run_that_did_nothing.py.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
|
||||
|
||||
from scraper_kit.orchestration import runs as kit_runs
|
||||
|
||||
from app.services import scrape_runs as app_runs
|
||||
|
||||
_MODULES = {"kit": kit_runs, "app": app_runs}
|
||||
|
||||
|
||||
def _capture_status(mod: Any, counters: dict[str, Any]) -> list[str]:
|
||||
"""Прогнать mark_done на фейковой сессии, вернуть статусы всех UPDATE'ов.
|
||||
|
||||
Тот же helper, что в test_2625_run_that_did_nothing.py — читаем СТАТУС В SQL, а не
|
||||
имя вызванной функции.
|
||||
"""
|
||||
statuses: list[str] = []
|
||||
|
||||
def _execute(stmt: Any, *args: Any, **kwargs: Any) -> MagicMock:
|
||||
sql = str(stmt)
|
||||
for status in ("done", "failed", "banned"):
|
||||
if f"status = '{status}'" in sql:
|
||||
statuses.append(status)
|
||||
return MagicMock()
|
||||
|
||||
db = MagicMock()
|
||||
db.execute.side_effect = _execute
|
||||
with patch.object(mod, "sentry_sdk", MagicMock()):
|
||||
mod.mark_done(db, 1, dict(counters))
|
||||
return statuses
|
||||
|
||||
|
||||
def _capture_backfill_status(
|
||||
counters: dict[str, Any], *, source: str = "avito_detail_backfill", aborted: bool = False
|
||||
) -> list[str]:
|
||||
"""Прогнать app_runs.mark_backfill_finished на фейковой сессии (mark_done НЕ мокан —
|
||||
в отличие от test_backfill_honest_status.py, здесь важно именно его РЕАЛЬНОЕ
|
||||
поведение: mark_backfill_finished решает вызвать mark_done, а решает ли mark_done
|
||||
остаться 'done' или сам себя переквалифицировать в 'failed' — предмет этого теста).
|
||||
|
||||
mark_backfill_finished есть только в app_runs (kit-копия его не держит — см.
|
||||
docstring модуля runs.py, "mark_skipped есть только здесь" — тот же принцип
|
||||
относится к продуктовым финализаторам detail-backfill'ов).
|
||||
"""
|
||||
statuses: list[str] = []
|
||||
|
||||
def _execute(stmt: Any, *args: Any, **kwargs: Any) -> MagicMock:
|
||||
sql = str(stmt)
|
||||
for status in ("done", "failed", "banned"):
|
||||
if f"status = '{status}'" in sql:
|
||||
statuses.append(status)
|
||||
return MagicMock()
|
||||
|
||||
db = MagicMock()
|
||||
db.execute.side_effect = _execute
|
||||
with patch.object(app_runs, "sentry_sdk", MagicMock()):
|
||||
app_runs.mark_backfill_finished(
|
||||
db, 1, dict(counters), source=source, aborted_by_blocks=aborted
|
||||
)
|
||||
return statuses
|
||||
|
||||
|
||||
# ── (a) failed_ratio: прод-факт avito_detail_backfill 15.08 ─────────────────────────
|
||||
|
||||
|
||||
def test_prod_fact_avito_15_08_no_longer_done() -> None:
|
||||
"""{"attempted":64,"failed":57,"enriched":6,"blocked":1} — 89% отказов — 'failed',
|
||||
НЕ 'done'. Красный на старом коде (produced=6 != 0 -> mark_done -> 'done')."""
|
||||
counters = {"attempted": 64, "failed": 57, "enriched": 6, "blocked": 1}
|
||||
assert _capture_backfill_status(counters) == ["failed"]
|
||||
|
||||
|
||||
def test_prod_fact_avito_reason_names_the_ratio() -> None:
|
||||
reason = app_runs._failed_ratio_too_high(
|
||||
{"attempted": 64, "failed": 57, "enriched": 6, "blocked": 1}
|
||||
)
|
||||
assert reason is not None
|
||||
assert "failed-ratio-honest-status" in reason
|
||||
assert "57 из 64" in reason
|
||||
assert "89%" in reason
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
@pytest.mark.parametrize(
|
||||
("counters", "flagged", "why"),
|
||||
[
|
||||
({"attempted": 64, "failed": 57}, True, "прод-факт: 89% отказов"),
|
||||
({"attempted": 10, "failed": 5}, True, "ровно порог failed (0.5)"),
|
||||
({"attempted": 20, "failed": 3}, True, "ровно порог degraded (0.15)"),
|
||||
({"attempted": 20, "failed": 2}, False, "ниже порога degraded (0.10)"),
|
||||
({"attempted": 2, "failed": 2}, False, "ratio=1.0, но < _FAILED_RATIO_MIN_ATTEMPTS"),
|
||||
({"attempted": 0, "failed": 0}, False, "нет попыток вовсе"),
|
||||
({"failed": 5}, False, "нет attempted — чужой словарь"),
|
||||
({"attempted": 50}, False, "нет failed — чужой словарь"),
|
||||
({}, False, "пустые counters"),
|
||||
(
|
||||
{"anchors_total": 5, "errors_count": 5, "lots_fetched": 0},
|
||||
False,
|
||||
"sweep-словарь (anchors_total), не detail-backfill",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_failed_ratio_classifier_boundaries(
|
||||
name: str, counters: dict[str, Any], flagged: bool, why: str
|
||||
) -> None:
|
||||
reason = _MODULES[name]._failed_ratio_too_high(counters)
|
||||
assert (reason is not None) is flagged, why
|
||||
|
||||
|
||||
# ── (5) не должен палить прогоны с малой/умеренной долей отказов ────────────────────
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_low_failure_ratio_stays_done(name: str) -> None:
|
||||
"""Штатный шум (10% отказов) не становится 'failed' — не каждый отказ диагноз."""
|
||||
counters = {"attempted": 50, "enriched": 45, "failed": 5}
|
||||
assert _capture_status(_MODULES[name], counters) == ["done"]
|
||||
|
||||
|
||||
def test_tiny_batch_zero_produced_fails_via_old_rule_not_ratio() -> None:
|
||||
"""2 попытки, обе отказали, produced=0 — доля тут не при чём (attempted < floor
|
||||
_FAILED_RATIO_MIN_ATTEMPTS, _failed_ratio_too_high вернул бы None); статус всё
|
||||
равно 'failed', но по СТАРОМУ правилу #2674 (produced==0), внутри
|
||||
mark_backfill_finished — mark_done/_failed_ratio_too_high тут не вызываются вовсе.
|
||||
Показывает, что новая проверка не дублирует и не подменяет старую."""
|
||||
counters = {"attempted": 2, "enriched": 0, "failed": 2}
|
||||
assert _capture_backfill_status(counters) == ["failed"]
|
||||
|
||||
|
||||
def test_tiny_batch_with_partial_success_stays_done() -> None:
|
||||
"""2 попытки, 1 успех, 1 отказ (ratio=0.5, но attempted < floor=3) — стрик слишком
|
||||
короткий, чтобы доля что-то значила -> остаётся 'done'."""
|
||||
counters = {"attempted": 2, "enriched": 1, "failed": 1}
|
||||
assert _capture_backfill_status(counters) == ["done"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_honest_empty_sweep_unaffected_by_failed_ratio(name: str) -> None:
|
||||
"""Сознательно спящее расписание (город без новостроек): sweep-словарь без
|
||||
attempted/failed вовсе -> failed_ratio не о чем судить, честная пустота остаётся
|
||||
'done' (см. также test_2625_run_that_did_nothing.py::test_honest_empty_stays_done)."""
|
||||
counters = {"anchors_total": 1, "errors_count": 0, "lots_fetched": 0}
|
||||
assert _capture_status(_MODULES[name], counters) == ["done"]
|
||||
|
||||
|
||||
# ── (b) _RESULT_COUNTER_KEYS: прод-факт yandex_newbuilding_sweep 26.07-10.08 ─────────
|
||||
|
||||
|
||||
def test_prod_fact_yandex_newbuilding_sweep_measured_as_zero() -> None:
|
||||
"""processed=5, succeeded=0, rows_inserted=0, failed_resolve=4 — раньше
|
||||
_run_result_count возвращал None ("не измерено"); теперь — измеренный 0 (через
|
||||
'succeeded', не 'rows_inserted' — см. ниже, почему ключ переигран ревью)."""
|
||||
counters = {
|
||||
"total": 309,
|
||||
"fetchable": 200,
|
||||
"pending": 50,
|
||||
"processed": 5,
|
||||
"skipped_already_enriched": 0,
|
||||
"succeeded": 0,
|
||||
"resolved_slug": 1,
|
||||
"failed_resolve": 4,
|
||||
"failed_fetch": 0,
|
||||
"rows_inserted": 0,
|
||||
"duration_sec": 42.0,
|
||||
}
|
||||
assert app_runs._run_result_count(counters) == 0
|
||||
assert kit_runs._run_result_count(counters) == 0
|
||||
|
||||
|
||||
def test_succeeded_is_the_measured_key_not_rows_inserted_or_processed() -> None:
|
||||
"""'succeeded' читается как результат; голые 'rows_inserted'/'processed' в
|
||||
_RESULT_COUNTER_KEYS больше не участвуют (были в первой версии правки, снято
|
||||
ревью — см. test_rosreestr_dkp_import_healthy_zero_stays_unmeasured и
|
||||
test_newbuilding_enrich_partial_failure_not_masked_by_processed ниже)."""
|
||||
counters = {"processed": 5, "rows_inserted": 0}
|
||||
assert app_runs._run_result_count(counters) is None
|
||||
assert kit_runs._run_result_count(counters) is None
|
||||
|
||||
|
||||
def test_rosreestr_dkp_import_healthy_zero_stays_unmeasured() -> None:
|
||||
"""Прод-факт rosreestr_dkp_import (2026-08-15, 66 из 67 прогонов за 90д): инкрементальный
|
||||
импорт догнал источник — rows_fetched==rows_skipped, rows_inserted=0. Это ЗДОРОВЫЙ
|
||||
ответ (нечего вставлять), а не отказ; словарь не содержит 'succeeded' вовсе.
|
||||
|
||||
Первая версия правки добавляла голый 'rows_inserted' в _RESULT_COUNTER_KEYS — тогда
|
||||
этот прод-факт читался бы как "измеренный провал" и копил бы практически
|
||||
непрерываемый zero-стрик (rosreestr_dkp_import не прерывается другим статусом:
|
||||
он либо 'done' с этим же нулём, либо не бежал). Ревью поймало это до деплоя —
|
||||
правильный ответ: "не измерено" (None), стрик не копится."""
|
||||
counters = {
|
||||
"last_id": 6829903,
|
||||
"batches_done": 49,
|
||||
"rows_errored": 0,
|
||||
"rows_fetched": 96974,
|
||||
"rows_skipped": 96974,
|
||||
"rows_updated": 0,
|
||||
"rows_inserted": 0,
|
||||
}
|
||||
assert app_runs._run_result_count(counters) is None
|
||||
assert kit_runs._run_result_count(counters) is None
|
||||
|
||||
|
||||
def test_newbuilding_enrich_partial_failure_not_masked_by_processed() -> None:
|
||||
"""Прод-факт newbuilding_enrich (09.08): processed=25 (счётчик ПОПЫТОК, ==limit),
|
||||
succeeded=14 — 44% отказов. Если бы сторож читал 'processed' как результат, партиальный
|
||||
провал замаскировался бы под measured-25 (сторож нулевого результата промолчал бы
|
||||
ровно там, где должен был сработать при полном провале). 'succeeded' даёт честные 14."""
|
||||
counters = {
|
||||
"failed": 11,
|
||||
"enriched": 14,
|
||||
"attempted": 25,
|
||||
"processed": 25,
|
||||
"succeeded": 14,
|
||||
"failed_fetch": 11,
|
||||
}
|
||||
assert app_runs._run_result_count(counters) == 14
|
||||
assert kit_runs._run_result_count(counters) == 14
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_zero_result_watchdog_now_fires_for_newbuilding_sweep_streak(name: str) -> None:
|
||||
"""(b) integration: 3 подряд yandex_newbuilding_sweep-подобных 'done' с succeeded=0
|
||||
-> алерт срабатывает. До фикса _RESULT_COUNTER_KEYS сторож считал результат "не
|
||||
измеренным" и молчал бы вечно (см. #2703 в docstring модуля)."""
|
||||
mod = _MODULES[name]
|
||||
row = MagicMock()
|
||||
row.status = "done"
|
||||
row.counters = {"processed": 5, "succeeded": 0, "rows_inserted": 0, "failed_resolve": 4}
|
||||
db = MagicMock()
|
||||
result = MagicMock()
|
||||
result.fetchall.return_value = [row, row, row]
|
||||
db.execute.return_value = result
|
||||
with patch.object(mod, "sentry_sdk") as mock_sentry:
|
||||
mod._alert_if_consecutive_zero_results(db, "yandex_newbuilding_sweep")
|
||||
mock_sentry.capture_message.assert_called_once()
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_zero_result_watchdog_silent_on_rosreestr_dkp_import_streak(name: str) -> None:
|
||||
"""Негативный аналог теста выше: та же лестница из 3 подряд 'done', но словарь
|
||||
rosreestr_dkp_import (нет 'succeeded') -> сторож не считает результат измеренным
|
||||
и НЕ шлёт алерт — регрессионный тест на замечание ревью (HIGH #1)."""
|
||||
mod = _MODULES[name]
|
||||
row = MagicMock()
|
||||
row.status = "done"
|
||||
row.counters = {
|
||||
"last_id": 6829903,
|
||||
"rows_fetched": 96974,
|
||||
"rows_skipped": 96974,
|
||||
"rows_inserted": 0,
|
||||
}
|
||||
db = MagicMock()
|
||||
result = MagicMock()
|
||||
result.fetchall.return_value = [row, row, row]
|
||||
db.execute.return_value = result
|
||||
with patch.object(mod, "sentry_sdk") as mock_sentry:
|
||||
mod._alert_if_consecutive_zero_results(db, "rosreestr_dkp_import")
|
||||
mock_sentry.capture_message.assert_not_called()
|
||||
|
||||
|
||||
# ── (c) _column_counts: прод-факт cian_full_load new_count=0 при saved_inserted>0 ───
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_prod_fact_cian_full_load_saved_inserted_surfaces_as_new_count(name: str) -> None:
|
||||
"""saved_inserted=482 (прод-факт: три подряд прогона 482/214/239) — new_count
|
||||
больше не 0, хотя ключей 'new_count'/'lots_inserted' в counters нет вовсе."""
|
||||
counters = {"unique_fetched": 1200, "saved_inserted": 482, "saved_updated": 30}
|
||||
total_seen, new_count = _MODULES[name]._column_counts(counters)
|
||||
assert total_seen == 1200
|
||||
assert new_count == 482
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_yandex_newbuilding_rows_inserted_surfaces_as_new_count(name: str) -> None:
|
||||
counters = {"rows_inserted": 7}
|
||||
_, new_count = _MODULES[name]._column_counts(counters)
|
||||
assert new_count == 7
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_new_count_priority_unchanged_by_new_keys(name: str) -> None:
|
||||
"""'new_count' явный ключ всё ещё побеждает 'lots_inserted'/'saved_inserted' —
|
||||
расширение списка не меняет приоритет уже существующих ключей."""
|
||||
counters = {"new_count": 5, "lots_inserted": 99, "saved_inserted": 1}
|
||||
_, new_count = _MODULES[name]._column_counts(counters)
|
||||
assert new_count == 5
|
||||
|
||||
|
||||
@pytest.mark.parametrize("name", list(_MODULES))
|
||||
def test_lots_inserted_still_beats_saved_inserted(name: str) -> None:
|
||||
"""Порядок пикулярно НЕ переставлен для уже существующей пары — 'lots_inserted'
|
||||
(city/newbuilding-sweep'ы) проверяется раньше 'saved_inserted' (full-load'ы),
|
||||
т.к. это разные, непересекающиеся семейства источников."""
|
||||
counters = {"lots_inserted": 12, "saved_inserted": 999}
|
||||
_, new_count = _MODULES[name]._column_counts(counters)
|
||||
assert new_count == 12
|
||||
|
|
@ -138,6 +138,28 @@ def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
|
|||
# unique_fetched — full-load'ы avito/cian/yandex (4 источника, 133 прогона) — раньше
|
||||
# сторож их не видел, хотя у cian_full_load 6 из 38 успешных прогонов
|
||||
# реально дали ноль.
|
||||
# succeeded — yandex_newbuilding_sweep (42 прогона/90д) и newbuilding_enrich
|
||||
# (65 прогонов/90д, единственные два писателя ключа на проде,
|
||||
# проверено 2026-08-15). НЕ 'rows_inserted': тот ключ пишет ЕЩЁ и
|
||||
# rosreestr_dkp_import (67 прогонов/90д) — у него rows_inserted=0 в
|
||||
# 66 из 67 это ЗДОРОВЫЙ ответ догнавшего инкрементального импорта
|
||||
# (rows_fetched=rows_skipped=96974, last_id не двигается неделями),
|
||||
# а не отказ; если бы 'rows_inserted' попал в этот список, сторож
|
||||
# зачитывал бы этот здоровый ноль как измеренный провал и копил бы
|
||||
# практически непрерываемый стрик (rosreestr_dkp_import не
|
||||
# прерывается другим статусом — импорт либо 'done', либо не бежал).
|
||||
# НЕ 'processed' по той же причине с другой стороны: это счётчик
|
||||
# ПОПЫТОК (у newbuilding_enrich processed==attempted==limit даже
|
||||
# когда succeeded меньше — прод-факт 09.08: processed=25 succeeded=14,
|
||||
# 44% отказов замаскировались бы под measured-25) — сторож нулевого
|
||||
# результата на нём молчал бы ровно там, где должен сработать, а на
|
||||
# будущем опустении очереди домов (cian_houses_pending) создал бы
|
||||
# свой вечный ложный zero-стрик. 'succeeded' у yandex_newbuilding_sweep
|
||||
# численно совпадает с 'rows_inserted' на всех 42/42 прод-прогонах —
|
||||
# замена не теряет исходную цель (десять прогонов подряд 26.07-10.08,
|
||||
# все 'done', succeeded=0 rows_inserted=0 failed_resolve=4-5 — раньше
|
||||
# ни total_seen/lots_fetched/unique_fetched не было, и
|
||||
# _run_result_count всегда возвращал None (honest-run-status)).
|
||||
# Сводить сюда счётчики ОСТАЛЬНЫХ задач бессмысленно: на проде 28 источников (2650
|
||||
# прогонов) не имеют общего результатного ключа вовсе — у каждого свой словарь
|
||||
# (deactivated / rows_written / poi_loaded / snapshotted / upserted / listings_matched
|
||||
|
|
@ -145,7 +167,12 @@ def _pick_int(counters: Mapping[str, Any], *keys: str) -> int | None:
|
|||
# трёх мониторов результата нет по смыслу. Ноль у них — часто ЗДОРОВЫЙ ответ
|
||||
# (deactivate_stale_* без протухших объявлений). Поэтому сторож не угадывает их
|
||||
# словарь, а честно признаёт, что мерить нечем — см. _run_result_count.
|
||||
_RESULT_COUNTER_KEYS = ("total_seen", "lots_fetched", "unique_fetched")
|
||||
_RESULT_COUNTER_KEYS = (
|
||||
"total_seen",
|
||||
"lots_fetched",
|
||||
"unique_fetched",
|
||||
"succeeded",
|
||||
)
|
||||
|
||||
|
||||
def _run_result_count(counters: Mapping[str, Any] | None) -> int | None:
|
||||
|
|
@ -277,6 +304,68 @@ def _phase_totally_failed(counters: Mapping[str, Any]) -> str | None:
|
|||
return None
|
||||
|
||||
|
||||
# honest-run-status (2026-08-15): доля отказов, которая обесценивает формально ненулевой
|
||||
# сбор. Прод-факт avito_detail_backfill 15.08: {"attempted":64,"failed":57,"enriched":6,
|
||||
# "blocked":1} — 89% попыток отказали, а mark_backfill_finished всё равно звал mark_done,
|
||||
# потому что "produced != 0" (6 обогащено). Ни _sweep_run_did_nothing (нужны
|
||||
# anchors_total/errors_count, у backfill'ов их нет), ни _phase_totally_failed (нужна пара
|
||||
# "<phase>_attempted"/"<phase>_failed" — здесь голые "attempted"/"failed" без фазового
|
||||
# префикса, `"attempted".endswith("_attempted")` не матчит) эту форму counters не ловят —
|
||||
# обе проверки написаны под СВОИ формы, а не под backfill'овскую.
|
||||
#
|
||||
# Порог 'failed' — половина и больше отказов: сбор для практических целей провалился,
|
||||
# даже если несколько записей всё же обогатились. Порог 'partial' НЕ заведён отдельным
|
||||
# статусом scrape_runs.status — это потребовало бы миграции (DROP+ADD CHECK constraint,
|
||||
# 051_scrape_runs_extend.sql) и обучило бы новому значению ещё 4 места (Literal-фильтр
|
||||
# admin API, хардкод статусов фронта, оба IN-списка сторожей) — тот же класс "оборванной
|
||||
# проводки", из-за которого заведён #2686/ban_kind. Вместо статуса — тот же диагноз, что и
|
||||
# у ban_kind: causa в тексте `error`, терминальный статус один ('failed'). 0.15..0.5 —
|
||||
# та же 'failed', но с другой формулировкой причины ("деградировал", не "провалился"), чтобы
|
||||
# оператор видел разницу читая error, не только status.
|
||||
#
|
||||
# mark_backfill_finished (единственный писатель "attempted"/"failed" на верхнем уровне
|
||||
# counters) живёт только в app.services.scrape_runs — здесь эта проверка сейчас неактивна
|
||||
# ни для одного реального вызывающего, но kit-копия держится байт-эквивалентной app-копии
|
||||
# (см. docstring модуля), и будущий kit-native job с тем же словарём получит её даром.
|
||||
FAILED_RATIO_FAILED_THRESHOLD = 0.5
|
||||
FAILED_RATIO_DEGRADED_THRESHOLD = 0.15
|
||||
# Минимум попыток, при котором доля вообще что-то значит — иначе 1 отказ из 2 (=0.5)
|
||||
# палит статус на шуме единичного случая. То же рассуждение и то же число, что у
|
||||
# _PHASE_MIN_ATTEMPTS (см. выше).
|
||||
_FAILED_RATIO_MIN_ATTEMPTS = _PHASE_MIN_ATTEMPTS
|
||||
|
||||
|
||||
def _failed_ratio_too_high(counters: Mapping[str, Any]) -> str | None:
|
||||
"""Прогон, у которого доля отказов слишком велика, даже если что-то собрано.
|
||||
|
||||
Возвращает текст причины (для error) либо None. Читает ГОЛЫЕ ключи "attempted"/
|
||||
"failed" (без фазового префикса) — сейчас это словарь только у четырёх
|
||||
detail-backfill'ов (avito/yandex/domclick/newbuilding_enrich), все идут через
|
||||
mark_backfill_finished → mark_done. `attempted < _FAILED_RATIO_MIN_ATTEMPTS` или
|
||||
отсутствие любого из ключей → None (нечем/не о чём судить — счётчики либо не
|
||||
заполнены, либо принадлежат другому источнику со своим словарём).
|
||||
|
||||
Что признак НЕ доказывает: КТО виноват (площадка, наш прокси, наш парсер) — поэтому
|
||||
'failed' без диагноза, как и у #2625/#2700/#2764.
|
||||
"""
|
||||
attempted = _pick_int(counters, "attempted")
|
||||
failed = _pick_int(counters, "failed")
|
||||
if attempted is None or failed is None or attempted < _FAILED_RATIO_MIN_ATTEMPTS:
|
||||
return None
|
||||
ratio = failed / max(attempted, 1)
|
||||
if ratio >= FAILED_RATIO_FAILED_THRESHOLD:
|
||||
verb = "провалился"
|
||||
elif ratio >= FAILED_RATIO_DEGRADED_THRESHOLD:
|
||||
verb = "деградировал"
|
||||
else:
|
||||
return None
|
||||
return (
|
||||
f"failed-ratio-honest-status: сбор {verb} — {failed} из {attempted} попыток "
|
||||
f"отказали (доля {ratio:.0%}); формально ненулевой результат этого не искупает. "
|
||||
f"Причина НЕ установлена — статус 'failed' без диагноза"
|
||||
)
|
||||
|
||||
|
||||
def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
|
||||
"""Извлечь значения для dedicated-колонок total_seen / new_count из jsonb-counters.
|
||||
|
||||
|
|
@ -287,13 +376,24 @@ def _column_counts(counters: dict[str, int]) -> tuple[int | None, int | None]:
|
|||
показывала total_seen=0 при реально сохранённых строках (audit #1871/#1926).
|
||||
|
||||
Приоритет ключей:
|
||||
- total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched)
|
||||
- new_count ← 'new_count' (если уже есть) иначе 'lots_inserted'
|
||||
- total_seen ← _RESULT_COUNTER_KEYS (total_seen / lots_fetched / unique_fetched /
|
||||
succeeded)
|
||||
- new_count ← 'new_count' / 'lots_inserted' / 'saved_inserted' / 'rows_inserted'
|
||||
(первый присутствующий). 'saved_inserted' — full-load'ы (cian/avito/yandex,
|
||||
CianFullLoadCounters и аналоги в pipeline.py): на проде витрина показывала
|
||||
new_count=0 у трёх подряд cian_full_load при реально сохранённых
|
||||
saved_inserted=482/214/239 (honest-run-status) — ключ 'new_count'/'lots_inserted'
|
||||
у full-load'ов в counters не пишется вовсе. 'rows_inserted' — тот же ключ,
|
||||
которым yandex_newbuilding_sweep и rosreestr_dkp_import сообщают число upsert'ов;
|
||||
здесь (для витринной колонки new_count) это безопасно — в отличие от
|
||||
_RESULT_COUNTER_KEYS этот список не участвует в подсчёте zero-result-стрика.
|
||||
|
||||
Возвращает (total_seen, new_count); None для ключа, которого нет в counters —
|
||||
тогда соответствующая колонка не перезаписывается (COALESCE-семантика в UPDATE).
|
||||
"""
|
||||
return _run_result_count(counters), _pick_int(counters, "new_count", "lots_inserted")
|
||||
return _run_result_count(counters), _pick_int(
|
||||
counters, "new_count", "lots_inserted", "saved_inserted", "rows_inserted"
|
||||
)
|
||||
|
||||
|
||||
def _alert_if_consecutive_failures(db: Session, source: str) -> None:
|
||||
|
|
@ -632,6 +732,11 @@ def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
|
|||
#2700: там же — отказ называть успехом прогон, у которого отказала КАЖДАЯ попытка
|
||||
целой фазы (см. _phase_totally_failed). Отличие от #2625: тот случай про «не сделано
|
||||
ничего», этот — про «одно направление работы мертво, а суммарный сбор это прячет».
|
||||
|
||||
honest-run-status: там же — отказ называть успехом прогон с высокой долей отказов,
|
||||
даже если собрано > 0 (см. _failed_ratio_too_high). Отличие от #2625/#2700: те два
|
||||
смотрят на «всё или ничего» (все якоря / вся фаза), этот — на ДОЛЮ отказов у
|
||||
detail-backfill'ов, где ни один из первых двух признаков не матчит форму counters.
|
||||
"""
|
||||
did_nothing = _sweep_run_did_nothing(counters)
|
||||
if did_nothing is not None:
|
||||
|
|
@ -643,6 +748,11 @@ def mark_done(db: Session, run_id: int, counters: dict[str, int]) -> None:
|
|||
logger.error("%s run_id=%d", phase_dead, run_id)
|
||||
mark_failed(db, run_id, phase_dead, counters)
|
||||
return
|
||||
ratio_bad = _failed_ratio_too_high(counters)
|
||||
if ratio_bad is not None:
|
||||
logger.error("%s run_id=%d", ratio_bad, run_id)
|
||||
mark_failed(db, run_id, ratio_bad, counters)
|
||||
return
|
||||
total_seen, new_count = _column_counts(counters)
|
||||
row = db.execute(
|
||||
text(
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue