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

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