From 77cda24b8a0922556a5f71ff7f7fd64e96f7ad50 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Sun, 24 May 2026 14:37:24 +0000 Subject: [PATCH] feat(tradein): bulk sweep CLI for Yandex valuation house-history (EKB) (#533) --- .../scripts/sweep-yandex-valuation-ekb.py | 258 ++++++++++++++++++ 1 file changed, 258 insertions(+) create mode 100644 tradein-mvp/scripts/sweep-yandex-valuation-ekb.py diff --git a/tradein-mvp/scripts/sweep-yandex-valuation-ekb.py b/tradein-mvp/scripts/sweep-yandex-valuation-ekb.py new file mode 100644 index 00000000..38fb366c --- /dev/null +++ b/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:@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())