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)

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:

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())
 

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."""