From 19a335f4def3b15f811622305ee78a5d42e22b19 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Mon, 7 Sep 2026 16:47:08 +0500 Subject: [PATCH] =?UTF-8?q?fix(tradein/scraper-kit):=20=D0=B1=D0=B0=D0=BA?= =?UTF-8?q?=D0=B5=D1=82=20=D0=B2=20done=5Fbuckets=20=D1=82=D0=BE=D0=BB?= =?UTF-8?q?=D1=8C=D0=BA=D0=BE=20=D0=B5=D1=81=D0=BB=D0=B8=20=D0=BA=D0=B0?= =?UTF-8?q?=D0=B6=D0=B4=D0=B0=D1=8F=20=D1=84=D0=B0=D0=B7=D0=B0=20=D0=B4?= =?UTF-8?q?=D0=B0=D0=BB=D0=B0=20=D1=83=D0=B4=D0=B0=D1=87=D1=83=20(#3415)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit cian_city_sweep 6179 (06.09) — `failed` по honest-status «фаза 'houses' отказала полностью — 30 из 30», но все пять якорей записаны в counters.done_buckets. Прогон 6276 (07.09) унаследовал точку, пропустил все якоря и отчитался `done` с houses_attempted=0, lots_fetched=0 — зелёный по построению: фаза не исполнялась. Из-за этого фикс #3396 (houses через пул прокси) простоял на проде сутки, ни разу не отработав. Корень: отметка «якорь пройден» опиралась на поток управления (исключение/таймаут → continue), а фаза отказывает и БЕЗ исключения — fetch_newbuilding возвращает None, растёт только счётчик. _bucket_phase_totally_failed сверяет прирост пары X_attempted/X_failed за якорь (та же конвенция, что у runs._phase_totally_failed, но порог одна попытка, а не три: здесь решается «собирать ли заново», и цена ошибок несимметрична). Форма чекпоинта не менялась — плоский список имён якорей: его читают ещё три свипа и scheduler._resume_decision, а ключ `bucket:phase` потребовал бы новых читателей ради того же решения; резюм со старой точкой работает как прежде. Тот же гейт — в run_avito_city_sweep (та же схема, отдельная функция; прод-следа за 45 суток у него нет, у cian — 4 прогона). Ветка «houses DB query failed» теперь растит houses_attempted вместе с houses_failed: пара attempted=0/failed=N не видна ни run-level правилу, ни гейту бакета. Соседи без правки: yandex combo-чекпоинт (единица пишется SERP-слоем до фазы, пробел покрывает yandex_address_backfill, 0 прогонов на проде), domclick (одна фаза), full-load'ы (бакет = SERP-страница, detail тянется из БД, не из бакета). Closes #3415 --- ...test_3415_bucket_done_only_if_phases_ok.py | 271 ++++++++++++++++++ .../src/scraper_kit/orchestration/pipeline.py | 94 +++++- 2 files changed, 363 insertions(+), 2 deletions(-) create mode 100644 tradein-mvp/backend/tests/test_3415_bucket_done_only_if_phases_ok.py diff --git a/tradein-mvp/backend/tests/test_3415_bucket_done_only_if_phases_ok.py b/tradein-mvp/backend/tests/test_3415_bucket_done_only_if_phases_ok.py new file mode 100644 index 00000000..f74aaaa9 --- /dev/null +++ b/tradein-mvp/backend/tests/test_3415_bucket_done_only_if_phases_ok.py @@ -0,0 +1,271 @@ +"""Бакет уходит в чекпоинт только если ВСЕ его фазы что-то дали (#3415). + +Прод-факт. `cian_city_sweep` 6179 (06.09) — статус `failed` с причиной +«phase-honest-status: фаза 'houses' отказала полностью — 30 из 30 попыток +неудачны, обогащено 0», а `counters.done_buckets` при этом несёт ВСЕ пять якорей +(«Академический», «Пионерский», «Уралмаш», «Центр», «ЮЗ»). Следующий прогон 6276 +(07.09) резюмировался от него (`resume_from: 6179`, `resume_buckets: 5`), +пропустил все якоря и отчитался `done` с `errors_count=0`, `houses_attempted=0`, +`lots_fetched=0` — зелёный по построению: фаза не исполнялась. Из-за этого фикс +#3396 (houses через пул прокси) простоял на проде с 06.09 и ни разу не отработал. + +ИНВАРИАНТ. Отказ ЦЕЛОЙ фазы внутри якоря — это «якорь пройден не был», ровно как +исключение или таймаут (#3074/#3319). Разница только в том, что отказ фазы виден +не потоком управления, а бухгалтерией счётчиков: `houses_attempted` попыток, +`houses_failed` отказов, обогащено ноль. Порог здесь — одна попытка, а не три, +как у run-level `_phase_totally_failed` (#2700): тот СТАВИТ ДИАГНОЗ прогону и +обязан отсеивать шум, а этот решает «собирать ли якорь заново», и цена ошибок +несимметрична — лишний повтор якоря стоит нескольких запросов, а ложная отметка +«пройден» теряет часть города навсегда и молча. +""" + +from __future__ import annotations + +import os + +# Settings собирается автофикстурой conftest'а и требует database_url. +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +import json +import types +from typing import Any +from unittest.mock import MagicMock, patch + +import pytest + +ANCHOR_A = (56.83, 60.60, "ekb-center") +ANCHOR_B = (56.79, 60.63, "ekb-south") + +# Два дома на якорь — НИЖЕ порога run-level правила (_PHASE_MIN_ATTEMPTS = 3): +# гейт бакета обязан сработать на своей, более строгой границе. +HOUSE_ROWS = [ + {"id": 101, "cian_zhk_url": "https://cian.ru/zhk/101"}, + {"id": 102, "cian_zhk_url": "https://cian.ru/zhk/102"}, +] +DETAIL_ROWS = [ + {"id": 201, "source_url": "/ekb/kvartiry/201"}, + {"id": 202, "source_url": "/ekb/kvartiry/202"}, +] + + +class _FakeDb: + def __init__(self, prev_counters: dict[str, Any] | None = None) -> None: + self.prev_counters = prev_counters or {} + self.heartbeats: list[dict[str, Any]] = [] + + def execute(self, _stmt: Any, params: dict[str, Any] | None = None) -> Any: + if params and "counters" in params: + self.heartbeats.append(json.loads(params["counters"])) + return MagicMock() + if params and "rid" in params: + return MagicMock(fetchone=lambda: types.SimpleNamespace(counters=self.prev_counters)) + if params and "ids" in params: + # houses-фаза: дома, у которых есть cian_zhk_url. + return MagicMock(mappings=lambda: MagicMock(all=lambda: HOUSE_ROWS)) + if params and "limit" in params: + # detail-фаза avito: карточки-кандидаты на обогащение. + return MagicMock(mappings=lambda: MagicMock(all=lambda: DETAIL_ROWS)) + return MagicMock() + + def commit(self) -> None: ... + def rollback(self) -> None: ... + + +def _lot() -> types.SimpleNamespace: + """Новостроечный лот с привязкой к ЖК — только такой доходит до houses-фазы.""" + return types.SimpleNamespace( + listing_segment="novostroyki", + house_source="cian_newbuilding", + house_ext_id="nb-1", + ) + + +class _FakeScraper: + """Двойник CianScraper: помнит, за какими якорями реально ходили.""" + + visited: list[str] = [] # noqa: RUF012 — тестовый сборник + + def __init__(self, *_a: Any, **_kw: Any) -> None: + self.state_extraction_attempts = 1 + self.state_extraction_failures = 0 + self.request_delay_sec = 0.0 + + async def __aenter__(self) -> _FakeScraper: + return self + + async def __aexit__(self, *_e: Any) -> None: + return None + + async def fetch_around_multi_room( + self, lat: float, lon: float, *_a: Any, **_kw: Any + ) -> list[types.SimpleNamespace]: + _FakeScraper.visited.append(f"{lat},{lon}") + return [_lot()] + + +def _config() -> types.SimpleNamespace: + return types.SimpleNamespace( + scraper_proxy_url=None, + scraper_fetch_mode="cffi", + use_proxy_pool_browser=False, + browser_http_endpoint=None, + environment="test", + ) + + +async def _run( + prev: dict[str, Any] | None, *, houses_ok: bool, run_id: int = 8401 +) -> tuple[_FakeDb, Any]: + from scraper_kit.orchestration import pipeline as pl + + _FakeScraper.visited = [] + db = _FakeDb(prev) + + async def _fake_newbuilding(*_a: Any, **_kw: Any) -> Any: + # None — ровно тот отказ, что был на проде 06.09: исключения нет, + # счётчик houses_failed растёт, фаза возвращается штатно. + return object() if houses_ok else None + + with ( + patch.object(pl, "CianScraper", _FakeScraper), + patch.object(pl, "save_listings", lambda *_a, **_kw: (0, 0)), + patch.object(pl, "fetch_newbuilding", _fake_newbuilding), + patch.object(pl, "save_newbuilding_enrichment", lambda *_a, **_kw: None), + patch.object(pl.runs, "is_cancelled", lambda *_a: False), + ): + counters = await pl.run_cian_city_sweep( + db, # type: ignore[arg-type] + run_id=run_id, + config=_config(), + matcher=MagicMock(), + anchors=[ANCHOR_A, ANCHOR_B], + enrich_houses=True, + detail_top_n=0, + request_delay_sec=0.0, + resume_run_id=(run_id - 1) if prev is not None else None, + ) + return db, counters + + +def _last_checkpoint(db: _FakeDb) -> list[str]: + with_ckpt = [hb for hb in db.heartbeats if "done_buckets" in hb] + assert with_ckpt, "ни один heartbeat не унёс done_buckets — чекпоинт не персистится" + return with_ckpt[-1]["done_buckets"] + + +@pytest.mark.asyncio +async def test_anchor_with_totally_failed_houses_is_not_checkpointed() -> None: + """(а) Якорь, где houses отказала на ВСЕХ домах, не попадает в чекпоинт.""" + db, counters = await _run(None, houses_ok=False) + assert counters.houses_attempted > 0, "houses-фаза вообще не исполнялась — тест не о том" + assert counters.houses_enriched == 0, "houses-фаза что-то обогатила — отказ не полный" + assert _last_checkpoint(db) == [], ( + "якорь с полностью отказавшей houses-фазой помечен пройденным — " + f"чекпоинт {_last_checkpoint(db)}" + ) + + +@pytest.mark.asyncio +async def test_next_run_re_enters_houses() -> None: + """(б) Следующий прогон с этим чекпоинтом снова заходит в houses.""" + db1, _ = await _run(None, houses_ok=False) + _db2, counters2 = await _run( + {"done_buckets": _last_checkpoint(db1)}, houses_ok=True, run_id=8402 + ) + assert counters2.houses_attempted > 0, ( + "прогон-наследник пропустил houses-фазу: чекпоинт предшественника " + "объявил якоря пройденными, хотя фаза в них не дала ничего" + ) + assert counters2.houses_enriched > 0, "houses-фаза исполнилась, но ничего не обогатила" + + +@pytest.mark.asyncio +async def test_healthy_anchor_is_still_checkpointed() -> None: + """(в) Контроль: якорь, где все фазы отработали, помечается как раньше.""" + db, counters = await _run(None, houses_ok=True) + assert counters.houses_enriched > 0 + assert _last_checkpoint(db) == ["ekb-center", "ekb-south"], ( + "перестали помечать пройденные якоря вовсе — резюм (#3074) сломан" + ) + + +class _FakeAvitoScraper: + """Двойник AvitoScraper: SERP отдаёт лоты, дальше работает detail-фаза.""" + + def __init__(self, *_a: Any, **_kw: Any) -> None: + self._browser = None + self._cffi = None + + async def fetch_around(self, *_a: Any, **_kw: Any) -> list[types.SimpleNamespace]: + return [types.SimpleNamespace(house_url=None)] + + +class _FakeAsyncSession: + def __init__(self, *_a: Any, **_kw: Any) -> None: ... + + async def __aenter__(self) -> _FakeAsyncSession: + return self + + async def __aexit__(self, *_e: Any) -> None: + return None + + +@pytest.mark.asyncio +async def test_avito_anchor_with_totally_failed_detail_is_not_checkpointed() -> None: + """Сосед: у avito-свипа тот же гейт — фаза без единой удачи не даёт отметки. + + Кода общего у свипов нет (две отдельные функции), общий только гейт. Прод-следа + у avito за 45 суток нет (0 прогонов против 4 у циана) — тест сторожит вторую + копию проводки, а не найденный в данных дефект. + """ + from scraper_kit.orchestration import pipeline as pl + + db = _FakeDb() + + async def _boom(*_a: Any, **_kw: Any) -> Any: + raise RuntimeError("detail отдал 403") + + with ( + patch.object(pl, "AvitoScraper", _FakeAvitoScraper), + patch.object(pl, "AsyncSession", _FakeAsyncSession), + patch.object(pl, "save_listings", lambda *_a, **_kw: (0, 0)), + patch.object(pl, "fetch_detail", _boom), + patch.object(pl.runs, "is_cancelled", lambda *_a: False), + ): + counters = await pl.run_avito_city_sweep( + db, # type: ignore[arg-type] + run_id=8404, + config=types.SimpleNamespace( + scraper_fetch_mode="cffi", + scraper_proxy_url=None, + use_proxy_pool_browser=False, + browser_http_endpoint=None, + environment="test", + avito_serp_ok_not_banned=True, + ), + matcher=MagicMock(), + enrichment=MagicMock(), + anchors=[ANCHOR_A], + enrich_houses=False, + enrich_imv=False, + detail_top_n=len(DETAIL_ROWS), + request_delay_sec=0.0, + ) + + assert counters.detail_attempted == len(DETAIL_ROWS) + assert counters.detail_enriched == 0 + assert _last_checkpoint(db) == [], ( + f"якорь с полностью отказавшей detail-фазой помечен пройденным: {_last_checkpoint(db)}" + ) + + +@pytest.mark.asyncio +async def test_old_format_checkpoint_still_resumes() -> None: + """(г) Чекпоинт СТАРОГО формата (плоский список имён якорей) читается как раньше.""" + db, _ = await _run({"done_buckets": ["ekb-center"]}, houses_ok=True, run_id=8403) + assert f"{ANCHOR_A[0]},{ANCHOR_A[1]}" not in _FakeScraper.visited, ( + "якорь из унаследованного чекпоинта всё-таки опрашивали" + ) + assert _last_checkpoint(db) == ["ekb-center", "ekb-south"], ( + "унаследованный якорь потерян — формат чекпоинта разъехался со старыми прогонами" + ) diff --git a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py index dcee421d..af24b9ed 100644 --- a/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py +++ b/tradein-mvp/packages/scraper-kit/src/scraper_kit/orchestration/pipeline.py @@ -36,7 +36,7 @@ from __future__ import annotations import asyncio import logging import random -from collections.abc import Callable +from collections.abc import Callable, Mapping from contextlib import AsyncExitStack from dataclasses import dataclass, field, fields from datetime import date, timedelta @@ -1100,6 +1100,43 @@ async def run_avito_pipeline( await browser_fetcher.__aexit__(None, None, None) +def _bucket_phase_totally_failed(before: Mapping[str, int], after: Mapping[str, int]) -> str | None: + """Фаза бакета, где отказала КАЖДАЯ попытка (#3415). Имя фазы или None. + + Считает ПРИРОСТ счётчиков за один бакет (якорь): counters у свипа run-level, + так что «фаза этого якоря» существует только как разница до/после. + + Прод-факт, ради которого гейт заведён: `cian_city_sweep` 6179 (06.09) — + `houses_attempted=30, houses_failed=30, houses_enriched=0`, статус `failed` по + run-level правилу (`_phase_totally_failed`, #2700), а в `done_buckets` записаны + все пять якорей. Прогон 6276 (07.09) унаследовал точку, пропустил все якоря и + отчитался `done` с `houses_attempted=0` — зелёный по построению, потому что фаза + не исполнялась. Фикс #3396 (houses через пул прокси) простоял на проде сутки, ни + разу не отработав. + + Пары ищутся В САМИХ counters (ключ `X_attempted` со спутником `X_failed`) — тот же + приём и та же причина, что у `runs._phase_totally_failed`: зашитый список фаз это + ровно то место, куда забывают дописать новую. Фазы без счётчика попыток гейт НЕ + видит — у avito houses роль attempted играет `unique_houses` (нет `houses_attempted`), + и притягивать её сюда значило бы блокировать отметку якоря из-за ОДНОГО упавшего + дома из десяти. + + Порог — одна попытка, а не три, как у run-level правила: тот ставит прогону + диагноз и обязан отсеивать шум, а этот решает «собирать ли якорь заново». Цена + ошибок несимметрична: лишний повтор якоря стоит нескольких запросов, ложная + отметка «пройден» теряет часть города навсегда и молча (прогон-то `done`). + """ + for key in sorted(after): + if not key.endswith("_attempted"): + continue + phase = key[: -len("_attempted")] + attempted = after[key] - before.get(key, 0) + failed = after.get(f"{phase}_failed", 0) - before.get(f"{phase}_failed", 0) + if attempted >= 1 and failed >= attempted: + return phase + return None + + @dataclass class CitySweepCounters: """Aggregate counters для full city sweep run.""" @@ -1349,6 +1386,12 @@ async def run_avito_city_sweep( # #3074: сбрасывается на КАЖДОЙ итерации — иначе один упавший якорь # заразил бы все последующие, и чекпоинт не пополнялся бы вовсе. _anchor_ok = True + # #3415: снимок счётчиков ДО якоря — по разнице видно, дала ли + # каждая фаза ЭТОГО якоря хоть одну удачную попытку (см. + # _bucket_phase_totally_failed). Тот же дефект, что у cian-свипа: + # detail-фаза, отказавшая на всех карточках без исключения, доходила + # до отметки «якорь пройден» наравне с целым якорем. + _before_anchor = counters.to_dict() # Capture loop variables in default args (B023): prevents stale binding # if the coroutine is scheduled after the loop variable changes. @@ -1854,6 +1897,22 @@ async def run_avito_city_sweep( # его как пройденный значило бы, что следующий прогон его пропустит и # объявления оттуда не соберутся НИКОГДА, причём молча: прогон # завершится штатно. Тот же инвариант, что у combo в yandex-свипе. + # + # #3415: флага мало — фаза отказывает и БЕЗ исключения (каждая + # detail-карточка отдала 403, счётчик вырос, корутина завершилась + # штатно). Замер на проде за 45 суток: у avito-свипов такой якорь + # пока не встречался (0 прогонов), у cian — 4; гейт стоит здесь + # потому, что дефект один и тот же, а не по следу в данных. + _failed_phase = _bucket_phase_totally_failed(_before_anchor, counters.to_dict()) + if _failed_phase is not None: + logger.warning( + "city-sweep run_id=%d: якорь %s НЕ помечен пройденным — фаза '%s' " + "не дала ни одной удачной попытки; следующий прогон соберёт его заново", + run_id, + name, + _failed_phase, + ) + _anchor_ok = False if _anchor_ok: _done_anchors.add(name) runs.update_heartbeat(db, run_id, _ckpt()) @@ -3028,6 +3087,10 @@ async def run_cian_city_sweep( ) # ── Per-anchor phases под watchdog-таймаутом ──────────────────── + # #3415: counters у свипа run-level — снимок ДО якоря даёт бухгалтерию + # его собственных фаз (разницей), по которой ниже решается, считать ли + # якорь пройденным. + _before_anchor = counters.to_dict() anchor_lots: list[ScrapedLot] = [] _c_lat, _c_lon, _c_name = lat, lon, name @@ -3243,6 +3306,12 @@ async def run_cian_city_sweep( len(nb_id_list), exc, ) + # #3415: попытки считаем ВМЕСТЕ с отказами. Раньше здесь рос + # только `houses_failed`, и пара получалась несуществующей + # (attempted=0, failed=N): и run-level `_phase_totally_failed` + # (#2700), и гейт бакета сверяют `failed` с `attempted`, так что + # полный отказ фазы на этой ветке не видел ни один из них. + counters.houses_attempted += len(nb_id_list) counters.houses_failed += len(nb_id_list) try: db.rollback() @@ -3434,7 +3503,28 @@ async def run_cian_city_sweep( # Записать упавший якорь пройденным значило бы, что следующий прогон # пропустит его навсегда — молча, потому что прогон завершится # штатно, просто часть города не соберётся. - _done_anchors.add(name) + # + # #3415: потока управления мало. Фаза внутри якоря отказывает БЕЗ + # исключения — `fetch_newbuilding` возвращает None, счётчик растёт, + # `_cian_anchor_phases` завершается штатно, — и такой якорь доходил + # сюда наравне с целым. Дальше ровно тот же исход, что и у записанного + # упавшего: следующий прогон пропускает якорь и рапортует `done` с + # нулевой фазой. Форма отметки прежняя (плоский список имён): чекпоинт + # читают ещё три свипа и `scheduler._resume_decision`, а ключ + # `bucket:phase` потребовал бы новых читателей ради того же решения. + _failed_phase = _bucket_phase_totally_failed(_before_anchor, counters.to_dict()) + if _failed_phase is None: + _done_anchors.add(name) + else: + logger.warning( + "cian-sweep run_id=%d: якорь %s НЕ помечен пройденным — фаза '%s' " + "не дала ни одной удачной попытки; следующий прогон соберёт его заново", + run_id, + name, + _failed_phase, + ) + # Heartbeat пишем в любом случае: без него reap_zombies посчитает живой + # прогон мёртвым, а jsonb-мерж сохранит унаследованную часть точки. runs.update_heartbeat( db, run_id, {**counters.to_dict(), "done_buckets": sorted(_done_anchors)} )