From f8779c0812fc25c35079c023743aa40221328c6a Mon Sep 17 00:00:00 2001 From: lekss361 Date: Wed, 17 Jun 2026 08:20:46 +0000 Subject: [PATCH] feat(yandex): switch SERP scraping to JSON gate-API (bypasses bandwidth tarpit) (#1665) --- .../app/services/scrapers/yandex_realty.py | 1447 ++++++---------- .../backend/tests/test_yandex_realty_serp.py | 1526 +++++++---------- 2 files changed, 1175 insertions(+), 1798 deletions(-) diff --git a/tradein-mvp/backend/app/services/scrapers/yandex_realty.py b/tradein-mvp/backend/app/services/scrapers/yandex_realty.py index e4df5de6..93723356 100644 --- a/tradein-mvp/backend/app/services/scrapers/yandex_realty.py +++ b/tradein-mvp/backend/app/services/scrapers/yandex_realty.py @@ -1,85 +1,58 @@ -"""Yandex.Недвижимость scraper (realty.yandex.ru) — DOM-based SERP parser. +"""Yandex.Nedvizhimost scraper (realty.yandex.ru) -- gate-API JSON parser. -History: Yandex SSR used to embed full state in ` - -
- Открыть -
- улица Тестовая, 10 -
-
{price_fmt} ₽
-
-""" - - -def test_area_from_state_json_no_dom_text(): - """area_m2 extracted from embedded JSON state when DOM text has no area.""" - html = _make_yandex_state_html(offer_id="1234567890", area=63.5) - s = YandexRealtyScraper() - lots = s._parse_html(html, page=0) - assert len(lots) == 1 - assert lots[0].area_m2 == pytest.approx(63.5) - - -def test_area_from_state_json_multiple_offers(): - """State with multiple offers — each card gets its own area from state.""" - state = { - "offers": [ - {"offerId": "111", "totalArea": 42.0}, - {"offerId": "222", "totalArea": 78.3}, - ] - } - state_json = json.dumps(state) - html = f""" - - -
- x -
3 000 000 ₽
-
-
- x -
5 000 000 ₽
-
-""" - s = YandexRealtyScraper() - lots = s._parse_html(html, page=0) - assert len(lots) == 2 - areas = {lot.source_id: lot.area_m2 for lot in lots} - assert areas["111"] == pytest.approx(42.0) - assert areas["222"] == pytest.approx(78.3) - - -def test_area_dom_fallback_when_no_state(): - """When state script absent, area falls back to DOM regex (existing behavior).""" - # SINGLE_CARD_HTML has area in DOM text — state script absent - s = YandexRealtyScraper() - lots = s._parse_html(SINGLE_CARD_HTML, page=0) - assert len(lots) == 1 - assert lots[0].area_m2 == pytest.approx(36.8) - - -def test_extract_offer_areas_from_state_nested_path(): - """State via pageData.offers path (SERP v2 shape).""" - state = { - "pageData": { - "offers": [ - {"offerId": "999", "spaceTotal": 55.0}, - ] - } - } - html = '" - areas = _extract_offer_areas_from_state(html) - assert areas == {"999": 55.0} - - -def test_extract_offer_areas_area_dict_value(): - """area field as nested dict {'value': 48.0, 'unit': 'SQM'} (some Yandex responses).""" - state = { - "offers": [ - {"offerId": "777", "area": {"value": 48.0, "unit": "SQM"}}, - ] - } - html = '" - areas = _extract_offer_areas_from_state(html) - assert areas == {"777": 48.0} - - -def test_extract_offer_areas_empty_when_no_script(): - """No state script → empty dict returned, no crash.""" - html = "

no state here

" - areas = _extract_offer_areas_from_state(html) - assert areas == {} - - -# ── PART B: T4 transport + NBSP fix ───────────────────────────────────────── - - -def test_nbsp_price_fix(): - """NBSP (\xa0) in price tokens must not break parse_rub → price_rub must be non-None. - - Yandex uses \xa0 as thousands separator in prices. Without the replace, parse_rub - returns None because regex can't capture across NBSP — the card is then skipped. - """ - # Price with \xa0 thousands separator (e.g. "4\xa0399\xa0000 ₽") - nbsp_card_html = """ - -
- x -
36,8\xa0м²\xa0·\xa01-комнатная\xa0·\xa08\xa0этаж\xa0из\xa09\xa0·\xa04\xa0399\xa0000\xa0₽
-
-""" - s = YandexRealtyScraper() - # _parse_html applies NBSP-fix before parsing — price must be extracted - lots = s._parse_html(nbsp_card_html, page=0) - assert len(lots) == 1 - assert lots[0].price_rub is not None - assert lots[0].price_rub > 0 - - -def test_is_captcha_detects_showcaptcha(): - """HTML containing /showcaptcha must be flagged as captcha.""" - html = '' - assert _is_captcha(html) is True - - -def test_is_captcha_detects_smartcaptcha(): - """HTML with smartcaptcha JS is flagged.""" - html = "" - assert _is_captcha(html) is True - - -def test_is_captcha_clean_page(): - """Normal SERP page is not flagged as captcha.""" - assert _is_captcha(SINGLE_CARD_HTML) is False - - -def test_is_captcha_robot_text(): - """Russian 'докажите, что вы не робот' message is detected.""" - html = "Докажите, что вы не робот — введите код." - assert _is_captcha(html) is True - - -def test_load_cookies_from_file_missing_path(): - """None or missing path → empty dict, no exception.""" - assert _load_cookies_from_file(None) == {} - assert _load_cookies_from_file("/nonexistent/path.json") == {} - - -def test_load_cookies_from_file_valid(tmp_path): - """Valid Netscape-format cookie JSON → dict with name→value.""" - data = [ - {"name": "yandex_login", "value": "user123", "domain": ".yandex.ru"}, - {"name": "Session_id", "value": "abc456", "domain": ".yandex.ru"}, - {"missing_name_key": "ignored"}, - ] - f = tmp_path / "cookies.json" - f.write_text(json.dumps(data), encoding="utf-8") - result = _load_cookies_from_file(str(f)) - assert result == {"yandex_login": "user123", "Session_id": "abc456"} - - -# ── PART C: T5 combos — _build_url with rooms + price filters ─────────────── - - -def test_build_url_with_rooms(): - """Room slug mapped correctly to Yandex path segment.""" - s = YandexRealtyScraper() - url = s._build_url(page=0, rooms="1") - assert "/odnokomnatnaya/vtorichniy-rynok/" in url - assert "page" not in url - - -def test_build_url_with_rooms_2(): - s = YandexRealtyScraper() - url = s._build_url(page=0, rooms="2") - assert "/dvuhkomnatnaya/vtorichniy-rynok/" in url - - -def test_build_url_with_rooms_studio(): - s = YandexRealtyScraper() - url = s._build_url(page=0, rooms="studio") - assert "/studiya/vtorichniy-rynok/" in url - - -def test_build_url_with_price_range(): - """Price range adds priceMin/priceMax + newFlat=NO_DEAL.""" - s = YandexRealtyScraper() - url = s._build_url(page=0, price_min=5_000_000, price_max=7_000_000) - assert "priceMin=5000000" in url - assert "priceMax=7000000" in url - assert "newFlat=NO_DEAL" in url - # No rooms → base citywide path - assert "/vtorichniy-rynok/" in url - assert "odnokomnatnaya" not in url - - -def test_build_url_rooms_and_price_and_page(): - """All parameters combined: rooms + price + paginated.""" - s = YandexRealtyScraper() - url = s._build_url(page=3, rooms="3", price_min=10_000_000, price_max=15_000_000) - assert "/trehkomnatnaya/vtorichniy-rynok/" in url - assert "priceMin=10000000" in url - assert "priceMax=15000000" in url - assert "newFlat=NO_DEAL" in url - assert "page=3" in url - - -def test_build_url_open_ended_price_min_only(): - """Only priceMin set → no priceMax param but newFlat present.""" - s = YandexRealtyScraper() - url = s._build_url(price_min=25_000_000) - assert "priceMin=25000000" in url - assert "priceMax" not in url - assert "newFlat=NO_DEAL" in url - - -def test_build_url_open_ended_price_max_only(): - """Only priceMax → no priceMin param.""" - s = YandexRealtyScraper() - url = s._build_url(price_max=5_000_000) - assert "priceMax=5000000" in url - assert "priceMin" not in url - assert "newFlat=NO_DEAL" in url - - -def test_build_url_unknown_rooms_falls_back_to_citywide(): - """Unknown/invalid rooms key falls back to citywide (no room path segment).""" - s = YandexRealtyScraper() - url = s._build_url(rooms="99-room-penthouse") - assert "/kupit/kvartira/vtorichniy-rynok/" in url - assert "studiya" not in url - assert "odnokomnatnaya" not in url - - -def test_combos_produce_different_urls(): - """T5: rooms × price-ranges produce distinct URLs (no URL collisions).""" - s = YandexRealtyScraper() - urls = set() - rooms_list = list(ROOM_PATH.keys()) - for rooms in rooms_list: - for lo, hi in DEFAULT_PRICE_RANGES: - urls.add(s._build_url(page=0, rooms=rooms, price_min=lo, price_max=hi)) - expected = len(rooms_list) * len(DEFAULT_PRICE_RANGES) - assert len(urls) == expected, f"URL collisions: {expected} combos but {len(urls)} unique URLs" - - -def test_combo_label_all(): - assert _combo_label(None, None, None) == "all-rooms/any-price" +def test_combo_label_none_none(): + result = _combo_label(None, 0, 200_000_000) + assert isinstance(result, str) + assert "any" in result def test_combo_label_room_and_price(): - assert _combo_label("1", 5_000_000, 7_000_000) == "1/5M-7M" + result = _combo_label("1", 5_000_000, 7_000_000) + assert "1" in result + assert "5000000" in result + assert "7000000" in result -def test_combo_label_open_ended(): - assert _combo_label("4+", 25_000_000, None) == "4+/25M-inf" - assert _combo_label("studio", None, 5_000_000) == "studio/0-5M" +def test_combo_label_different_ranges_differ(): + k1 = _combo_label("1", 5_000_000, 7_000_000) + k2 = _combo_label("1", 7_000_000, 10_000_000) + k3 = _combo_label("2", 5_000_000, 7_000_000) + assert k1 != k2 + assert k1 != k3 -# ── PART D: T4 transport — async context manager + _http_get mock ─────────── - - -@pytest.mark.asyncio -async def test_aenter_creates_cffi_session(): - """__aenter__ creates a _CurlCffiSession (Chrome120 impersonation).""" - mock_settings = MagicMock() - mock_settings.yandex_cookies_file = None - - with patch("app.services.scrapers.yandex_realty._CurlCffiSession") as mock_cls: - mock_session = AsyncMock() - mock_cls.return_value = mock_session - with patch("app.services.scrapers.yandex_realty._load_cookies_from_file", return_value={}): - # Patch settings import inside __aenter__ without triggering Settings() - with patch.dict("sys.modules", {"app.core.config": MagicMock(settings=mock_settings)}): - s = YandexRealtyScraper() - result = await s.__aenter__() - assert result is s - mock_cls.assert_called_once() - call_kwargs = mock_cls.call_args - assert call_kwargs.kwargs.get("impersonate") == "chrome120" +# --------------------------------------------------------------------------- +# PART F: transport layer +# --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_aexit_closes_session(): - """__aexit__ closes the curl_cffi session and sets it to None.""" s = YandexRealtyScraper() mock_session = AsyncMock() s._cffi_session = mock_session @@ -552,53 +401,25 @@ async def test_aexit_closes_session(): @pytest.mark.asyncio async def test_http_get_raises_when_no_context(): - """_http_get must raise RuntimeError when called outside async context manager - when no socks5 proxy is configured (legacy cffi path).""" + """_http_get must raise RuntimeError outside async context manager (cffi path).""" s = YandexRealtyScraper() mock_settings = MagicMock() - mock_settings.yandex_proxy_url = None # no proxy → cffi path → RuntimeError - with patch("app.services.scrapers.yandex_realty.settings", mock_settings, create=True): - with patch.dict("sys.modules", {"app.core.config": MagicMock(settings=mock_settings)}): - with pytest.raises(RuntimeError, match="async context manager"): - await s._http_get("https://realty.yandex.ru/test/") - - -@pytest.mark.asyncio -async def test_fetch_around_returns_empty_on_captcha(): - """fetch_around returns [] and logs warning when captcha detected.""" - s = YandexRealtyScraper() - mock_resp = MagicMock() - mock_resp.status_code = 200 - mock_resp.text = ( - "Докажите, что вы не робот. " - 'пройдите проверку' - ) - s._cffi_session = AsyncMock() - s._cffi_session.get = AsyncMock(return_value=mock_resp) - - mock_settings = MagicMock() - mock_settings.yandex_proxy_url = None # no socks5 → cffi path + mock_settings.yandex_proxy_url = None with patch.dict("sys.modules", {"app.core.config": MagicMock(settings=mock_settings)}): - with patch.object(s, "sleep_between_requests", new_callable=AsyncMock): - lots = await s.fetch_around(lat=0.0, lon=0.0) - assert lots == [] - - -# ── PART G: subprocess curl transport ──────────────────────────────────────── + with pytest.raises(RuntimeError, match="async context manager"): + await s._http_get("https://realty.yandex.ru/test/") @pytest.mark.asyncio async def test_http_get_routes_socks5_to_curl_subprocess(): - """_http_get with a socks5h proxy must call _curl_subprocess_get, - NOT the curl_cffi session.""" s = YandexRealtyScraper() - sentinel = _CurlResponse(status_code=200, text="") + sentinel = _CurlResponse(status_code=200, text="{}") mock_settings = MagicMock() mock_settings.yandex_proxy_url = "socks5h://x:y@h:1" cffi_mock = AsyncMock() - s._cffi_session = cffi_mock # session present — must NOT be called + s._cffi_session = cffi_mock async def fake_curl_subprocess(url: str, proxy: str, timeout: int) -> object: return sentinel @@ -614,36 +435,31 @@ async def test_http_get_routes_socks5_to_curl_subprocess(): @pytest.mark.asyncio async def test_curl_subprocess_get_parses_status_and_strips_marker(): - """_curl_subprocess_get must parse the -w status marker and strip it from body.""" s = YandexRealtyScraper() - s._cookies = {} - - fake_stdout = b"cards" + _CURL_STATUS_MARKER.encode() + b"200" - fake_stderr = b"" + fake_body = b"{" + b'"response":{}}' + fake_stdout = fake_body + _CURL_STATUS_MARKER.encode() + b"200" fake_proc = MagicMock() fake_proc.returncode = 0 - fake_proc.communicate = AsyncMock(return_value=(fake_stdout, fake_stderr)) + fake_proc.communicate = AsyncMock(return_value=(fake_stdout, b"")) - with patch("asyncio.create_subprocess_exec", return_value=fake_proc) as mock_exec: + with patch("asyncio.create_subprocess_exec", return_value=fake_proc): resp = await s._curl_subprocess_get( "https://realty.yandex.ru/test/", "socks5h://x:y@h:1", 30 ) - mock_exec.assert_called_once() assert isinstance(resp, _CurlResponse) assert resp.status_code == 200 - assert resp.text == "cards" + assert resp.text == "{" + '"response":{}}' @pytest.mark.asyncio async def test_http_get_http_proxy_uses_cffi_not_curl(): - """When proxy is http://, _http_get must NOT call _curl_subprocess_get; - it falls through to the legacy curl_cffi path.""" + """When proxy is http://, must NOT call _curl_subprocess_get.""" s = YandexRealtyScraper() sentinel_resp = MagicMock() sentinel_resp.status_code = 200 - sentinel_resp.text = "" + sentinel_resp.text = "{}" cffi_mock = AsyncMock() cffi_mock.get = AsyncMock(return_value=sentinel_resp) @@ -661,119 +477,225 @@ async def test_http_get_http_proxy_uses_cffi_not_curl(): assert result is sentinel_resp +# --------------------------------------------------------------------------- +# PART G: fetch_around — tarpit resilience +# --------------------------------------------------------------------------- + + +def _valid_gate_text(entity: dict | None = None) -> str: + ent = entity or _ENTITY_FULL + pager = {"page": 0, "pageSize": 20, "totalItems": 1, "totalPages": 1} + return json.dumps(_make_gate_payload([ent], pager)) + + @pytest.mark.asyncio -async def test_fetch_around_multi_room_dedup(): - """fetch_around_multi_room deduplicates offers seen in multiple combos.""" +async def test_fetch_around_returns_lots_on_200_json(): + s = YandexRealtyScraper() + body = _valid_gate_text() + + async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse: + return _CurlResponse(status_code=200, text=body) + + with patch.object(s, "_http_get", side_effect=fake_http_get): + with patch.object(s, "sleep_between_requests", new_callable=AsyncMock): + lots = await s.fetch_around(lat=56.8, lon=60.6, rooms="1") + + assert len(lots) == 1 + assert lots[0].source_id == str(_ENTITY_FULL["offerId"]) + + +@pytest.mark.asyncio +async def test_fetch_around_rotates_and_retries_on_status_0(): + """fetch_around rotates IP and retries on status_code==0 (tarpit).""" + s = YandexRealtyScraper() + call_count = 0 + body = _valid_gate_text() + + async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse: + nonlocal call_count + call_count += 1 + if call_count == 1: + return _CurlResponse(status_code=0, text="") + return _CurlResponse(status_code=200, text=body) + + with patch.object(s, "_http_get", side_effect=fake_http_get): + with patch.object(s, "_rotate_ip", new_callable=AsyncMock) as mock_rotate: + with patch("asyncio.sleep", new_callable=AsyncMock): + with patch.object(s, "sleep_between_requests", new_callable=AsyncMock): + lots = await s.fetch_around(lat=56.84, lon=60.60, rooms="1") + + assert mock_rotate.call_count == 1 + assert call_count == 2 + assert len(lots) >= 1 + + +@pytest.mark.asyncio +async def test_fetch_around_rotates_on_json_error(): + """fetch_around rotates when response body is not valid JSON.""" + s = YandexRealtyScraper() + call_count = 0 + body = _valid_gate_text() + + async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse: + nonlocal call_count + call_count += 1 + if call_count == 1: + return _CurlResponse(status_code=200, text="not json") + return _CurlResponse(status_code=200, text=body) + + with patch.object(s, "_http_get", side_effect=fake_http_get): + with patch.object(s, "_rotate_ip", new_callable=AsyncMock) as mock_rotate: + with patch("asyncio.sleep", new_callable=AsyncMock): + with patch.object(s, "sleep_between_requests", new_callable=AsyncMock): + lots = await s.fetch_around(lat=56.84, lon=60.60, rooms="2") + + assert mock_rotate.call_count == 1 + assert len(lots) >= 1 + + +@pytest.mark.asyncio +async def test_fetch_around_exhausts_retries_returns_empty(): + """fetch_around returns [] when all retries exhausted.""" s = YandexRealtyScraper() - # Two combos both return the same offer_id "1234" plus unique ids - combo_results: list[list[ScrapedLot]] = [] - for i in range(2): - lot_shared = MagicMock(spec=ScrapedLot) - lot_shared.source_id = "shared_offer_1234" - lot_shared.source_url = "https://realty.yandex.ru/offer/shared_offer_1234/" + async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse: + return _CurlResponse(status_code=0, text="") - lot_unique = MagicMock(spec=ScrapedLot) - lot_unique.source_id = f"unique_offer_{i}" - lot_unique.source_url = f"https://realty.yandex.ru/offer/unique_offer_{i}/" + with patch.object(s, "_http_get", side_effect=fake_http_get): + with patch.object(s, "_rotate_ip", new_callable=AsyncMock) as mock_rotate: + with patch("asyncio.sleep", new_callable=AsyncMock): + lots = await s.fetch_around(lat=56.84, lon=60.60, rooms="1") - combo_results.append([lot_shared, lot_unique]) - - call_count = 0 - - async def fake_fetch_around(*args, **kwargs): - nonlocal call_count - idx = call_count - call_count += 1 - if idx < len(combo_results): - return combo_results[idx] - return [] - - with patch.object(s, "fetch_around", side_effect=fake_fetch_around): - lots = await s.fetch_around_multi_room( - lat=0.0, - lon=0.0, - rooms_list=["1"], - price_ranges=[(None, 5_000_000), (5_000_000, 7_000_000)], - max_pages=1, - ) - - # shared_offer_1234 appears once; 2 unique offers - ids = {lot.source_id for lot in lots} - assert "shared_offer_1234" in ids - assert len(ids) == 3 # shared + 2 unique + assert mock_rotate.call_count >= 1 + assert lots == [] @pytest.mark.asyncio -async def test_fetch_around_multi_room_legacy_no_combos(): +async def test_fetch_around_no_rotate_on_404(): + """fetch_around does NOT rotate on 404 -- only on tarpit (status==0) / bad JSON.""" + s = YandexRealtyScraper() + + async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse: + return _CurlResponse(status_code=404, text="Not Found") + + with patch.object(s, "_http_get", side_effect=fake_http_get): + with patch.object(s, "_rotate_ip", new_callable=AsyncMock) as mock_rotate: + lots = await s.fetch_around(lat=56.84, lon=60.60) + + assert mock_rotate.call_count == 0 + assert lots == [] + + +# --------------------------------------------------------------------------- +# PART H: fetch_around_multi_room +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_fetch_around_multi_room_dedup(): + """Multi-room fetch deduplicates offers seen in multiple combos. + + fetch_around_multi_room fetches page=1 inline via _http_get, not via + fetch_around. We mock _http_get to return JSON payloads with shared/unique + offer IDs to verify deduplication. + """ + s = YandexRealtyScraper() + + # Combo 0: shared_offer + unique_0 + # Combo 1: shared_offer + unique_1 + entity_shared = dict( + _ENTITY_FULL, offerId="shared_offer_1234", url="//realty.yandex.ru/offer/shared_offer_1234" + ) + entity_unique_0 = dict( + _ENTITY_FULL, offerId="unique_offer_0", url="//realty.yandex.ru/offer/unique_offer_0" + ) + entity_unique_1 = dict( + _ENTITY_FULL, offerId="unique_offer_1", url="//realty.yandex.ru/offer/unique_offer_1" + ) + + pager = {"page": 0, "pageSize": 20, "totalItems": 2, "totalPages": 1} + payload_combo0 = _make_gate_payload([entity_shared, entity_unique_0], pager) + payload_combo1 = _make_gate_payload([entity_shared, entity_unique_1], pager) + + combo_payloads = [payload_combo0, payload_combo1] + call_idx = 0 + + async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse: + nonlocal call_idx + idx = call_idx + call_idx += 1 + if idx < len(combo_payloads): + return _CurlResponse(status_code=200, text=json.dumps(combo_payloads[idx])) + return _CurlResponse(status_code=200, text=json.dumps(_make_gate_payload([]))) + + mock_settings = MagicMock() + mock_settings.yandex_proxy_url = None + + with patch.dict("sys.modules", {"app.core.config": MagicMock(settings=mock_settings)}): + with patch.object(s, "_http_get", side_effect=fake_http_get): + with patch.object(s, "sleep_between_requests", new_callable=AsyncMock): + lots = await s.fetch_around_multi_room( + lat=0.0, + lon=0.0, + rooms_list=["1"], + price_ranges=[(None, 5_000_000), (5_000_000, 7_000_000)], + max_pages=1, + ) + + ids = {lot.source_id for lot in lots} + assert "shared_offer_1234" in ids + assert len(ids) == 3 + + +@pytest.mark.asyncio +async def test_fetch_around_multi_room_legacy_single_sweep(): """Legacy mode (no rooms_list/price_ranges) uses single citywide sweep.""" s = YandexRealtyScraper() calls: list[dict] = [] async def fake_fetch_around( - lat, lon, radius_m=1000, page=0, rooms=None, price_min=None, price_max=None + lat, lon, radius_m=1000, page=1, rooms=None, price_min=None, price_max=None ): calls.append({"page": page, "rooms": rooms, "price_min": price_min}) - return [] # empty → stops pagination + return [] with patch.object(s, "fetch_around", side_effect=fake_fetch_around): await s.fetch_around_multi_room(lat=0.0, lon=0.0, max_pages=3) - # Should call with rooms=None, price_min=None (citywide) assert all(c["rooms"] is None for c in calls) assert all(c["price_min"] is None for c in calls) -# ── PART E: exhaustive load — fetch_all_secondary + _walk_price_range ──────── - - -def _make_lots(n: int, prefix: str = "lot") -> list[MagicMock]: - """Create n mock ScrapedLot objects with unique source_id.""" - lots = [] - for i in range(n): - lot = MagicMock(spec=ScrapedLot) - lot.source_id = f"{prefix}_{i}" - lot.source_url = f"https://realty.yandex.ru/offer/{prefix}_{i}/" - lots.append(lot) - return lots +# --------------------------------------------------------------------------- +# PART I: exhaustive load -- fetch_all_secondary + _walk_price_range +# --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_walk_price_range_splits_when_total_exceeds_cap(): - """When totalItems > price_cap_per_bucket and bracket >= MIN_BRACKET, - _walk_price_range recurses and subdivides — each leaf has ≤ cap items.""" + """When totalItems > cap, _walk_price_range splits into sub-ranges recursively.""" s = YandexRealtyScraper() - s._cffi_session = AsyncMock() - call_ranges: list[tuple[int | None, int]] = [] + call_ranges: list[tuple] = [] + call_count = 0 - async def fake_fetch_page_html( - rooms: str | None, page: int, price_min: int | None, price_max: int - ) -> str: + async def fake_fetch_page_json(rooms, page, price_min, price_max): + nonlocal call_count + call_count += 1 call_ranges.append((price_min, price_max)) - # Return captcha-free HTML with 0 cards (parsing gives no lots, which is fine — - # we only care that the range is split, not about actual lot count in this test) - return "empty" - - # total_count returns 600 for the first (wide) probe, then 0 for sub-ranges - # so recursion terminates after the first split. - call_counts: list[int] = [0] - - def fake_extract_total(html: str) -> int | None: - call_counts[0] += 1 - # First call (wide range) returns 600 > cap=500, triggering a split - # Subsequent calls (sub-ranges) return 0 → no further recursion - if call_counts[0] == 1: - return 600 - return 0 + # First call (wide range): 600 items -> triggers split + total = 600 if call_count == 1 else 10 + entities = [dict(_ENTITY_FULL, offerId=f"offer_{call_count}")] + pager = { + "page": 0, + "pageSize": 20, + "totalItems": total, + "totalPages": max(1, total // 20), + } + return _make_gate_payload(entities, pager) seen: dict = {} - with ( - patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html), - patch.object(s, "_extract_total_count", side_effect=fake_extract_total), - patch.object(s, "_is_captcha_html", return_value=False, create=True), - ): - # Patch asyncio.sleep to be instant + with patch.object(s, "_fetch_page_json", side_effect=fake_fetch_page_json): with patch("asyncio.sleep", new_callable=AsyncMock): await s._walk_price_range( rooms="1", @@ -781,206 +703,111 @@ async def test_walk_price_range_splits_when_total_exceeds_cap(): hi=10_000_000, seen=seen, price_cap_per_bucket=500, - max_pages_per_bucket=100, + max_pages_per_bucket=50, concurrency=2, ) - # Should have probed at least the initial range + 2 sub-ranges - assert len(call_ranges) >= 3, f"Expected >= 3 probes (wide + 2 sub-ranges), got {call_ranges}" - # Initial probe covers the full range - assert call_ranges[0] == (None, 10_000_000) # lo=0 → _lo_param=None - - -@pytest.mark.asyncio -async def test_walk_price_range_fires_on_bucket_per_leaf(): - """on_bucket callback is called once per leaf bucket with (bucket_key, lots).""" - s = YandexRealtyScraper() - s._cffi_session = AsyncMock() - - bucket_calls: list[tuple[str, list]] = [] - - def fake_on_bucket(bucket_key: str, lots: list) -> None: - bucket_calls.append((bucket_key, list(lots))) - - # total = 10 (< cap=500) → paginate leaf immediately, 1 page (page=0) - async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str: - # Return HTML with 1 parseable card for page=0, empty for others - if page == 0: - return SINGLE_CARD_HTML - return EMPTY_PAGE_HTML - - seen: dict = {} - with ( - patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html), - patch.object(s, "_extract_total_count", return_value=10), - patch("asyncio.sleep", new_callable=AsyncMock), - ): - await s._walk_price_range( - rooms="1", - lo=0, - hi=20_000_000, - seen=seen, - price_cap_per_bucket=500, - max_pages_per_bucket=5, - concurrency=2, - on_bucket=fake_on_bucket, - ) - - # on_bucket must have been called at least once - assert len(bucket_calls) >= 1 - # All calls contain (bucket_key, lots) - for key, call_lots in bucket_calls: - assert isinstance(key, str), f"bucket_key must be str, got {type(key)}" - assert len(call_lots) >= 0 # may be 0 if card parse yields empty — still fires + assert len(call_ranges) >= 3, f"Expected >=3 probes, got {call_ranges}" + assert call_ranges[0] == (None, 10_000_000) # lo=0 -> _lo_param=None @pytest.mark.asyncio async def test_walk_price_range_fallback_when_total_none(): - """When totalItems=None (state unavailable), falls back to paginate-until-empty - and still fires on_bucket(bucket_key, lots).""" + """totalItems absent in pager triggers paginate-until-empty fallback.""" s = YandexRealtyScraper() - s._cffi_session = AsyncMock() + call_count = 0 - bucket_calls: list[tuple[str, list]] = [] - - def fake_on_bucket(bucket_key: str, lots: list) -> None: - bucket_calls.append((bucket_key, list(lots))) - - page_htmls = { - 0: SINGLE_CARD_HTML, # page=0: 1 card - 1: EMPTY_PAGE_HTML, # page=1: empty → stop - } - - async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str | None: - return page_htmls.get(page, EMPTY_PAGE_HTML) + async def fake_fetch_page_json(rooms, page, price_min, price_max): + nonlocal call_count + call_count += 1 + if call_count == 1: + # Probe: pager without totalItems + pager = {"page": 0, "pageSize": 20} + return _make_gate_payload([_ENTITY_FULL], pager) + if call_count == 2: + pager = {"page": 0, "pageSize": 20} + return _make_gate_payload([dict(_ENTITY_FULL, offerId=f"offer_{call_count}")], pager) + return _make_gate_payload([]) seen: dict = {} - with ( - patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html), - patch.object(s, "_extract_total_count", return_value=None), - patch("asyncio.sleep", new_callable=AsyncMock), - ): - await s._walk_price_range( - rooms="2", - lo=0, - hi=15_000_000, - seen=seen, - price_cap_per_bucket=500, - max_pages_per_bucket=10, - concurrency=2, - on_bucket=fake_on_bucket, - ) + bucket_calls: list = [] - # Fallback path must fire on_bucket with the collected lots - assert len(bucket_calls) >= 1, "on_bucket must be called in fallback path" - # The card from SINGLE_CARD_HTML should have been collected - all_lots = [lot for _key, call in bucket_calls for lot in call] - assert len(all_lots) >= 1 + with patch.object(s, "_fetch_page_json", side_effect=fake_fetch_page_json): + with patch("asyncio.sleep", new_callable=AsyncMock): + await s._walk_price_range( + rooms="2", + lo=0, + hi=15_000_000, + seen=seen, + price_cap_per_bucket=500, + max_pages_per_bucket=10, + concurrency=2, + on_bucket=lambda k, n: bucket_calls.append((k, n)), + ) - -# ── PART F: checkpoint / resume — skip_buckets ─────────────────────────────── + assert len(bucket_calls) >= 1, "on_bucket must fire in fallback path" @pytest.mark.asyncio -async def test_skip_buckets_skips_pagination_and_on_bucket(): - """When a leaf bucket_key is in skip_buckets, pagination (page>0) and on_bucket - are NOT called for it. The probe fetch (page=0) may still run.""" +async def test_walk_price_range_fires_on_bucket_for_leaf(): + """on_bucket callback fires once for a leaf bucket with small totalItems.""" s = YandexRealtyScraper() - s._cffi_session = AsyncMock() + bucket_calls: list = [] - on_bucket_calls: list[str] = [] - pagination_fetches: list[int] = [] - - def fake_on_bucket(bucket_key: str, lots: list) -> None: - on_bucket_calls.append(bucket_key) - - async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str: - if page > 0: - pagination_fetches.append(page) - return SINGLE_CARD_HTML - - # bucket_key for rooms="1", lo=0, hi=5_000_000 is "1:0:5000000" - target_bucket_key = "1:0:5000000" - skip_set = {target_bucket_key} + async def fake_fetch_page_json(rooms, page, price_min, price_max): + pager = {"page": 0, "pageSize": 20, "totalItems": 5, "totalPages": 1} + return _make_gate_payload([_ENTITY_FULL], pager) seen: dict = {} - with ( - patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html), - patch.object(s, "_extract_total_count", return_value=10), - patch("asyncio.sleep", new_callable=AsyncMock), - ): - await s._walk_price_range( - rooms="1", - lo=0, - hi=5_000_000, - seen=seen, - price_cap_per_bucket=500, - max_pages_per_bucket=5, - concurrency=2, - on_bucket=fake_on_bucket, - skip_buckets=skip_set, - ) + with patch.object(s, "_fetch_page_json", side_effect=fake_fetch_page_json): + with patch("asyncio.sleep", new_callable=AsyncMock): + await s._walk_price_range( + rooms="1", + lo=0, + hi=20_000_000, + seen=seen, + price_cap_per_bucket=500, + max_pages_per_bucket=5, + concurrency=2, + on_bucket=lambda k, n: bucket_calls.append(k), + ) - # on_bucket must NOT have been called for the skipped bucket - assert ( - target_bucket_key not in on_bucket_calls - ), f"on_bucket must not fire for skipped bucket, but got: {on_bucket_calls}" - # No pagination pages (page > 0) should have been fetched for the skipped bucket - assert ( - pagination_fetches == [] - ), f"pagination must not run for skipped bucket, but page fetches: {pagination_fetches}" + assert len(bucket_calls) >= 1 @pytest.mark.asyncio -async def test_on_bucket_receives_key_and_lots(): - """on_bucket receives (bucket_key: str, lots: list) where bucket_key matches - '{rooms}:{lo}:{hi}' shape for the normal-leaf path.""" +async def test_walk_price_range_skip_buckets(): + """skip_buckets prevents on_bucket from firing for already-done bucket.""" s = YandexRealtyScraper() - s._cffi_session = AsyncMock() + on_bucket_calls: list = [] - received_keys: list[str] = [] - received_lots: list[list] = [] - - def fake_on_bucket(bucket_key: str, lots: list) -> None: - received_keys.append(bucket_key) - received_lots.append(list(lots)) - - async def fake_fetch_page_html(rooms, page, price_min, price_max) -> str: - if page == 0: - return SINGLE_CARD_HTML - return EMPTY_PAGE_HTML + async def fake_fetch_page_json(rooms, page, price_min, price_max): + pager = {"page": 0, "pageSize": 20, "totalItems": 10, "totalPages": 1} + return _make_gate_payload([_ENTITY_FULL], pager) + # lo=0 -> _lo_param=None -> bucket key uses None + skip_set = {_combo_label("1", 0, 10_000_000)} seen: dict = {} - with ( - patch.object(s, "_fetch_page_html", side_effect=fake_fetch_page_html), - patch.object(s, "_extract_total_count", return_value=5), - patch("asyncio.sleep", new_callable=AsyncMock), - ): - await s._walk_price_range( - rooms="3", - lo=1_000_000, - hi=8_000_000, - seen=seen, - price_cap_per_bucket=500, - max_pages_per_bucket=5, - concurrency=2, - on_bucket=fake_on_bucket, - ) + with patch.object(s, "_fetch_page_json", side_effect=fake_fetch_page_json): + with patch("asyncio.sleep", new_callable=AsyncMock): + await s._walk_price_range( + rooms="1", + lo=0, + hi=10_000_000, + seen=seen, + price_cap_per_bucket=500, + max_pages_per_bucket=5, + concurrency=2, + on_bucket=lambda k, n: on_bucket_calls.append(k), + skip_buckets=skip_set, + ) - assert len(received_keys) >= 1, "on_bucket must have been called" - # bucket_key must match the expected format - for key in received_keys: - assert ( - key == "3:1000000:8000000" - ), f"bucket_key format mismatch: expected '3:1000000:8000000', got {key!r}" - # lots must be a list - for lots in received_lots: - assert isinstance(lots, list) + assert skip_set.pop() not in on_bucket_calls @pytest.mark.asyncio async def test_fetch_all_secondary_deduplicates_across_rooms(): - """fetch_all_secondary deduplicates offers that appear in multiple room buckets.""" + """fetch_all_secondary deduplicates offers across room buckets.""" s = YandexRealtyScraper() shared_lot = MagicMock(spec=ScrapedLot) @@ -997,9 +824,8 @@ async def test_fetch_all_secondary_deduplicates_across_rooms(): call_count = 0 - async def fake_walk(*, rooms, lo, hi, seen, **kwargs): # type: ignore[override] + async def fake_walk(*, rooms, lo, hi, seen, **kwargs): nonlocal call_count - # Both room buckets yield the shared lot + one unique each seen["shared_999"] = shared_lot if call_count == 0: seen["unique_001"] = unique_lot_1 @@ -1014,112 +840,4 @@ async def test_fetch_all_secondary_deduplicates_across_rooms(): assert "shared_999" in ids assert "unique_001" in ids assert "unique_002" in ids - assert len(ids) == 3 # shared deduplicated — only 3 unique - - -# ── PART H: tarpit resilience — fetch_around rotate-on-status-0 / captcha ──── - - -@pytest.mark.asyncio -async def test_fetch_around_rotates_and_retries_on_status_0(): - """fetch_around rotates IP and retries when status_code==0 (tarpit/curl-timeout). - - Sequence: first call → status 0 (tarpit), second call → status 200 + valid HTML. - Expect: _rotate_ip called once, parsed lots returned from the second attempt. - """ - s = YandexRealtyScraper() - - # Valid SERP page that parses to >=1 lot - valid_html = SINGLE_CARD_HTML.replace(" ", " ") - - call_count = 0 - - async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse: - nonlocal call_count - call_count += 1 - if call_count == 1: - # First attempt: tarpit — status 0 - return _CurlResponse(status_code=0, text="") - # Second attempt: success - return _CurlResponse(status_code=200, text=valid_html) - - mock_settings = MagicMock() - mock_settings.yandex_proxy_url = "socks5h://x:y@h:1" - mock_settings.yandex_proxy_rotate_url = "http://rotate.example.com/changeip" - mock_settings.avito_proxy_rotate_url = None - - with patch.dict("sys.modules", {"app.core.config": MagicMock(settings=mock_settings)}): - with patch.object(s, "_http_get", side_effect=fake_http_get): - with patch.object(s, "_rotate_ip", new_callable=AsyncMock) as mock_rotate: - with patch("asyncio.sleep", new_callable=AsyncMock): - with patch.object(s, "sleep_between_requests", new_callable=AsyncMock): - lots = await s.fetch_around(lat=56.84, lon=60.60, rooms="1") - - assert ( - mock_rotate.call_count == 1 - ), f"_rotate_ip must be called once, got {mock_rotate.call_count}" - assert ( - call_count == 2 - ), f"_http_get must be called twice (tarpit then success), got {call_count}" - assert len(lots) >= 1, f"Expected >=1 parsed lots after retry, got {len(lots)}" - - -@pytest.mark.asyncio -async def test_fetch_around_rotates_on_captcha(): - """fetch_around rotates IP and retries when HTML is a captcha page.""" - s = YandexRealtyScraper() - - captcha_html = "captcha" - valid_html = SINGLE_CARD_HTML.replace(" ", " ") - - call_count = 0 - - async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse: - nonlocal call_count - call_count += 1 - if call_count == 1: - return _CurlResponse(status_code=200, text=captcha_html) - return _CurlResponse(status_code=200, text=valid_html) - - with patch.object(s, "_http_get", side_effect=fake_http_get): - with patch.object(s, "_rotate_ip", new_callable=AsyncMock) as mock_rotate: - with patch("asyncio.sleep", new_callable=AsyncMock): - with patch.object(s, "sleep_between_requests", new_callable=AsyncMock): - lots = await s.fetch_around(lat=56.84, lon=60.60, rooms="2") - - assert mock_rotate.call_count == 1 - assert len(lots) >= 1 - - -@pytest.mark.asyncio -async def test_fetch_around_exhausts_retries_returns_empty(): - """fetch_around returns [] when all retries are exhausted (tarpit persists).""" - s = YandexRealtyScraper() - - async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse: - return _CurlResponse(status_code=0, text="") - - with patch.object(s, "_http_get", side_effect=fake_http_get): - with patch.object(s, "_rotate_ip", new_callable=AsyncMock) as mock_rotate: - with patch("asyncio.sleep", new_callable=AsyncMock): - lots = await s.fetch_around(lat=56.84, lon=60.60, rooms="1") - - # Rotated on every attempt: 1 initial + _YANDEX_TARPIT_MAX_RETRIES retries - assert mock_rotate.call_count == 1 + _YANDEX_TARPIT_MAX_RETRIES - assert lots == [] - - -@pytest.mark.asyncio -async def test_fetch_around_no_rotate_on_404(): - """fetch_around does NOT rotate on non-zero non-200 status (e.g. 404).""" - s = YandexRealtyScraper() - - async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse: - return _CurlResponse(status_code=404, text="Not Found") - - with patch.object(s, "_http_get", side_effect=fake_http_get): - with patch.object(s, "_rotate_ip", new_callable=AsyncMock) as mock_rotate: - lots = await s.fetch_around(lat=56.84, lon=60.60) - - assert mock_rotate.call_count == 0, "Must not rotate on 404 — only on status==0 / captcha" - assert lots == [] + assert len(ids) == 3