"""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:@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())