From bc4c6852a1daea5135d2a704c999d4d069e9d839 Mon Sep 17 00:00:00 2001 From: bot-backend Date: Wed, 17 Jun 2026 22:22:17 +0300 Subject: [PATCH] feat(yandex): ingest novostroyki via gate-API newFlat=YES --- tradein-mvp/backend/app/services/scheduler.py | 1 + .../backend/app/services/scrape_pipeline.py | 20 +- .../app/services/scrapers/yandex_realty.py | 241 ++++++++++-------- .../backend/tests/test_yandex_realty_serp.py | 113 ++++++++ 4 files changed, 269 insertions(+), 106 deletions(-) diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 696cf7a7..0555a7b6 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -321,6 +321,7 @@ async def trigger_yandex_city_sweep_run(db: Session, schedule_row: dict[str, Any request_delay_sec=float(params.get("request_delay_sec", 9.0)), radius_m=int(params.get("radius_m", 1500)), enrich_address=bool(params.get("enrich_address", True)), + segments=params.get("segments"), ) except Exception: logger.exception("scheduler: run_yandex_city_sweep crashed run_id=%d", run_id) diff --git a/tradein-mvp/backend/app/services/scrape_pipeline.py b/tradein-mvp/backend/app/services/scrape_pipeline.py index 2d02d49c..793a3101 100644 --- a/tradein-mvp/backend/app/services/scrape_pipeline.py +++ b/tradein-mvp/backend/app/services/scrape_pipeline.py @@ -1033,6 +1033,7 @@ async def run_yandex_city_sweep( enrich_address: bool = True, rooms_list: list[str] | None = None, price_ranges: list[tuple[int | None, int | None]] | None = None, + segments: list[str] | None = None, ) -> YandexCitySweepCounters: """Yandex.Недвижимость city sweep: rooms × price combos от центра ЕКБ → save → address-enrich. @@ -1078,6 +1079,11 @@ async def run_yandex_city_sweep( _rooms_list = rooms_list or list(ROOM_PATH.keys()) _price_ranges = price_ranges or DEFAULT_PRICE_RANGES + # newFlat-сегменты: оба прохода (vtorichka + novostroyki) по одним combos. + # Дефолт ["NO", "YES"] — оба; dedup по source_id span'ит оба прохода (один seen + # внутри fetch_around_multi_room). Counters (lots_fetched/inserted/updated) + # агрегируют оба прохода через len(anchor_lots). + _segments = segments or ["NO", "YES"] counters = YandexCitySweepCounters(anchors_total=len(_anchors)) inter_anchor_delay = request_delay_sec if request_delay_sec is not None else 7.0 @@ -1086,16 +1092,21 @@ async def run_yandex_city_sweep( consecutive_failures = 0 # Вычисляем watchdog-таймаут для combos-режима (центр, anchors=None). - # В combos-режиме один "anchor" выполняет num_combos × max_pages fetch'ей — - # каждый занимает примерно _resolved_delay + _YANDEX_COMBOS_PER_FETCH_S секунд. - # Для 30 combos × 3 pages × (9+12) + 300 ≈ 2190s (~37 мин). + # В combos-режиме один "anchor" выполняет num_segments × num_combos × max_pages + # fetch'ей — каждый занимает примерно _resolved_delay + _YANDEX_COMBOS_PER_FETCH_S сек. + # Для 2 segments × 30 combos × 3 pages × (9+12) + 300 ≈ 4080s (~68 мин). + # Множитель len(_segments) масштабирует watchdog пропорционально числу проходов + # (vtorichka + novostroyki ≈ ×2 wall-clock) — иначе второй проход гильотинится. # В explicit-anchor (тест/override) режиме оставляем ANCHOR_TIMEOUT_SEC (240s) — # там каждый anchor небольшой и watchdog работает как старый защитный барьер. _num_combos = len(_rooms_list) * len(_price_ranges) if anchors is None and _num_combos > 0: _sweep_timeout = max( ANCHOR_TIMEOUT_SEC, - _num_combos * pages_per_anchor * (_resolved_delay + _YANDEX_COMBOS_PER_FETCH_S) + len(_segments) + * _num_combos + * pages_per_anchor + * (_resolved_delay + _YANDEX_COMBOS_PER_FETCH_S) + _YANDEX_ADDRESS_ENRICH_BUDGET_S, ) else: @@ -1162,6 +1173,7 @@ async def run_yandex_city_sweep( max_pages=pages_per_anchor, rooms_list=_rooms_list, price_ranges=_price_ranges, + segments=_segments, ) counters.lots_fetched += len(anchor_lots) if anchor_lots: diff --git a/tradein-mvp/backend/app/services/scrapers/yandex_realty.py b/tradein-mvp/backend/app/services/scrapers/yandex_realty.py index d078465c..a9310ce8 100644 --- a/tradein-mvp/backend/app/services/scrapers/yandex_realty.py +++ b/tradein-mvp/backend/app/services/scrapers/yandex_realty.py @@ -129,7 +129,14 @@ def _extract_json_from_content(content: str) -> str | None: return None -def _parse_gate_json(payload: dict[str, Any], page_param: int = 1) -> list[ScrapedLot]: +def _segment_for(new_flat: str) -> str: + """Map a newFlat gate-API value to the listings.listing_segment value.""" + return "novostroyki" if new_flat == "YES" else "vtorichka" + + +def _parse_gate_json( + payload: dict[str, Any], page_param: int = 1, new_flat: str = "NO" +) -> list[ScrapedLot]: """Parse gate-API JSON payload into a list of ScrapedLot. Verified field mapping (from .issdump/yandex_gate_api_sample.json + live prod 2026-06-17): @@ -150,7 +157,7 @@ def _parse_gate_json(payload: dict[str, Any], page_param: int = 1) -> list[Scrap location.geocoderAddress -> address (full, with house number) mainImages[] -> photo_urls (prepend "https:", up to 5) building.siteId -> house_ext_id (JK linkage) - listing_segment = "vtorichka" (all requests use newFlat=NO) + listing_segment -> "novostroyki" (newFlat=YES) | "vtorichka" (newFlat=NO) """ result = _extract_gate_data(payload) if result is None: @@ -160,13 +167,15 @@ def _parse_gate_json(payload: dict[str, Any], page_param: int = 1) -> list[Scrap entities, _pager = result lots: list[ScrapedLot] = [] for entity in entities: - lot = _entity_to_lot(entity, page_param=page_param) + lot = _entity_to_lot(entity, page_param=page_param, new_flat=new_flat) if lot is not None: lots.append(lot) return lots -def _entity_to_lot(entity: dict[str, Any], page_param: int = 1) -> ScrapedLot | None: +def _entity_to_lot( + entity: dict[str, Any], page_param: int = 1, new_flat: str = "NO" +) -> ScrapedLot | None: """Convert one gate-API entity dict to ScrapedLot. None on missing required fields.""" try: offer_id = str(entity.get("offerId") or "") @@ -263,7 +272,7 @@ def _entity_to_lot(entity: dict[str, Any], page_param: int = 1) -> ScrapedLot | house_source=house_source, house_ext_id=house_ext_id, house_url=None, - listing_segment="vtorichka", + listing_segment=_segment_for(new_flat), raw_payload={ "page_param": page_param, "ceiling_height": ceiling_height, @@ -366,13 +375,18 @@ class YandexRealtyScraper(BaseScraper): rooms: str | None = None, price_min: int | None = None, price_max: int | None = None, + new_flat: str = "NO", ) -> str: - """Build gate-API URL. page: 1-based (page=0 -> API error).""" + """Build gate-API URL. page: 1-based (page=0 -> API error). + + new_flat: "NO" -> vtorichka (default, preserves legacy behavior), + "YES" -> novostroyki (primary-sale / reassignment segment). + """ params: dict[str, str | int] = { "rgid": _EKB_RGID, "type": "SELL", "category": "APARTMENT", - "newFlat": "NO", + "newFlat": new_flat, "_pageType": "search", "_providers": "react-search-results-data", "page": page, @@ -391,9 +405,12 @@ class YandexRealtyScraper(BaseScraper): page: int, price_min: int | None, price_max: int | None, + new_flat: str = "NO", ) -> dict[str, Any] | None: """Fetch one gate-API page. Returns parsed JSON dict or None on error/invalid.""" - url = self._build_url(page=page, rooms=rooms, price_min=price_min, price_max=price_max) + url = self._build_url( + page=page, rooms=rooms, price_min=price_min, price_max=price_max, new_flat=new_flat + ) try: resp = await self._http_get(url, timeout=60) except Exception: @@ -421,14 +438,18 @@ class YandexRealtyScraper(BaseScraper): rooms: str | None = None, price_min: int | None = None, price_max: int | None = None, + new_flat: str = "NO", ) -> list[ScrapedLot]: """Fetch ONE page of gate-API results. lat/lon/radius_m ignored (uses rgid). page: 1-based. Tarpit resilience: status_code==0 or JSON error -> rotate IP + retry up to _YANDEX_TARPIT_MAX_RETRIES times. Non-tarpit failures (4xx/5xx) are not retried. + new_flat: "NO" -> vtorichka, "YES" -> novostroyki. """ - url = self._build_url(page=page, rooms=rooms, price_min=price_min, price_max=price_max) + url = self._build_url( + page=page, rooms=rooms, price_min=price_min, price_max=price_max, new_flat=new_flat + ) payload: dict[str, Any] | None = None for attempt in range(1 + _YANDEX_TARPIT_MAX_RETRIES): @@ -494,13 +515,14 @@ class YandexRealtyScraper(BaseScraper): ) return [] - lots = _parse_gate_json(payload, page_param=page) + lots = _parse_gate_json(payload, page_param=page, new_flat=new_flat) logger.info( - "yandex gate page=%d rooms=%s price=[%s-%s]: %d cards", + "yandex gate page=%d rooms=%s price=[%s-%s] newFlat=%s: %d cards", page, rooms or "all", price_min, price_max, + new_flat, len(lots), ) await self.sleep_between_requests() @@ -514,16 +536,20 @@ class YandexRealtyScraper(BaseScraper): max_pages: int = _GATE_MAX_PAGES_CAP, rooms_list: list[str] | None = None, price_ranges: list[tuple[int | None, int | None]] | None = None, + segments: list[str] | None = None, **_legacy_kwargs: Any, ) -> list[ScrapedLot]: - """Fetch via rooms x price-range combos; paginate each combo to totalPages. + """Fetch via segment x rooms x price-range combos; paginate each combo to totalPages. Pagination driven by pager.totalPages from first page response. Each combo iterates page=1 .. min(totalPages, max_pages, _GATE_MAX_PAGES_CAP). - Deduplicates by source_id/source_url across all combos. + Deduplicates by source_id/source_url across ALL combos AND segments (one `seen`). Legacy mode (rooms_list=None, price_ranges=None): single citywide sweep. + segments: list of newFlat values, default ["NO"] (vtorichka only). Pass + ["NO", "YES"] to sweep both vtorichka and novostroyki in one call. """ seen: dict[str, ScrapedLot] = {} + _segments = segments or ["NO"] if rooms_list is None and price_ranges is None: combos: list[tuple[str | None, int | None, int | None]] = [(None, None, None)] @@ -532,112 +558,123 @@ class YandexRealtyScraper(BaseScraper): p_ranges = price_ranges or DEFAULT_PRICE_RANGES combos = [(r, lo, hi) for r in r_list for lo, hi in p_ranges] - for rooms, price_min, price_max in combos: - combo_label = _combo_label(rooms, price_min, price_max) - total_pages: int | None = None + for new_flat in _segments: + _seg = _segment_for(new_flat) + for rooms, price_min, price_max in combos: + combo_label = f"{_seg}/{_combo_label(rooms, price_min, price_max)}" + total_pages: int | None = None - for page in range(1, max_pages + 1): - if page == 1: - payload_p1: dict[str, Any] | None = None - p1_url = self._build_url( - page=1, rooms=rooms, price_min=price_min, price_max=price_max - ) - for _attempt in range(1 + _YANDEX_TARPIT_MAX_RETRIES): - try: - resp = await self._http_get(p1_url, timeout=60) - except Exception: - logger.exception("yandex gate combo fetch failed combo=%s", combo_label) + for page in range(1, max_pages + 1): + if page == 1: + payload_p1: dict[str, Any] | None = None + p1_url = self._build_url( + page=1, + rooms=rooms, + price_min=price_min, + price_max=price_max, + new_flat=new_flat, + ) + for _attempt in range(1 + _YANDEX_TARPIT_MAX_RETRIES): + try: + resp = await self._http_get(p1_url, timeout=60) + except Exception: + logger.exception( + "yandex gate combo fetch failed combo=%s", combo_label + ) + break + if resp.status_code == 0: # type: ignore[union-attr] + logger.warning( + "yandex gate: tarpit combo=%s page=1 -- rotating", combo_label + ) + await self._rotate_ip() + await asyncio.sleep(2) + continue + if resp.status_code != 200: # type: ignore[union-attr] + logger.warning( + "yandex gate: HTTP %d combo=%s page=1", + resp.status_code, + combo_label, # type: ignore[union-attr] + ) + break + try: + payload_p1 = json.loads(resp.text) # type: ignore[union-attr] + except (json.JSONDecodeError, ValueError): + logger.warning( + "yandex gate: JSON parse error combo=%s page=1", combo_label + ) + payload_p1 = None + await self._rotate_ip() + await asyncio.sleep(2) + continue + if _is_gate_error(payload_p1): + logger.warning( + "yandex gate: error payload combo=%s page=1", combo_label + ) + payload_p1 = None + break break - if resp.status_code == 0: # type: ignore[union-attr] - logger.warning( - "yandex gate: tarpit combo=%s page=1 -- rotating", combo_label - ) - await self._rotate_ip() - await asyncio.sleep(2) - continue - if resp.status_code != 200: # type: ignore[union-attr] - logger.warning( - "yandex gate: HTTP %d combo=%s page=1", - resp.status_code, - combo_label, # type: ignore[union-attr] + + if payload_p1 is None: + logger.debug( + "yandex gate combo [%s] page=1: failed -- skipping", combo_label ) break - try: - payload_p1 = json.loads(resp.text) # type: ignore[union-attr] - except (json.JSONDecodeError, ValueError): - logger.warning( - "yandex gate: JSON parse error combo=%s page=1", combo_label - ) - payload_p1 = None - await self._rotate_ip() - await asyncio.sleep(2) - continue - if _is_gate_error(payload_p1): - logger.warning( - "yandex gate: error payload combo=%s page=1", combo_label - ) - payload_p1 = None + + result = _extract_gate_data(payload_p1) + if result is None: break + _entities_p1, pager_p1 = result + total_pages = min( + pager_p1.get("totalPages", 1), + max_pages, + _GATE_MAX_PAGES_CAP, + ) + lots_p1 = _parse_gate_json(payload_p1, page_param=1, new_flat=new_flat) + if not lots_p1: + logger.debug( + "yandex gate combo [%s] page=1: empty -- stopping", combo_label + ) + break + for lot in lots_p1: + key = lot.source_id or lot.source_url + if key and key not in seen: + seen[key] = lot + await self.sleep_between_requests() + if total_pages <= 1: + break + continue + + if total_pages is not None and page > total_pages: break - if payload_p1 is None: + lots = await self.fetch_around( + lat, + lon, + radius_m, + page=page, + rooms=rooms, + price_min=price_min, + price_max=price_max, + new_flat=new_flat, + ) + if not lots: logger.debug( - "yandex gate combo [%s] page=1: failed -- skipping", combo_label + "yandex gate combo [%s] page=%d: empty -- stopping", + combo_label, + page, ) break - - result = _extract_gate_data(payload_p1) - if result is None: - break - _entities_p1, pager_p1 = result - total_pages = min( - pager_p1.get("totalPages", 1), - max_pages, - _GATE_MAX_PAGES_CAP, - ) - lots_p1 = _parse_gate_json(payload_p1, page_param=1) - if not lots_p1: - logger.debug( - "yandex gate combo [%s] page=1: empty -- stopping", combo_label - ) - break - for lot in lots_p1: + for lot in lots: key = lot.source_id or lot.source_url if key and key not in seen: seen[key] = lot - await self.sleep_between_requests() - if total_pages <= 1: - break - continue - - if total_pages is not None and page > total_pages: - break - - lots = await self.fetch_around( - lat, - lon, - radius_m, - page=page, - rooms=rooms, - price_min=price_min, - price_max=price_max, - ) - if not lots: - logger.debug( - "yandex gate combo [%s] page=%d: empty -- stopping", - combo_label, - page, - ) - break - for lot in lots: - key = lot.source_id or lot.source_url - if key and key not in seen: - seen[key] = lot logger.info( - "yandex gate aggregate: %d unique lots (%d combos)", + "yandex gate aggregate: %d unique lots (%d combos x %d segments=%s)", len(seen), len(combos), + len(_segments), + _segments, ) return list(seen.values()) diff --git a/tradein-mvp/backend/tests/test_yandex_realty_serp.py b/tradein-mvp/backend/tests/test_yandex_realty_serp.py index 2a3119ea..2561b0c5 100644 --- a/tradein-mvp/backend/tests/test_yandex_realty_serp.py +++ b/tradein-mvp/backend/tests/test_yandex_realty_serp.py @@ -182,6 +182,29 @@ def test_entity_to_lot_full_entity(): assert lot.raw_payload["kitchen_area_m2"] == pytest.approx(12.7) +def test_entity_to_lot_segment_vtorichka_default(): + """Default new_flat='NO' -> listing_segment='vtorichka'.""" + lot = _entity_to_lot(_ENTITY_FULL) + assert lot is not None + assert lot.listing_segment == "vtorichka" + + +def test_entity_to_lot_segment_novostroyki_on_new_flat_yes(): + """new_flat='YES' -> listing_segment='novostroyki'.""" + lot = _entity_to_lot(_ENTITY_FULL, new_flat="YES") + assert lot is not None + assert lot.listing_segment == "novostroyki" + + +def test_parse_gate_json_propagates_segment(): + """_parse_gate_json threads new_flat down to each lot's listing_segment.""" + payload = _make_gate_payload([_ENTITY_FULL, _ENTITY_STUDIO]) + lots_nb = _parse_gate_json(payload, new_flat="YES") + assert lots_nb and all(lot.listing_segment == "novostroyki" for lot in lots_nb) + lots_vt = _parse_gate_json(payload, new_flat="NO") + assert lots_vt and all(lot.listing_segment == "vtorichka" for lot in lots_vt) + + def test_entity_to_lot_studio_rooms_zero(): lot = _entity_to_lot(_ENTITY_STUDIO) assert lot is not None @@ -350,6 +373,21 @@ def test_build_url_combos_are_unique(): assert len(urls) == expected, f"URL collisions: {expected} combos but {len(urls)} unique" +def test_build_url_new_flat_default_no(): + """Default new_flat must keep newFlat=NO (vtorichka) — backward compat.""" + s = YandexRealtyScraper() + url = s._build_url(page=1) + assert "newFlat=NO" in url + assert "newFlat=YES" not in url + + +def test_build_url_new_flat_yes(): + """new_flat='YES' must produce newFlat=YES (novostroyki segment).""" + s = YandexRealtyScraper() + url = s._build_url(page=1, new_flat="YES") + assert "newFlat=YES" in url + + def test_default_city(): assert DEFAULT_CITY == "ekaterinburg" @@ -665,6 +703,81 @@ async def test_fetch_around_multi_room_dedup(): assert len(ids) == 3 +@pytest.mark.asyncio +async def test_fetch_around_multi_room_runs_both_segments_with_shared_dedup(): + """segments=['NO','YES'] sweeps both vtorichka+novostroyki; dedup spans both passes. + + Mocks _http_get and inspects requested URLs: both newFlat=NO and newFlat=YES + must be requested. The same offerId returned in both passes must be deduped to + one lot (single shared `seen` across passes), tagged by the FIRST pass's segment. + """ + s = YandexRealtyScraper() + + requested_new_flat: list[str] = [] + + entity_shared = dict( + _ENTITY_FULL, offerId="shared_xseg", url="//realty.yandex.ru/offer/shared_xseg" + ) + entity_vt = dict(_ENTITY_FULL, offerId="vt_only", url="//realty.yandex.ru/offer/vt_only") + entity_nb = dict(_ENTITY_FULL, offerId="nb_only", url="//realty.yandex.ru/offer/nb_only") + pager = {"page": 0, "pageSize": 20, "totalItems": 2, "totalPages": 1} + + async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse: + if "newFlat=YES" in url: + requested_new_flat.append("YES") + payload = _make_gate_payload([entity_shared, entity_nb], pager) + else: + requested_new_flat.append("NO") + payload = _make_gate_payload([entity_shared, entity_vt], pager) + return _CurlResponse(status_code=200, text=json.dumps(payload)) + + 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)], + segments=["NO", "YES"], + max_pages=1, + ) + + assert "NO" in requested_new_flat + assert "YES" in requested_new_flat + + by_id = {lot.source_id: lot for lot in lots} + assert set(by_id) == {"shared_xseg", "vt_only", "nb_only"} + # Shared offer deduped to ONE lot, tagged by the first pass (NO -> vtorichka). + assert by_id["shared_xseg"].listing_segment == "vtorichka" + assert by_id["vt_only"].listing_segment == "vtorichka" + assert by_id["nb_only"].listing_segment == "novostroyki" + + +@pytest.mark.asyncio +async def test_fetch_around_multi_room_default_segment_no_only(): + """Without segments arg, only newFlat=NO (vtorichka) is requested (legacy default).""" + s = YandexRealtyScraper() + requested_new_flat: list[str] = [] + pager = {"page": 0, "pageSize": 20, "totalItems": 1, "totalPages": 1} + + async def fake_http_get(url: str, **kwargs: object) -> _CurlResponse: + requested_new_flat.append("YES" if "newFlat=YES" in url else "NO") + body = json.dumps(_make_gate_payload([_ENTITY_FULL], pager)) + 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): + await s.fetch_around_multi_room( + lat=0.0, + lon=0.0, + rooms_list=["1"], + price_ranges=[(None, 5_000_000)], + max_pages=1, + ) + + assert set(requested_new_flat) == {"NO"} + + @pytest.mark.asyncio async def test_fetch_around_multi_room_legacy_single_sweep(): """Legacy mode (no rooms_list/price_ranges) uses single citywide sweep.""" -- 2.45.3