feat(tradein): bulk sweep CLI for Yandex valuation house-history (EKB)

- New tradein-mvp/scripts/sweep-yandex-valuation-ekb.py
- Iterates DISTINCT addresses from listings (source='yandex' AND region_code=66)
- Calls existing estimator helpers (_get_or_fetch_yandex_valuation_cached +
  _save_yandex_history_items) — page=1 uses 24h cache; pages 2..N go direct
- Idempotent — skips addresses with external_valuations.fetched_at < N days
- CLI: --limit / --max-pages / --delay / --address-delay / --skip-recent-days /
  --address (single-address mode)
- Backfills the 160 orphan rows now that PR #526 (parser) + #531 (house_id link)
  are wired

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

@@ -0,0 +1,258 @@
+"""Bulk sweep of Yandex.Недвижимость house-history for all EKB yandex addresses.
+
+Iterates `SELECT DISTINCT address FROM listings WHERE source='yandex' AND region_code=66
+AND address IS NOT NULL AND is_active=TRUE` and for each address calls the same
+cached fetch + save pipeline used by estimator. Idempotent — repeat runs deduplicate
+via house_placement_history UNIQUE (source, ext_item_id) and external_valuations cache.
+
+Why this script:
+  - 160 orphan rows in house_placement_history have house_id=NULL because match
+    pipeline was wired only after PRs #526 + #531 (2026-05-24).
+  - 3704 yandex listings on prod with ~hundreds of unique addresses → after one
+    sweep we get a population of house-history rows linked to houses + with
+    removed_date / total_floors / source_confidence filled.
+
+Setup (same as local-sweep-ekb-yandex.py):
+  ssh -f -N -L 15433:172.20.0.2:5432 gendesign   # tradein-postgres tunnel
+  cd tradein-mvp/backend
+  export DATABASE_URL='postgresql+psycopg://tradein:<PASS>@localhost:15433/tradein'
+  export ENVIRONMENT='local'
+  export GLITCHTIP_DSN=''
+
+Usage:
+  uv run python ../scripts/sweep-yandex-valuation-ekb.py                       # full
+  uv run python ../scripts/sweep-yandex-valuation-ekb.py --limit 10            # smoke
+  uv run python ../scripts/sweep-yandex-valuation-ekb.py --max-pages 3
+  uv run python ../scripts/sweep-yandex-valuation-ekb.py --skip-recent-days 0  # force re-scrape
+  uv run python ../scripts/sweep-yandex-valuation-ekb.py \
+      --address "Екатеринбург, улица Куйбышева, 106"
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import random
+import sys
+import time
+import types
+from pathlib import Path
+
+# Auto-inject backend/ in sys.path
+_BACKEND_DIR = Path(__file__).resolve().parent.parent / "backend"
+if str(_BACKEND_DIR) not in sys.path:
+    sys.path.insert(0, str(_BACKEND_DIR))
+
+# Stub weasyprint (Windows GTK-less import chain compat)
+for _m in (
+    "weasyprint",
+    "weasyprint.text",
+    "weasyprint.text.ffi",
+    "weasyprint.css",
+    "weasyprint.css.targets",
+):
+    sys.modules[_m] = types.ModuleType(_m)
+sys.modules["weasyprint"].HTML = type("HTML", (), {"__init__": lambda self, *a, **kw: None})
+sys.modules["weasyprint"].CSS = type("CSS", (), {"__init__": lambda self, *a, **kw: None})
+
+# Backend imports
+from sqlalchemy import text  # noqa: E402
+
+from app.core.db import SessionLocal  # noqa: E402
+from app.services.estimator import (  # noqa: E402
+    _get_or_fetch_yandex_valuation_cached,
+    _save_yandex_history_items,
+)
+
+
+def _addresses_to_sweep(
+    limit: int | None,
+    skip_recent_days: int,
+    single_address: str | None,
+) -> list[str]:
+    """Pull unique active EKB yandex addresses, optionally skipping recently-cached ones."""
+    if single_address:
+        return [single_address.strip()]
+
+    db = SessionLocal()
+    try:
+        if skip_recent_days > 0:
+            sql = text(
+                """
+                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
+                  AND NOT EXISTS (
+                      SELECT 1 FROM external_valuations ev
+                      WHERE ev.source = 'yandex_valuation'
+                        AND ev.address = l.address
+                        AND ev.fetched_at > NOW() - (:days || ' days')::interval
+                  )
+                ORDER BY l.address
+                """
+            )
+            rows = db.execute(sql, {"days": skip_recent_days}).mappings().all()
+        else:
+            sql = text(
+                """
+                SELECT DISTINCT address
+                FROM listings
+                WHERE source = 'yandex'
+                  AND region_code = 66
+                  AND is_active = TRUE
+                  AND address IS NOT NULL
+                ORDER BY address
+                """
+            )
+            rows = db.execute(sql).mappings().all()
+        addrs = [r["address"] for r in rows if r.get("address")]
+    finally:
+        db.close()
+
+    if limit is not None and limit > 0:
+        addrs = addrs[:limit]
+    return addrs
+
+
+async def _sweep_one_address(
+    address: str,
+    max_pages: int,
+    delay: float,
+) -> tuple[int, int]:
+    """Fetch+save up to max_pages for one address.
+
+    Returns (items_fetched_total, items_saved_total).
+    """
+    total_fetched = 0
+    total_saved = 0
+    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.
+            if page == 1:
+                result = await _get_or_fetch_yandex_valuation_cached(
+                    db,
+                    address=address,
+                )
+            else:
+                from app.services.scrapers.yandex_valuation import YandexValuationScraper
+
+                async with YandexValuationScraper() as scraper:
+                    result = await scraper.fetch_house_history(
+                        address=address,
+                        offer_category="APARTMENT",
+                        offer_type="SELL",
+                        page=page,
+                    )
+            if result is None or not result.history_items:
+                break
+
+            saved = _save_yandex_history_items(db, result)
+            total_fetched += len(result.history_items)
+            total_saved += saved
+            print(
+                f"    page={page}/{max_pages}: items={len(result.history_items)} saved={saved}",
+                flush=True,
+            )
+            # Stop early if Yandex doesn't paginate further (less than full page returned)
+            if len(result.history_items) < 13:
+                break
+        except Exception as e:
+            print(
+                f"    page={page} FAIL: {type(e).__name__}: {e}",
+                flush=True,
+            )
+            break
+        finally:
+            db.close()
+
+        if page < max_pages:
+            await asyncio.sleep(delay * random.uniform(0.85, 1.3))
+
+    return total_fetched, total_saved
+
+
+async def main() -> None:
+    p = argparse.ArgumentParser(
+        description="Bulk sweep Yandex.Недвижимость house-history for EKB yandex addresses"
+    )
+    p.add_argument("--limit", type=int, default=None, help="Cap number of addresses (default all)")
+    p.add_argument(
+        "--max-pages",
+        type=int,
+        default=12,
+        help="Max pages per address (Yandex pagination cap; default 12)",
+    )
+    p.add_argument(
+        "--delay",
+        type=float,
+        default=5.0,
+        help="Sec between page requests within one address (default 5)",
+    )
+    p.add_argument(
+        "--address-delay",
+        type=float,
+        default=15.0,
+        help="Sec between switching to next address (default 15)",
+    )
+    p.add_argument(
+        "--skip-recent-days",
+        type=int,
+        default=1,
+        help="Skip addresses cached within N days (default 1; 0 = no skip)",
+    )
+    p.add_argument(
+        "--address",
+        type=str,
+        default=None,
+        help="Single address mode — sweep only this address",
+    )
+    args = p.parse_args()
+
+    addresses = _addresses_to_sweep(args.limit, args.skip_recent_days, args.address)
+    if not addresses:
+        print("No addresses to sweep (all recently cached or none match query).", flush=True)
+        return
+
+    print(
+        f"=== Sweep {len(addresses)} addresses, max_pages={args.max_pages}, "
+        f"delay={args.delay}s, address-delay={args.address_delay}s, "
+        f"skip-recent={args.skip_recent_days}d ===",
+        flush=True,
+    )
+
+    t0 = time.perf_counter()
+    grand_fetched = 0
+    grand_saved = 0
+    for i, addr in enumerate(addresses, 1):
+        print(f"\n[{i}/{len(addresses)}] {addr[:90]}", flush=True)
+        try:
+            fetched, saved = await _sweep_one_address(addr, args.max_pages, args.delay)
+        except Exception as e:
+            print(f"  FATAL on this address: {type(e).__name__}: {e}", flush=True)
+            continue
+        grand_fetched += fetched
+        grand_saved += saved
+        print(
+            f"  done: fetched={fetched} saved={saved} [grand: f={grand_fetched} s={grand_saved}]",
+            flush=True,
+        )
+        if i < len(addresses):
+            await asyncio.sleep(args.address_delay * random.uniform(0.85, 1.3))
+
+    elapsed = time.perf_counter() - t0
+    print(
+        f"\n=== ALL DONE in {elapsed / 60:.1f} min. "
+        f"addresses={len(addresses)} fetched={grand_fetched} saved={grand_saved} ===",
+        flush=True,
+    )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())