diff --git a/tradein-mvp/backend/app/tasks/yandex_detail_backfill.py b/tradein-mvp/backend/app/tasks/yandex_detail_backfill.py index 65bf213c..dd614bec 100644 --- a/tradein-mvp/backend/app/tasks/yandex_detail_backfill.py +++ b/tradein-mvp/backend/app/tasks/yandex_detail_backfill.py @@ -8,12 +8,19 @@ Yandex city sweep does not call YandexDetailScraper — SERP data only. area_m2 coverage 23%, living/kitchen 0%, repair_state 1% on prod. Solution: single snapshot SELECT at start (guarantees termination), fetch each -offer detail page via YandexDetailScraper (httpx + BaseScraper retry), persist -with save_detail_enrichment. No explicit block-exception from YandexDetailScraper — -fetch_detail returns None on HTTP error or parse failure. Track consecutive None -results; abort after max_consecutive_blocks (mark_done, not mark_failed — retry next -night via NULL detail_enriched_at). YandexDetailScraper has no _rotate_ip: plain -httpx with UA rotation; no proxy layer required. +offer detail page via curl_cffi AsyncSession (chrome120 + proxy) — mirrors +yandex_address_backfill.py which already gets full HTML from Yandex on prod. +Parse HTML via YandexDetailScraper.parse (pure, no network). Persist via +save_detail_enrichment. Track consecutive parse→None results; abort after +max_consecutive_blocks (mark_done, not mark_failed — retry next night via +NULL detail_enriched_at). + +Why curl_cffi and not YandexDetailScraper.fetch_detail: + fetch_detail uses BaseScraper._http_get (plain httpx, no proxy, no TLS + fingerprinting). On datacenter IPs Yandex returns captcha / shell-HTML + → parse always returns None → backfill would be 0% effective. The + curl_cffi path (chrome120 impersonation + mobile proxy) is already proven + by yandex_address_backfill, which fetches identical offer detail pages. """ from __future__ import annotations @@ -23,9 +30,11 @@ import logging import time from dataclasses import dataclass, field +from curl_cffi.requests import AsyncSession from sqlalchemy import text from sqlalchemy.orm import Session +from app.core.config import settings from app.services import scrape_runs as runs_mod from app.services.scrapers.yandex_detail import YandexDetailScraper, save_detail_enrichment @@ -61,21 +70,22 @@ async def run_yandex_detail_backfill( run_id: int, params: dict, ) -> YandexDetailBackfillResult: - """Backfill detail_enriched_at for legacy yandex listings via YandexDetailScraper. + """Backfill detail_enriched_at for legacy yandex listings via curl_cffi + parse. Params (from default_params jsonb in scrape_schedules): batch_size: int -- snapshot size (SELECT LIMIT), default 800. budget_sec: float -- wall-clock budget per run, default 3600s. - request_delay_sec: float -- delay between listings (overrides scraper default), default 5s. - max_consecutive_blocks: int -- consecutive None results before abort, default 5. + request_delay_sec: float -- delay between listings, default 5s. + max_consecutive_blocks: int -- consecutive parse→None before abort, default 5. + + Fetch mechanism: + curl_cffi AsyncSession(impersonate="chrome120") + scraper_proxy_url — mirrors + yandex_address_backfill. On HTTP 200: pass resp.text to + YandexDetailScraper().parse(html, offer_url). parse→None counts as a fail + (possible captcha wall); consecutive None → abort after max_consecutive_blocks. Lifecycle: update_heartbeat -> snapshot -> loop with budget guard -> mark_done (incl. partial / consecutive-None abort) / mark_failed (exception only). - - YandexDetailScraper.fetch_detail returns None on HTTP error or parse failure - (no explicit block exception). Consecutive None results signal a possible - temporary ban or network issue; we abort after max_consecutive_blocks and let - the scheduler retry the next night (NULL detail_enriched_at rows reappear). """ batch_size = int(params.get("batch_size", 800)) budget_sec = float(params.get("budget_sec", 3600)) @@ -130,10 +140,22 @@ async def run_yandex_detail_backfill( max_consecutive_blocks, ) + # Build proxies dict once — mirrors yandex_address_backfill.py + _proxy = settings.scraper_proxy_url + _proxies = {"http": _proxy, "https": _proxy} if _proxy else None + consecutive_none = 0 do_sleep = False + scraper = YandexDetailScraper() - async with YandexDetailScraper() as scraper: + async with AsyncSession( + impersonate="chrome120", + timeout=30.0, + proxies=_proxies, + headers={ + "Accept-Language": "ru-RU,ru;q=0.9,en;q=0.8", + }, + ) as session: for idx, row in enumerate(snapshot): # Budget guard elapsed = time.monotonic() - start @@ -159,16 +181,64 @@ async def run_yandex_detail_backfill( counters.attempted += 1 try: - enrichment = await scraper.fetch_detail(source_url) + try: + resp = await session.get(source_url, allow_redirects=True) + except Exception as fetch_exc: + consecutive_none += 1 + counters.failed += 1 + logger.warning( + "yandex_detail_backfill: run_id=%d listing_id=%d " + "fetch error (consecutive=%d): %s", + run_id, + listing_id, + consecutive_none, + fetch_exc, + ) + if consecutive_none >= max_consecutive_blocks: + logger.error( + "yandex_detail_backfill: run_id=%d ABORT -- %d consecutive " + "errors. enriched=%d attempted=%d", + run_id, + consecutive_none, + counters.enriched, + counters.attempted, + ) + break + continue + + if resp.status_code != 200: + consecutive_none += 1 + counters.failed += 1 + logger.warning( + "yandex_detail_backfill: run_id=%d listing_id=%d " + "HTTP %d (consecutive=%d)", + run_id, + listing_id, + resp.status_code, + consecutive_none, + ) + if consecutive_none >= max_consecutive_blocks: + logger.error( + "yandex_detail_backfill: run_id=%d ABORT -- %d consecutive " + "non-200 responses. enriched=%d attempted=%d", + run_id, + consecutive_none, + counters.enriched, + counters.attempted, + ) + break + continue + + enrichment = scraper.parse(resp.text, offer_url=source_url) if enrichment is None: - # fetch_detail returns None on block / parse failure. + # parse→None: captcha wall / shell-HTML / no JSON-LD. # Do not mark listing as done — retry next night. consecutive_none += 1 counters.failed += 1 logger.warning( "yandex_detail_backfill: run_id=%d listing_id=%d source_url=%s " - "-> None (consecutive=%d)", + "-> parse None (consecutive=%d)", run_id, listing_id, source_url, @@ -177,7 +247,7 @@ async def run_yandex_detail_backfill( if consecutive_none >= max_consecutive_blocks: logger.error( "yandex_detail_backfill: run_id=%d ABORT -- %d consecutive " - "None results (possible ban). enriched=%d attempted=%d", + "parse-None results (captcha wall?). enriched=%d attempted=%d", run_id, consecutive_none, counters.enriched, diff --git a/tradein-mvp/backend/tests/tasks/test_yandex_detail_backfill.py b/tradein-mvp/backend/tests/tasks/test_yandex_detail_backfill.py index 3b285ad1..c801ea2e 100644 --- a/tradein-mvp/backend/tests/tasks/test_yandex_detail_backfill.py +++ b/tradein-mvp/backend/tests/tasks/test_yandex_detail_backfill.py @@ -1,4 +1,8 @@ -"""Tests for app.tasks.yandex_detail_backfill and save_detail_enrichment.""" +"""Tests for app.tasks.yandex_detail_backfill. + +Fetch mechanism changed: curl_cffi AsyncSession(chrome120+proxy) + YandexDetailScraper.parse +(instead of old fetch_detail path). Mocks target session.get and scraper.parse. +""" from __future__ import annotations @@ -23,10 +27,12 @@ from app.tasks.yandex_detail_backfill import ( # noqa: E402 # Path constants for patching # --------------------------------------------------------------------------- -_FETCH = "app.tasks.yandex_detail_backfill.YandexDetailScraper" +_ASYNC_SESSION = "app.tasks.yandex_detail_backfill.AsyncSession" +_PARSE = "app.tasks.yandex_detail_backfill.YandexDetailScraper.parse" _SAVE = "app.tasks.yandex_detail_backfill.save_detail_enrichment" _RUNS = "app.tasks.yandex_detail_backfill.runs_mod" _SLEEP = "app.tasks.yandex_detail_backfill.asyncio.sleep" +_SETTINGS = "app.tasks.yandex_detail_backfill.settings" # --------------------------------------------------------------------------- # Helpers @@ -48,6 +54,33 @@ def _mock_db(snapshot: list[dict]) -> MagicMock: return db +def _make_resp(status: int = 200, text: str = "ok") -> MagicMock: + """Fake curl_cffi response.""" + resp = MagicMock() + resp.status_code = status + resp.text = text + return resp + + +def _make_session_ctx(get_side_effect) -> MagicMock: + """Build AsyncSession context-manager mock with session.get side_effect.""" + session = AsyncMock() + session.get = AsyncMock(side_effect=get_side_effect) + + ctx = MagicMock() + ctx.__aenter__ = AsyncMock(return_value=session) + ctx.__aexit__ = AsyncMock(return_value=None) + + session_cls = MagicMock(return_value=ctx) + return session_cls, session + + +def _mock_settings(proxy: str | None = "http://proxy:3128") -> MagicMock: + s = MagicMock() + s.scraper_proxy_url = proxy + return s + + # --------------------------------------------------------------------------- # Tests: run_yandex_detail_backfill # --------------------------------------------------------------------------- @@ -55,17 +88,15 @@ def _mock_db(snapshot: list[dict]) -> MagicMock: @pytest.mark.asyncio async def test_backfill_empty_snapshot_marks_done() -> None: - """Empty snapshot -> mark_done immediately, no fetch calls.""" + """Empty snapshot -> mark_done immediately, no session.get calls.""" db = _mock_db([]) runs = MagicMock() - mock_scraper_cls = MagicMock() - mock_scraper_inst = AsyncMock() - mock_scraper_cls.return_value.__aenter__ = AsyncMock(return_value=mock_scraper_inst) - mock_scraper_cls.return_value.__aexit__ = AsyncMock(return_value=None) + session_cls, session = _make_session_ctx([]) with ( - patch(_FETCH, mock_scraper_cls), + patch(_ASYNC_SESSION, session_cls), patch(_RUNS, runs), + patch(_SETTINGS, _mock_settings()), ): result = await run_yandex_detail_backfill( db, run_id=1, params={"batch_size": 10, "budget_sec": 60} @@ -74,30 +105,29 @@ async def test_backfill_empty_snapshot_marks_done() -> None: assert isinstance(result, YandexDetailBackfillResult) assert result.attempted == 0 assert result.enriched == 0 - mock_scraper_inst.fetch_detail.assert_not_called() + session.get.assert_not_called() runs.mark_done.assert_called_once() runs.mark_failed.assert_not_called() @pytest.mark.asyncio async def test_backfill_processes_snapshot_to_completion() -> None: - """3 listings -> all fetched and enriched, mark_done called.""" + """3 listings -> all fetched (200) + parse success -> enriched=3, mark_done.""" snapshot = _make_snapshot(3) db = _mock_db(snapshot) runs = MagicMock() mock_enrichment = MagicMock() - mock_scraper_cls = MagicMock() - mock_scraper_inst = AsyncMock() - mock_scraper_inst.fetch_detail = AsyncMock(return_value=mock_enrichment) - mock_scraper_cls.return_value.__aenter__ = AsyncMock(return_value=mock_scraper_inst) - mock_scraper_cls.return_value.__aexit__ = AsyncMock(return_value=None) + responses = [_make_resp(200), _make_resp(200), _make_resp(200)] + session_cls, session = _make_session_ctx(responses) with ( - patch(_FETCH, mock_scraper_cls), + patch(_ASYNC_SESSION, session_cls), + patch(_PARSE, return_value=mock_enrichment), patch(_RUNS, runs), patch(_SAVE, return_value=True), patch(_SLEEP, new_callable=AsyncMock), + patch(_SETTINGS, _mock_settings()), ): result = await run_yandex_detail_backfill( db, run_id=2, params={"batch_size": 10, "budget_sec": 3600} @@ -106,28 +136,27 @@ async def test_backfill_processes_snapshot_to_completion() -> None: assert result.attempted == 3 assert result.enriched == 3 assert result.failed == 0 - assert mock_scraper_inst.fetch_detail.call_count == 3 + assert session.get.call_count == 3 runs.mark_done.assert_called_once() runs.mark_failed.assert_not_called() @pytest.mark.asyncio -async def test_backfill_none_abort_after_max_consecutive() -> None: - """5 consecutive None results -> abort, mark_done (NOT mark_failed).""" +async def test_backfill_parse_none_abort_after_max_consecutive() -> None: + """5 consecutive parse→None results -> abort, mark_done (NOT mark_failed).""" snapshot = _make_snapshot(10) db = _mock_db(snapshot) runs = MagicMock() - mock_scraper_cls = MagicMock() - mock_scraper_inst = AsyncMock() - mock_scraper_inst.fetch_detail = AsyncMock(return_value=None) - mock_scraper_cls.return_value.__aenter__ = AsyncMock(return_value=mock_scraper_inst) - mock_scraper_cls.return_value.__aexit__ = AsyncMock(return_value=None) + responses = [_make_resp(200)] * 10 + session_cls, _session = _make_session_ctx(responses) with ( - patch(_FETCH, mock_scraper_cls), + patch(_ASYNC_SESSION, session_cls), + patch(_PARSE, return_value=None), patch(_RUNS, runs), patch(_SLEEP, new_callable=AsyncMock), + patch(_SETTINGS, _mock_settings()), ): result = await run_yandex_detail_backfill( db, @@ -143,24 +172,24 @@ async def test_backfill_none_abort_after_max_consecutive() -> None: @pytest.mark.asyncio -async def test_backfill_none_resets_on_success() -> None: - """1 None + 1 success + 1 None -> consecutive resets, no abort after 1 success.""" +async def test_backfill_parse_none_resets_on_success() -> None: + """1 parse→None + 1 success + 1 parse→None -> consecutive resets, no abort.""" snapshot = _make_snapshot(3) db = _mock_db(snapshot) runs = MagicMock() mock_enrichment = MagicMock() - mock_scraper_cls = MagicMock() - mock_scraper_inst = AsyncMock() - mock_scraper_inst.fetch_detail = AsyncMock(side_effect=[None, mock_enrichment, None]) - mock_scraper_cls.return_value.__aenter__ = AsyncMock(return_value=mock_scraper_inst) - mock_scraper_cls.return_value.__aexit__ = AsyncMock(return_value=None) + responses = [_make_resp(200)] * 3 + session_cls, _session = _make_session_ctx(responses) + parse_results = [None, mock_enrichment, None] with ( - patch(_FETCH, mock_scraper_cls), + patch(_ASYNC_SESSION, session_cls), + patch(_PARSE, side_effect=parse_results), patch(_RUNS, runs), patch(_SAVE, return_value=True), patch(_SLEEP, new_callable=AsyncMock), + patch(_SETTINGS, _mock_settings()), ): result = await run_yandex_detail_backfill( db, @@ -174,27 +203,55 @@ async def test_backfill_none_resets_on_success() -> None: runs.mark_done.assert_called_once() +@pytest.mark.asyncio +async def test_backfill_non200_counts_as_fail_and_aborts() -> None: + """5 consecutive HTTP 403 responses -> abort after max_consecutive_blocks.""" + snapshot = _make_snapshot(10) + db = _mock_db(snapshot) + runs = MagicMock() + + responses = [_make_resp(403)] * 10 + session_cls, _session = _make_session_ctx(responses) + + with ( + patch(_ASYNC_SESSION, session_cls), + patch(_PARSE, return_value=MagicMock()), + patch(_RUNS, runs), + patch(_SLEEP, new_callable=AsyncMock), + patch(_SETTINGS, _mock_settings()), + ): + result = await run_yandex_detail_backfill( + db, + run_id=5, + params={"batch_size": 10, "budget_sec": 3600, "max_consecutive_blocks": 5}, + ) + + assert result.failed == 5 + assert result.enriched == 0 + runs.mark_done.assert_called_once() + runs.mark_failed.assert_not_called() + + @pytest.mark.asyncio async def test_backfill_budget_guard_stops_loop() -> None: - """Budget expired before first listing -> fetch_detail not called.""" + """Budget expired before first listing -> session.get not called.""" snapshot = _make_snapshot(5) db = _mock_db(snapshot) runs = MagicMock() - mock_scraper_cls = MagicMock() - mock_scraper_inst = AsyncMock() - mock_scraper_cls.return_value.__aenter__ = AsyncMock(return_value=mock_scraper_inst) - mock_scraper_cls.return_value.__aexit__ = AsyncMock(return_value=None) + responses = [_make_resp(200)] * 5 + session_cls, session = _make_session_ctx(responses) mono_values = iter([0.0, 999.0, 999.0]) with ( - patch(_FETCH, mock_scraper_cls), + patch(_ASYNC_SESSION, session_cls), patch(_RUNS, runs), patch("app.tasks.yandex_detail_backfill.time.monotonic", side_effect=mono_values), + patch(_SETTINGS, _mock_settings()), ): - await run_yandex_detail_backfill(db, run_id=5, params={"batch_size": 5, "budget_sec": 1}) + await run_yandex_detail_backfill(db, run_id=6, params={"batch_size": 5, "budget_sec": 1}) - mock_scraper_inst.fetch_detail.assert_not_called() + session.get.assert_not_called() runs.mark_done.assert_called_once() @@ -207,10 +264,11 @@ async def test_backfill_top_level_exception_marks_failed() -> None: with ( patch(_RUNS, runs), + patch(_SETTINGS, _mock_settings()), ): with pytest.raises(RuntimeError, match="DB connection lost"): await run_yandex_detail_backfill( - db, run_id=6, params={"batch_size": 5, "budget_sec": 60} + db, run_id=7, params={"batch_size": 5, "budget_sec": 60} ) runs.mark_failed.assert_called_once() @@ -219,39 +277,67 @@ async def test_backfill_top_level_exception_marks_failed() -> None: @pytest.mark.asyncio async def test_backfill_fetch_exception_continues() -> None: - """RuntimeError on one listing -> failed++, db.rollback(), loop continues for next.""" + """session.get raises on first listing -> failed++, loop continues for second.""" snapshot = _make_snapshot(2) db = _mock_db(snapshot) runs = MagicMock() mock_enrichment = MagicMock() - mock_scraper_cls = MagicMock() - mock_scraper_inst = AsyncMock() - mock_scraper_inst.fetch_detail = AsyncMock( - side_effect=[RuntimeError("parse error"), mock_enrichment] - ) - mock_scraper_cls.return_value.__aenter__ = AsyncMock(return_value=mock_scraper_inst) - mock_scraper_cls.return_value.__aexit__ = AsyncMock(return_value=None) + # First call raises, second returns 200 OK + responses = [RuntimeError("connection reset"), _make_resp(200)] + session_cls, _session = _make_session_ctx(responses) with ( - patch(_FETCH, mock_scraper_cls), + patch(_ASYNC_SESSION, session_cls), + patch(_PARSE, return_value=mock_enrichment), patch(_RUNS, runs), patch(_SAVE, return_value=True), patch(_SLEEP, new_callable=AsyncMock), + patch(_SETTINGS, _mock_settings()), ): result = await run_yandex_detail_backfill( - db, run_id=7, params={"batch_size": 10, "budget_sec": 3600} + db, run_id=8, params={"batch_size": 10, "budget_sec": 3600} ) assert result.failed == 1 assert result.enriched == 1 assert result.attempted == 2 - db.rollback.assert_called() + # fetch exception is caught by inner try/except (not DB) — no rollback needed + db.rollback.assert_not_called() runs.mark_done.assert_called_once() +@pytest.mark.asyncio +async def test_backfill_no_proxy_when_settings_none() -> None: + """scraper_proxy_url=None -> AsyncSession called with proxies=None.""" + snapshot = _make_snapshot(1) + db = _mock_db(snapshot) + runs = MagicMock() + mock_enrichment = MagicMock() + + responses = [_make_resp(200)] + session_cls, _session = _make_session_ctx(responses) + + with ( + patch(_ASYNC_SESSION, session_cls), + patch(_PARSE, return_value=mock_enrichment), + patch(_RUNS, runs), + patch(_SAVE, return_value=True), + patch(_SLEEP, new_callable=AsyncMock), + patch(_SETTINGS, _mock_settings(proxy=None)), + ): + result = await run_yandex_detail_backfill( + db, run_id=9, params={"batch_size": 10, "budget_sec": 3600} + ) + + assert result.enriched == 1 + # Verify AsyncSession was constructed with proxies=None + call_kwargs = session_cls.call_args[1] + assert call_kwargs.get("proxies") is None + + # --------------------------------------------------------------------------- -# Tests: save_detail_enrichment +# Tests: save_detail_enrichment (unchanged — kept for coverage) # --------------------------------------------------------------------------- @@ -296,21 +382,18 @@ def test_save_detail_enrichment_maps_fields_to_update() -> None: db.execute.assert_called_once() db.commit.assert_called_once() - # Verify the SQL call contains metro_stations and photo_urls as JSON call_args = db.execute.call_args - params = call_args[0][1] # positional second arg to execute() + params = call_args[0][1] assert params["rooms"] == 2 assert params["area_m2"] == 54.5 assert params["agency_name"] == "Агентство «Тест»" - # metro_stations must be JSON-serialised metro_val = params["metro_stations"] assert metro_val is not None parsed_metro = json.loads(metro_val) assert parsed_metro[0]["name"] == "Чкаловская" assert parsed_metro[0]["walk_min"] == 11 - # photo_urls must be JSON-serialised photo_val = params["photo_urls"] assert photo_val is not None parsed_photos = json.loads(photo_val)