tradein-mvp/scripts/sweep-yandex-valuation-ekb.py

@@ -69,22 +69,34 @@ def _addresses_to_sweep(
     limit: int | None,
     skip_recent_days: int,
     single_address: str | None,
+    require_house_number: bool = True,
 ) -> list[str]:
-    """Pull unique active EKB yandex addresses, optionally skipping recently-cached ones."""
+    """Pull unique active EKB yandex addresses, optionally skipping recently-cached ones.
+
+    require_house_number: filter to addresses containing a house number. Yandex's
+    valuation endpoint needs a specific building — bare street names ('Академическая
+    улица') return 0 items. Heuristic: address must contain a digit after a comma
+    OR end with a digit token. Default True.
+    """
     if single_address:
         return [single_address.strip()]
 
+    # Postgres regex for house-number tokens — used when require_house_number=True.
+    # Matches ", 12" / ", 12к3" / ", 5Б" anywhere in the address.
+    house_num_clause = " AND address ~ ',\\s*\\d+' " if require_house_number else ""
+
     db = SessionLocal()
     try:
         if skip_recent_days > 0:
             sql = text(
-                """
+                f"""
                 SELECT DISTINCT l.address
                 FROM listings l
                 WHERE l.source = 'yandex'
                   AND l.region_code = 66
                   AND l.is_active = TRUE
                   AND l.address IS NOT NULL
+                  {house_num_clause}
                   AND NOT EXISTS (
                       SELECT 1 FROM external_valuations ev
                       WHERE ev.source = 'yandex_valuation'
@@ -97,13 +109,14 @@ def _addresses_to_sweep(
             rows = db.execute(sql, {"days": skip_recent_days}).mappings().all()
         else:
             sql = text(
-                """
+                f"""
                 SELECT DISTINCT address
                 FROM listings
                 WHERE source = 'yandex'
                   AND region_code = 66
                   AND is_active = TRUE
                   AND address IS NOT NULL
+                  {house_num_clause}
                 ORDER BY address
                 """
             )
@@ -124,17 +137,23 @@ async def _sweep_one_address(
 ) -> tuple[int, int]:
     """Fetch+save up to max_pages for one address.
 
+    Pagination stop conditions (whichever fires first):
+      1. Page returns 0 items (empty / past last page)
+      2. Cumulative fetched >= total_objects from page-1 header (when known)
+      3. max_pages reached
+
     Returns (items_fetched_total, items_saved_total).
     """
     total_fetched = 0
     total_saved = 0
+    total_objects_expected: int | None = None
+    seen_keys: set[tuple] = set()  # dedup within sweep (page overlap protection)
+
     for page in range(1, max_pages + 1):
         db = SessionLocal()
         try:
-            # _get_or_fetch_yandex_valuation_cached signature accepts address+offer_*;
-            # page is not in its signature — for paginated walk we must call the
-            # scraper directly. Use the cached helper for page=1 only, then call
-            # YandexValuationScraper.fetch_house_history for pages >= 2.
+            # _get_or_fetch_yandex_valuation_cached only fetches page 1 (no `page`
+            # arg); cache key includes address only. Pages >= 2 go direct to scraper.
             if page == 1:
                 result = await _get_or_fetch_yandex_valuation_cached(
                     db,
@@ -153,15 +172,32 @@ async def _sweep_one_address(
             if result is None or not result.history_items:
                 break
 
+            # Capture total_objects on first response (subsequent pages may not carry it)
+            if total_objects_expected is None and result.house.total_objects:
+                total_objects_expected = result.house.total_objects
+
+            # Dedup against prior pages (Yandex sometimes repeats items at page boundaries)
+            new_items = []
+            for it in result.history_items:
+                key = (it.publish_date, it.area_m2, it.floor, it.start_price)
+                if key not in seen_keys:
+                    seen_keys.add(key)
+                    new_items.append(it)
+
             saved = _save_yandex_history_items(db, result)
-            total_fetched += len(result.history_items)
+            total_fetched += len(new_items)
             total_saved += saved
+            exp_str = f"/{total_objects_expected}" if total_objects_expected is not None else ""
             print(
-                f"    page={page}/{max_pages}: items={len(result.history_items)} saved={saved}",
+                f"    page={page}/{max_pages}: items={len(result.history_items)} "
+                f"new={len(new_items)} saved={saved} cum={total_fetched}{exp_str}",
                 flush=True,
             )
-            # Stop early if Yandex doesn't paginate further (less than full page returned)
-            if len(result.history_items) < 13:
+            # Stop if we've collected at least what the header promised
+            if total_objects_expected is not None and total_fetched >= total_objects_expected:
+                break
+            # Or if this page yielded zero NEW items (pure duplicates → end of feed)
+            if not new_items:
                 break
         except Exception as e:
             print(
@@ -213,9 +249,29 @@ async def main() -> None:
         default=None,
         help="Single address mode — sweep only this address",
     )
+    p.add_argument(
+        "--require-house-number",
+        dest="require_house_number",
+        action="store_true",
+        default=True,
+        help="Skip addresses without a house number (default: True). "
+        "Yandex valuation endpoint needs a specific building; street-only addresses "
+        "return 0 items. Use --no-require-house-number to disable.",
+    )
+    p.add_argument(
+        "--no-require-house-number",
+        dest="require_house_number",
+        action="store_false",
+        help="Include all addresses (sweep street-only entries too — most return 0).",
+    )
     args = p.parse_args()
 
-    addresses = _addresses_to_sweep(args.limit, args.skip_recent_days, args.address)
+    addresses = _addresses_to_sweep(
+        args.limit,
+        args.skip_recent_days,
+        args.address,
+        require_house_number=args.require_house_number,
+    )
     if not addresses:
         print("No addresses to sweep (all recently cached or none match query).", flush=True)
         return