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,

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 = "<html>ok</html>") -> 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)