fix(tradein/scrapers): диагноз блока терялся при схлопывании, а в алерт шла непроверенная причина

Прогоны 5140-5190 писали ban_kind=unknown при том, что причина была
одна и та же во всех: AvitoBlockedError, firewall/soft-block (browser-mode).
Диагноз обнулял один нетипичный блок из пяти — правило схлопывания
требовало РОВНО одного вида, иначе unknown. Явное большинство (4 из 5)
пропадало вместе с редким.

Теперь mark_backfill_finished принимает перепись диагнозов, кладёт её
в counters["ban_kinds"] и выбирает по строгому большинству. Замысел
#2764 сохранён: пустая перепись и настоящая ничья по-прежнему дают
unknown — диагноз не назначается там, где его не видели.

Заодно убраны захардкоженные причины из ABORT-логов: "IP rate-limited"
у Авито и "QRATOR reputation likely burned" у Домклика. Ни одна из них
не проверялась кодом, и первая скрывала настоящий диагноз, лежавший
рядом в error_text прогона. Авито теперь пишет измеренную подпись
отказа, Домклик — "причина не определена": там нет failure_census,
и выдумывать измерение вместо него нечестно.

Refs #3178

tradein-mvp/backend/app/services/scrape_runs.py

@@ -33,6 +33,7 @@ from __future__ import annotations
 
 import json
 import logging
+from collections import Counter
 from collections.abc import Callable, Collection, Mapping
 from functools import cache
 from typing import Any
@@ -819,6 +820,32 @@ def mark_banned(
     _alert_on_run_id(db, run_id)
 
 
+def _dominant_ban_kind(census: Mapping[str, int]) -> str:
+    """Диагноз по переписи блоков прогона: kind -> сколько раз он встретился (#3178).
+
+    Раньше вызывающий код терял кратности до вызова (`set()`), поэтому 4 блока
+    'platform' + 1 'infra' и 2+2 давали функции один и тот же вход {'platform',
+    'infra'} — неотличимые случаи, хотя первый явно платформенный, а второй
+    действительно спорный. Перепись приходит уже с кратностями (Counter), здесь —
+    только выбор:
+      - пусто → 'unknown' (диагнозов не было вовсе);
+      - один вид → он, независимо от количества;
+      - несколько видов, но один строго больше половины всех блоков → он
+        (доминирующий диагноз, единичные выбросы других типов его не размывают);
+      - иначе (нет строгого большинства) → 'unknown' — по какой причине оборвался
+        именно этот прогон, честно не знаем.
+    """
+    if not census:
+        return BAN_KIND_UNKNOWN
+    if len(census) == 1:
+        return next(iter(census))
+    total = sum(census.values())
+    kind, count = max(census.items(), key=lambda kv: kv[1])
+    if count > total / 2:
+        return kind
+    return BAN_KIND_UNKNOWN
+
+
 def mark_backfill_finished(
     db: Session,
     run_id: int,
@@ -827,7 +854,7 @@ def mark_backfill_finished(
     source: str,
     aborted_by_blocks: bool = False,
     fail_hint: str | None = None,
-    ban_kinds: Collection[str] = (),
+    ban_kinds: Collection[str] | Mapping[str, int] = (),
 ) -> None:
     """Честный финал detail-backfill'а (#2674): нулевой прогон ≠ 'done'.
 
@@ -860,11 +887,18 @@ def mark_backfill_finished(
     контейнера, то есть на первом же деплое после ночного прогона.
 
     `ban_kinds` — диагнозы (ban_kind_of_exception) ВСЕХ блоков, которые задача
-    поймала за прогон; пустой (дефолт) = задача типы не различает. Схлопываем сами,
+    поймала за прогон; пустой (дефолт) = задача типы не различает. Принимает либо
-    в одном месте на все три backfill'а: все блоки сошлись в одном диагнозе → он и
+    Collection[str] (старые вызовы — список/set диагнозов, кратности не несут) либо
-    пишется; разошлись (или их типы ничего не доказывают) → 'unknown'. Смешанный
+    уже готовую перепись Mapping[str, int] (kind -> сколько раз). Раньше здесь стоял
-    прогон честнее пометить неизвестным, чем выбрать из двух причин ту, что
+    set(ban_kinds) — терял кратности ДО решения: 4 блока 'platform' + 1 'infra'
-    попалась последней — какая из них оборвала прогон, мы не знаем (#2764).
+    схлопывались в тот же вход {'platform', 'infra'}, что и настоящие 2+2, и оба
+    давали 'unknown' (#3178, прод: 5 прогонов подряд 4×platform+1×infra → unknown,
+    один прогон 5/5 одного вида → platform — при том же исключении на каждом блоке,
+    AvitoBlockedError firewall/soft-block). Перепись кладём в
+    counters["ban_kinds"] (kind -> count) — переживает финализацию наравне с
+    остальными counters, диагноз строки прогона выбирает _dominant_ban_kind: один
+    вид → он; явное большинство (строго > половины блоков) → он; иначе — 'unknown',
+    честно «не знаем, какой из них оборвал прогон» (#2764).
     """
     attempted = int(counters.get("attempted") or 0)
     enriched = int(counters.get("enriched") or 0)
@@ -882,13 +916,18 @@ def mark_backfill_finished(
             f"blocked={blocked}, обогащено {enriched} из {attempted} попыток{hint} (#2674)"
         )
         logger.error("%s run_id=%d", reason, run_id)
-        kinds = set(ban_kinds)
+        # Counter() принимает и Collection (считает элементы — старые set/list-вызовы),
+        # и Mapping (копирует кратности как есть — census от вызывающего) одним и тем
+        # же конструктором.
+        census = Counter(ban_kinds)
+        if census:
+            counters["ban_kinds"] = dict(census)  # type: ignore[assignment]
         mark_banned(
             db,
             run_id,
             reason,
             counters,
-            ban_kind=kinds.pop() if len(kinds) == 1 else BAN_KIND_UNKNOWN,
+            ban_kind=_dominant_ban_kind(census),
         )
         return
 

tradein-mvp/backend/app/tasks/avito_detail_backfill.py

@@ -407,10 +407,12 @@ async def run_avito_detail_backfill(
         # Перепись причин (блоки + отказы) — переживает пересоздание контейнера,
         # в отличие от логов; см. _failure_signature.
         failure_census: Counter[str] = Counter()
-        # #2764: диагнозы всех блоков прогона по ТИПУ исключения. Сойдутся в один —
+        # #2764/#3178: диагнозы всех блоков прогона по ТИПУ исключения, С кратностями
-        # он и попадёт в scrape_runs.ban_kind, разойдутся — 'unknown' (схлопывает
+        # (Counter, не set) — set терял их до решения: 5 прогонов подряд 4×platform+
-        # mark_backfill_finished, один узел на все три backfill'а).
+        # 1×infra и настоящий 2+2 приходили в mark_backfill_finished одинаково и
-        block_ban_kinds: set[str] = set()
+        # получали 'unknown' оба раза, хотя первый явно платформенный. Перепись
+        # (kind -> count) решает _dominant_ban_kind в scrape_runs.py.
+        block_ban_kinds: Counter[str] = Counter()
 
         for idx, row in enumerate(snapshot):
             # Budget guard
@@ -587,7 +589,7 @@ async def run_avito_detail_backfill(
                 consecutive_blocks += 1
                 counters.blocked += 1
                 failure_census[_failure_signature(e)] += 1
-                block_ban_kinds.add(ban_kind_of_exception(e))
+                block_ban_kinds[ban_kind_of_exception(e)] += 1
                 do_sleep = False
                 logger.warning(
                     "avito_detail_backfill: run_id=%d BLOCKED #%d/%d (consecutive=%d): %s",
@@ -602,9 +604,10 @@ async def run_avito_detail_backfill(
                 if consecutive_blocks >= max_consecutive_blocks:
                     logger.error(
                         "avito_detail_backfill: run_id=%d ABORT -- %d consecutive blocks, "
-                        "IP rate-limited. enriched=%d attempted=%d",
+                        "частая причина: %s. enriched=%d attempted=%d",
                         run_id,
                         consecutive_blocks,
+                        _top_failure(failure_census) or "причина не определена",
                         counters.enriched,
                         counters.attempted,
                     )

tradein-mvp/backend/app/tasks/domclick_detail_backfill.py

@@ -338,9 +338,12 @@ async def run_domclick_detail_backfill(
                         e,
                     )
                     if consecutive_blocks >= max_consecutive_blocks:
+                        # #2764/#3178: DomClickBlockedError не разводит площадку (QRATOR)
+                        # и наш браузерный тракт (см. докстринг класса выше) — причину
+                        # НЕ выдумываем, пишем как есть.
                         logger.error(
                             "domclick_detail_backfill: run_id=%d ABORT -- %d consecutive "
-                            "blocks, QRATOR reputation likely burned for the session/proxy. "
+                            "blocks, причина не определена (площадка либо наш тракт). "
                             "enriched=%d attempted=%d",
                             run_id,
                             consecutive_blocks,

tradein-mvp/backend/tests/tasks/test_avito_detail_backfill.py

@@ -243,6 +243,86 @@ async def test_backfill_reports_ban_kind_of_the_blocks_it_saw(
     assert set(runs.mark_backfill_finished.call_args.kwargs["ban_kinds"]) == expected_kinds
 
 
+@pytest.mark.asyncio
+async def test_backfill_ban_kinds_census_keeps_multiplicities() -> None:
+    """4×AvitoBlockedError + 1×AvitoSidecarUnavailableError -> перепись 4/1, не set (#3178).
+
+    Фальсификация: до правки block_ban_kinds был set() -- .add() схлопнул бы этот
+    прогон в тот же {'platform', 'infra'}, что и настоящие 2+2, и mark_backfill_finished
+    не смог бы отличить явное большинство от честной ничьей.
+    """
+    from scraper_kit.avito_exceptions import AvitoBlockedError, AvitoSidecarUnavailableError
+
+    snapshot = _make_snapshot(10)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    # abort-check срабатывает ПЕРЕД recovery на 5-м блоке -- ровно 5 попыток.
+    mock_fetch = AsyncMock(
+        side_effect=[
+            AvitoBlockedError("firewall"),
+            AvitoBlockedError("firewall"),
+            AvitoBlockedError("firewall"),
+            AvitoBlockedError("firewall"),
+            AvitoSidecarUnavailableError("browser unavailable"),
+        ]
+    )
+    mock_scraper = MagicMock()
+    mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
+    fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION, return_value=AsyncMock()),
+        patch(_SCRAPER, mock_scraper),
+        patch(_RUNS, runs),
+        patch(_RESOLVE_PROXY_URL, MagicMock(return_value="http://test-proxy.local:8080")),
+        patch(_FETCH, mock_fetch),
+        patch(_SLEEP, new_callable=AsyncMock),
+    ):
+        await run_avito_detail_backfill(
+            db, run_id=3, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}
+        )
+
+    runs.mark_backfill_finished.assert_called_once()
+    census = runs.mark_backfill_finished.call_args.kwargs["ban_kinds"]
+    assert dict(census) == {"platform": 4, "infra": 1}
+
+
+@pytest.mark.asyncio
+async def test_backfill_abort_log_has_no_ip_rate_limited_literal(caplog: Any) -> None:
+    """ABORT-лог называет измеренную причину, а не литерал 'IP rate-limited' (#3178).
+
+    До правки текст ABORT всегда писал 'IP rate-limited' -- даже когда блоки были
+    отказом НАШЕГО сайдкара (AvitoSidecarUnavailableError), не площадки.
+    """
+    from scraper_kit.avito_exceptions import AvitoBlockedError
+
+    snapshot = _make_snapshot(10)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    mock_fetch = AsyncMock(side_effect=AvitoBlockedError("firewall/soft-block"))
+    mock_scraper = MagicMock()
+    mock_scraper.return_value._rotate_ip = AsyncMock(return_value=True)
+    fake_settings = MagicMock(scraper_fetch_mode="cffi", avito_detail_backfill_use_curl=False)
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION, return_value=AsyncMock()),
+        patch(_SCRAPER, mock_scraper),
+        patch(_RUNS, runs),
+        patch(_RESOLVE_PROXY_URL, MagicMock(return_value="http://test-proxy.local:8080")),
+        patch(_FETCH, mock_fetch),
+        patch(_SLEEP, new_callable=AsyncMock),
+        caplog.at_level("ERROR"),
+    ):
+        await run_avito_detail_backfill(
+            db, run_id=3, params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}
+        )
+
+    abort_records = [r.message for r in caplog.records if "ABORT" in r.message]
+    assert abort_records, "ожидался ABORT-лог"
+    assert "IP rate-limited" not in abort_records[0]
+    assert "AvitoBlockedError" in abort_records[0]
+
+
 @pytest.mark.asyncio
 async def test_backfill_blocked_abort_after_max_consecutive() -> None:
     """5 consecutive AvitoBlockedError -> abort с пометкой aborted_by_blocks (#2674).

tradein-mvp/backend/tests/tasks/test_domclick_detail_backfill.py

@@ -249,6 +249,45 @@ async def test_backfill_blocked_abort_after_max_consecutive() -> None:
     runs.mark_failed.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_backfill_abort_log_has_no_qrator_literal(caplog: pytest.LogCaptureFixture) -> None:
+    """ABORT-лог не утверждает конкретную причину, которую задача не устанавливает (#3178).
+
+    DomClickBlockedError поднимается и на распознанном QRATOR-маркере (площадка), и
+    на любом другом сбое браузерного fetch (наш тракт) -- см. докстринг класса выше
+    (#2764: диагноз здесь НЕ установлен). До правки ABORT всегда писал 'QRATOR
+    reputation likely burned for the session/proxy' -- утверждение, для которого нет
+    основания в этом прогоне.
+    """
+    snapshot = _make_snapshot(10)
+    db = _mock_db(snapshot)
+    runs = MagicMock()
+    blocked_exc = DomClickBlockedError("browser fetch failed")
+    mock_fetch = AsyncMock(side_effect=blocked_exc)
+    fake_settings = MagicMock(browser_http_endpoint="http://browser:9000")
+    mock_svc = _mock_session_svc({"CAS_ID": "123"})
+    mock_bf_cls = _mock_browser_fetcher_cls()
+    with (
+        patch(_SETTINGS, fake_settings),
+        patch(_SESSION_SVC, mock_svc),
+        patch(_RUNS, runs),
+        patch(_BROWSER_FETCHER, mock_bf_cls),
+        patch(_FETCH, mock_fetch),
+        patch(_SLEEP, new_callable=AsyncMock),
+        caplog.at_level("ERROR"),
+    ):
+        await run_domclick_detail_backfill(
+            db,
+            run_id=4,
+            params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 3},
+        )
+
+    abort_records = [r.message for r in caplog.records if "ABORT" in r.message]
+    assert abort_records, "ожидался ABORT-лог"
+    assert "QRATOR reputation likely burned" not in abort_records[0]
+    assert "причина не определена" in abort_records[0]
+
+
 @pytest.mark.asyncio
 async def test_backfill_parse_error_counts_failed_no_abort() -> None:
     """DomClickParseError (schema drift, not a block) -> failed++, consecutive-block

tradein-mvp/backend/tests/test_2764_ban_kind_no_default.py

@@ -21,6 +21,7 @@
 from __future__ import annotations
 
 import os
+from collections import Counter
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -96,6 +97,51 @@ def test_finalizer_with_mixed_diagnoses_writes_unknown() -> None:
     )
 
 
+# ── 1b. Перепись (#3178): кратности решают, set() их терял ───────────────────
+
+
+def test_finalizer_census_majority_wins_over_minority() -> None:
+    """4×platform + 1×infra → 'platform': явное большинство, не 'unknown' (#3178).
+
+    Фальсификация: до правки вызывающий терял кратности через set(ban_kinds) ДО
+    решения — {'platform', 'infra'} от 4+1 был неотличим от настоящего 2+2, и оба
+    давали 'unknown'. Прод: 5 прогонов подряд с одним и тем же AvitoBlockedError
+    (firewall/soft-block, browser-mode) — 4 блока сошлись в 'platform', 1 в 'infra',
+    строка прогона получала 'unknown' там, где явное большинство прямо говорило
+    'platform'.
+    """
+    census = Counter({runs_mod.BAN_KIND_PLATFORM: 4, runs_mod.BAN_KIND_INFRA: 1})
+    assert _ban_kind_of_finished(ban_kinds=census) == runs_mod.BAN_KIND_PLATFORM
+
+
+def test_finalizer_census_tie_writes_unknown() -> None:
+    """2×platform + 2×infra — ровно поровну, строгого большинства нет → 'unknown'."""
+    census = Counter({runs_mod.BAN_KIND_PLATFORM: 2, runs_mod.BAN_KIND_INFRA: 2})
+    assert _ban_kind_of_finished(ban_kinds=census) == runs_mod.BAN_KIND_UNKNOWN
+
+
+def test_finalizer_census_all_same_kind() -> None:
+    """5 из 5 одного вида → он же, вне зависимости от абсолютного счёта блоков."""
+    census = Counter({runs_mod.BAN_KIND_PLATFORM: 5})
+    assert _ban_kind_of_finished(ban_kinds=census) == runs_mod.BAN_KIND_PLATFORM
+
+
+def test_finalizer_census_lands_in_counters() -> None:
+    """Перепись (kind -> count) остаётся в counters['ban_kinds'] после выбора
+    диагноза строки прогона — не теряется вместе с решением (#3178)."""
+    counters = dict(_BLOCKED_RUN)
+    with patch.object(runs_mod, "mark_banned", lambda *a, **k: None):
+        runs_mod.mark_backfill_finished(
+            MagicMock(),
+            1,
+            counters,
+            source="avito_detail_backfill",
+            aborted_by_blocks=True,
+            ban_kinds=Counter({runs_mod.BAN_KIND_PLATFORM: 4, runs_mod.BAN_KIND_INFRA: 1}),
+        )
+    assert counters["ban_kinds"] == {"platform": 4, "infra": 1}
+
+
 # ── 2. Диагноз не врёт там, где он передаётся: browser-ветка fetch_detail ────