tradein-mvp/scripts/backfill-yandex-addresses-ekb.py

@@ -0,0 +1,173 @@
+"""Backfill listings.address with full address (street + house number) for
+active EKB yandex listings that currently have street-only addresses.
+
+Why: Yandex SERP returns only street (e.g. 'улица Горького') in card snippet.
+Detail page <title> has the full address ('Екатеринбург, улица Горького, 36').
+With house number we can call /otsenka-kvartiry-po-adresu-onlayn/ and get the
+building's history.
+
+Usage:
+  cd tradein-mvp/backend
+  uv run python ../scripts/backfill-yandex-addresses-ekb.py --limit 5         # smoke
+  uv run python ../scripts/backfill-yandex-addresses-ekb.py                   # full
+  uv run python ../scripts/backfill-yandex-addresses-ekb.py --dry-run         # no DB write
+
+Requires SSH tunnel + DATABASE_URL env (see sweep-yandex-valuation-ekb.py docstring).
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import random
+import re
+import sys
+import time
+import types
+from pathlib import Path
+
+# sys.path injection
+_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)
+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})
+
+from curl_cffi.requests import AsyncSession  # noqa: E402
+from sqlalchemy import text  # noqa: E402
+
+from app.core.db import SessionLocal  # noqa: E402
+
+# Extract '<City>, <addr> — id <number>' from <title>
+RE_TITLE_ADDRESS = re.compile(
+    r"<title>[^<]*?—\s*"
+    r"(?:[^—,]*?,\s*)?"  # optional ЖК/жилой комплекс prefix
+    r"(Екатеринбург,\s*[^<]+?)"
+    r"\s*—\s*id\s*\d+\s*</title>",
+    re.IGNORECASE,
+)
+
+
+def _eligible_listings(limit: int | None) -> list[dict]:
+    """Active EKB yandex listings without house number in address."""
+    db = SessionLocal()
+    try:
+        sql = text(
+            """
+            SELECT id, source_id, address, source_url
+            FROM listings
+            WHERE source = 'yandex'
+              AND region_code = 66
+              AND is_active = TRUE
+              AND address IS NOT NULL
+              AND NOT (address ~ ',\\s*\\d+')
+              AND source_url IS NOT NULL
+            ORDER BY scraped_at DESC
+            """
+        )
+        rows = db.execute(sql).mappings().all()
+        out = [dict(r) for r in rows]
+    finally:
+        db.close()
+    if limit is not None and limit > 0:
+        out = out[:limit]
+    return out
+
+
+def _update_listing_address(listing_id: int, new_address: str) -> None:
+    db = SessionLocal()
+    try:
+        db.execute(
+            text("UPDATE listings SET address = :addr WHERE id = :id"),
+            {"addr": new_address, "id": listing_id},
+        )
+        db.commit()
+    finally:
+        db.close()
+
+
+async def _fetch_address(session: AsyncSession, url: str) -> str | None:
+    """Fetch detail page, extract full address from <title>."""
+    try:
+        r = await session.get(url, timeout=30)
+    except Exception as e:
+        print(f"    fetch FAIL: {type(e).__name__}: {e}", flush=True)
+        return None
+    if r.status_code != 200:
+        print(f"    HTTP {r.status_code}", flush=True)
+        return None
+    html = r.text.replace("\xa0", " ")
+    m = RE_TITLE_ADDRESS.search(html)
+    if not m:
+        return None
+    addr = m.group(1).strip().rstrip(",").strip()
+    return addr if addr else None
+
+
+async def main() -> None:
+    p = argparse.ArgumentParser(
+        description="Backfill yandex listings.address with house number from detail page"
+    )
+    p.add_argument("--limit", type=int, default=None, help="Cap items (smoke testing)")
+    p.add_argument("--delay", type=float, default=3.0, help="Sec between requests")
+    p.add_argument("--dry-run", action="store_true", help="Don't UPDATE DB")
+    args = p.parse_args()
+
+    items = _eligible_listings(args.limit)
+    if not items:
+        print("No eligible listings.", flush=True)
+        return
+
+    print(
+        f"=== Backfill {len(items)} listings (delay={args.delay}s dry_run={args.dry_run}) ===",
+        flush=True,
+    )
+
+    t0 = time.perf_counter()
+    enriched = 0
+    skipped = 0
+    failed = 0
+
+    async with AsyncSession(impersonate="chrome120") as session:
+        for i, item in enumerate(items, 1):
+            url = item["source_url"]
+            old_addr = item["address"]
+            print(
+                f"[{i}/{len(items)}] id={item['id']} old={old_addr!r}",
+                flush=True,
+            )
+            new_addr = await _fetch_address(session, url)
+            if new_addr is None:
+                print("    -> no address extracted", flush=True)
+                failed += 1
+            elif new_addr == old_addr:
+                print("    -> unchanged", flush=True)
+                skipped += 1
+            else:
+                print(f"    -> {new_addr!r}", flush=True)
+                if not args.dry_run:
+                    _update_listing_address(item["id"], new_addr)
+                enriched += 1
+            if i < len(items):
+                await asyncio.sleep(args.delay * random.uniform(0.85, 1.3))
+
+    elapsed = time.perf_counter() - t0
+    print(
+        f"\n=== DONE in {elapsed / 60:.1f} min. "
+        f"enriched={enriched} skipped={skipped} failed={failed} ===",
+        flush=True,
+    )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())