gendesign/tradein-mvp/backend/scripts/backfill_listing_sources.py

"""Backfill listing_sources for ~18k existing listings — retroactive matching.

PR I (commit 7e24ccb) hooked the matching service into `save_listings()` so NEW
scraped listings now write a `listing_sources` row and resolve a canonical
`houses` row per scrape. Existing listings (~18,428: avito 9302, cian 5158,
yandex 3704, n1 264) were never matched — `listing_sources` only has rows
written since PR I shipped.

This script walks every `listings` row that does NOT already have a
`listing_sources` entry for its `(source, source_id)` (or `(source,
dedup_hash)` when `source_id` is NULL — same fallback as PR I) and performs
the same hook inline:

    1. Resolve a canonical `houses` row via:
       a. `house_source` / `house_ext_id` (Avito Houses Catalog, Cian
          newbuilding) when the listing carries them — these are stable
          source-side house identifiers that match `houses.(source,
          ext_house_id)` directly.
       b. Otherwise `match_or_create_house()` Tier 0-3 with the listing's
          `address` / `lat` / `lon` / cadastrals.
    2. `upsert_listing_source()` with method='backfill', confidence=0.9
       (slightly below real-time `source_link` 1.0 so audits can distinguish).
    3. UPDATE `listings.house_id_fk = matched.house_id` when the listing
       didn't already have one — keeps the direct JOIN path consistent with
       the matching graph.

Per-row SAVEPOINT (`db.begin_nested()`) per `.claude/rules/backend.md` so a
single bad row never aborts the whole batch.

Idempotent / resumable:
    - Source query: `WHERE NOT EXISTS (SELECT 1 FROM listing_sources ls
      WHERE ls.ext_source = ... AND ls.ext_id = ...)`. Re-runs skip already-
      processed rows naturally.
    - `upsert_listing_source` uses `ON CONFLICT (ext_source, ext_id) DO
      UPDATE` — also safe.

No network calls — Yandex Geocoder is blocked on prod right now and the
matching service is purely in-database.

Usage:
    DATABASE_URL=postgresql+psycopg://... \\
        python -m scripts.backfill_listing_sources --batch-size 500

    # Canary first
    python -m scripts.backfill_listing_sources --limit 100 --dry-run

    # Limit to a single source for staged rollout
    python -m scripts.backfill_listing_sources --source avito --limit 1000
"""

from __future__ import annotations

import argparse
import hashlib
import logging
import sys
from collections import defaultdict
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

from sqlalchemy import text
from sqlalchemy.orm import Session

# Allow running both as `python -m scripts.backfill_listing_sources` (preferred)
# and as a stand-alone file (fallback for adhoc invocation).
try:
    from app.core.db import SessionLocal  # type: ignore[import-not-found]
    from app.services.matching import (  # type: ignore[import-not-found]
        match_or_create_house,
        upsert_listing_source,
    )
except ImportError:  # pragma: no cover — fallback for adhoc invocation
    sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
    from app.core.db import SessionLocal
    from app.services.matching import match_or_create_house, upsert_listing_source

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s %(name)s %(message)s",
)
logger = logging.getLogger("backfill_listing_sources")

# Lower confidence than the real-time `source_link` (1.0) — backfilled rows
# weren't observed inline with the scrape, so audits can tell the two apart
# via `listing_sources.matched_method = 'backfill'` (the canonical filter)
# or via `confidence < 1.0` (the secondary check).
_BACKFILL_CONFIDENCE = 0.9
_BACKFILL_METHOD = "backfill"


# ---------------------------------------------------------------------------
# Domain types
# ---------------------------------------------------------------------------


@dataclass
class ListingRow:
    """One listing pulled by the source query — all fields we need for matching.

    Mirrors the subset of ScrapedLot fields that match_or_create_house and
    upsert_listing_source consume. Kept as a plain dataclass (no Pydantic) so
    the script has zero parse overhead for 18k rows.
    """

    id: int
    source: str
    source_id: str | None
    source_url: str
    dedup_hash: str
    address: str | None
    lat: float | None
    lon: float | None
    rooms: int | None
    area_m2: float | None
    floor: int | None
    price_rub: int
    year_built: int | None
    cadastral_number: str | None
    building_cadastral_number: str | None
    house_source: str | None
    house_ext_id: str | None
    house_url: str | None
    house_id_fk: int | None


@dataclass
class Stats:
    """Aggregate counters — printed per batch and at the end of the run."""

    processed: int = 0
    matched: int = 0  # listing_sources row upserted (with or without house)
    house_resolved: int = 0
    house_failed: int = 0
    skipped: int = 0  # rare: no address/lat/lon/source_id at all
    errors: int = 0
    by_source: dict[str, dict[str, int]] = field(default_factory=dict)

    def bump(self, source: str, key: str) -> None:
        """Increment a per-source counter (matched, house_resolved, errors...)."""
        bucket = self.by_source.setdefault(source, defaultdict(int))
        bucket[key] += 1


# ---------------------------------------------------------------------------
# Source query — stream listings without a listing_sources row
# ---------------------------------------------------------------------------

# All listings columns the matching hook consumes. Keep field list in sync
# with ListingRow.
_LISTING_COLUMNS = (
    "id, source, source_id, source_url, dedup_hash, "
    "address, lat, lon, rooms, area_m2, floor, price_rub, year_built, "
    "cadastral_number, building_cadastral_number, "
    "house_source, house_ext_id, house_url, house_id_fk"
)


def _build_select_sql(*, source: str | None, batch_size: int) -> str:
    """Streaming SELECT — skips rows already in listing_sources via NOT EXISTS.

    The NOT EXISTS predicate matches the same (ext_source, ext_id) shape used
    by `upsert_listing_source` so re-runs are zero-cost: any row PR I or a
    previous backfill batch already inserted is excluded.

    `ext_id` here mirrors the PR I hook's fallback chain: prefer
    `listings.source_id`, fall back to `dedup_hash`. Yandex listings that
    lack a stable `source_id` are still uniquely identifiable via
    `dedup_hash` (sha256 of source + source_url + price).
    """
    where_source = ""
    if source is not None:
        # Bind parameter — caller still passes :source. The string is just for
        # SQL composition; psycopg fills :source from kwargs.
        where_source = " AND source = :source "
    sql = (
        f"SELECT {_LISTING_COLUMNS} "
        f"FROM listings "
        f"WHERE id > :after_id "
        f"  {where_source} "
        f"  AND NOT EXISTS ("
        f"      SELECT 1 FROM listing_sources ls "
        f"      WHERE ls.ext_source = listings.source "
        f"        AND ls.ext_id = COALESCE(listings.source_id, listings.dedup_hash)"
        f"  ) "
        f"ORDER BY id "
        f"LIMIT CAST(:limit AS int)"
    )
    return sql


def _fetch_batch(
    db: Session, *, after_id: int, batch_size: int, source: str | None
) -> list[ListingRow]:
    """Pull the next batch of unmatched listings ordered by id."""
    sql = _build_select_sql(source=source, batch_size=batch_size)
    params: dict[str, Any] = {"after_id": after_id, "limit": batch_size}
    if source is not None:
        params["source"] = source

    rows = db.execute(text(sql), params).mappings().all()
    return [
        ListingRow(
            id=r["id"],
            source=r["source"],
            source_id=r["source_id"],
            source_url=r["source_url"],
            dedup_hash=r["dedup_hash"],
            address=r["address"],
            lat=r["lat"],
            lon=r["lon"],
            rooms=r["rooms"],
            area_m2=float(r["area_m2"]) if r["area_m2"] is not None else None,
            floor=r["floor"],
            price_rub=r["price_rub"],
            year_built=r["year_built"],
            cadastral_number=r["cadastral_number"],
            building_cadastral_number=r["building_cadastral_number"],
            house_source=r["house_source"],
            house_ext_id=r["house_ext_id"],
            house_url=r["house_url"],
            house_id_fk=r["house_id_fk"],
        )
        for r in rows
    ]


# ---------------------------------------------------------------------------
# Per-row work
# ---------------------------------------------------------------------------


def _ext_id_for(row: ListingRow) -> str:
    """Mirror PR I hook: prefer source_id, fall back to dedup_hash.

    `dedup_hash` is already a sha256 hex string. Yandex listings without
    stable source_id rely on this fallback — same hash on re-scrape, so
    listing_sources stays unique per logical listing.
    """
    if row.source_id:
        return row.source_id
    if row.dedup_hash:
        return row.dedup_hash
    # Defensive — should not happen since dedup_hash is NOT NULL UNIQUE in
    # the schema, but compute one on the fly so we never write '' as ext_id.
    h = hashlib.sha256(f"{row.source}|{row.source_url}|{row.price_rub}".encode()).hexdigest()
    return h


def _link_listing_to_house(
    db: Session, row: ListingRow, *, dry_run: bool
) -> tuple[int | None, bool]:
    """Resolve canonical house + upsert listing_sources for one listing.

    Returns:
        (house_id_or_None, house_resolved_bool).
        `house_resolved=True` means match_or_create_house produced a non-null
        id (either direct via house_source/ext_id or via Tier 0-3 fallback).
    """
    ext_id = _ext_id_for(row)

    # House resolution — only attempted if we have address or coords. Without
    # them, match_or_create_house Tier 2 (fingerprint) trivially mismatches and
    # Tier 3 (geo-proximity) is impossible — we'd be creating an island house
    # row keyed off a synthetic ext_id with no real data.
    house_id: int | None = None
    house_resolved = False
    if row.address or (row.lat is not None and row.lon is not None):
        # Prefer the scraped house_source/house_ext_id (Avito Houses Catalog,
        # Cian newbuilding) — those map 1:1 to `houses.(source, ext_house_id)`
        # via Tier 1 (`source_exact`) and avoid fingerprint/geo lookups.
        h_src = row.house_source or row.source
        h_ext = row.house_ext_id or ext_id
        if dry_run:
            # Don't touch the DB. Pretend a house was resolved if scraped
            # extras are present — used by the dry-run summary only.
            house_id = row.house_id_fk
            house_resolved = (
                row.house_id_fk is not None
                or row.house_source is not None
                or (row.address is not None)
            )
        else:
            try:
                with db.begin_nested():
                    house_id, _conf, _method = match_or_create_house(
                        db,
                        ext_source=h_src,
                        ext_id=h_ext,
                        address=row.address,
                        lat=row.lat,
                        lon=row.lon,
                        year_built=row.year_built,
                        # Кадастр КВАРТИРЫ (row.cadastral_number) в ключ дома не идёт (#2674).
                        building_cadastral_number=row.building_cadastral_number,
                        source_url=row.house_url or row.source_url,
                    )
                    house_resolved = house_id is not None
            except Exception as e:
                # Fall through to listing-only upsert — the listing_sources row
                # is still useful (e.g. for a later backfill that can geocode).
                logger.warning(
                    "backfill:house_match_failed listing_id=%s source=%s ext_id=%s: %s",
                    row.id,
                    row.source,
                    ext_id,
                    e,
                )
                house_id = None
                house_resolved = False

    if not dry_run:
        upsert_listing_source(
            db,
            listing_id=row.id,
            ext_source=row.source,
            ext_id=str(ext_id),
            method=_BACKFILL_METHOD,
            confidence=_BACKFILL_CONFIDENCE,
            price_rub=row.price_rub,
            area_m2=row.area_m2,
            floor=row.floor,
            rooms_count=row.rooms,
            source_url=row.source_url,
            source_data={"house_id": house_id} if house_id else None,
        )

        # Mirror the matching graph into listings.house_id_fk so direct
        # `JOIN houses ON house_id_fk = id` keeps working. Only updates when
        # the row didn't already have a linkage (preserves any prior 063
        # backfill).
        if house_id is not None and row.house_id_fk is None:
            db.execute(
                text(
                    "UPDATE listings "
                    "   SET house_id_fk = CAST(:hid AS bigint) "
                    " WHERE id = CAST(:lid AS bigint) "
                    "   AND house_id_fk IS NULL"
                ),
                {"hid": house_id, "lid": row.id},
            )

    return house_id, house_resolved


def _process_row(db: Session, row: ListingRow, *, dry_run: bool, stats: Stats) -> None:
    """Wrap _link_listing_to_house in a per-row SAVEPOINT.

    Per backend.md `Bug_Pzz_Loader_Missing_Savepoint_May14`: never bare
    `db.rollback()` inside a row loop — it kills the surrounding tx and the
    counters get out of sync with what actually committed.
    """
    stats.processed += 1
    stats.bump(row.source, "processed")

    # Skip listings that lack any signal we can use — without source_id we'd
    # have to rely on dedup_hash + no address → effectively orphan rows in
    # listing_sources. Still upsert via dedup_hash but mark as skipped from
    # house-resolution counts.
    if not row.source_id and not row.address and (row.lat is None or row.lon is None):
        stats.skipped += 1
        stats.bump(row.source, "skipped")

    try:
        if dry_run:
            # No SAVEPOINT in dry-run — we don't touch the DB at all.
            house_id, house_resolved = _link_listing_to_house(db, row, dry_run=True)
        else:
            with db.begin_nested():
                house_id, house_resolved = _link_listing_to_house(db, row, dry_run=False)

        stats.matched += 1
        stats.bump(row.source, "matched")
        if house_resolved:
            stats.house_resolved += 1
            stats.bump(row.source, "house_resolved")
        else:
            stats.house_failed += 1
            stats.bump(row.source, "house_failed")

        logger.debug(
            "backfill ok listing_id=%s source=%s ext_id=%s house_id=%s",
            row.id,
            row.source,
            _ext_id_for(row),
            house_id,
        )
    except Exception as e:
        stats.errors += 1
        stats.bump(row.source, "errors")
        logger.warning(
            "backfill failed listing_id=%s source=%s: %s",
            row.id,
            row.source,
            e,
        )


# ---------------------------------------------------------------------------
# Driver
# ---------------------------------------------------------------------------


def _coverage_summary(db: Session) -> dict[str, Any]:
    """Per-source coverage % after the run — useful for the final log line."""
    rows = (
        db.execute(
            text(
                "SELECT "
                "  l.source, "
                "  COUNT(*) AS total, "
                "  COUNT(*) FILTER ("
                "    WHERE EXISTS ("
                "      SELECT 1 FROM listing_sources ls "
                "      WHERE ls.ext_source = l.source "
                "        AND ls.ext_id = COALESCE(l.source_id, l.dedup_hash)"
                "    )"
                "  ) AS linked "
                "FROM listings l "
                "GROUP BY l.source "
                "ORDER BY total DESC"
            )
        )
        .mappings()
        .all()
    )
    return {
        r["source"]: {
            "total": int(r["total"]),
            "linked": int(r["linked"]),
            "pct": (float(r["linked"]) / float(r["total"]) * 100.0) if r["total"] else 0.0,
        }
        for r in rows
    }


def run_backfill(
    db: Session,
    *,
    batch_size: int,
    limit: int | None,
    source: str | None,
    dry_run: bool,
) -> Stats:
    """Main driver — streams batches and writes listing_sources rows.

    Args:
        db: SQLAlchemy Session.
        batch_size: how many rows to fetch per SELECT. 500 keeps memory flat
            and gives a checkpoint every commit.
        limit: stop after processing this many total rows (--limit). None =
            run to exhaustion.
        source: filter to one ext_source ('avito'/'cian'/'yandex').
            None = all sources (historical 'n1' rows still processed in a full run).
        dry_run: skip all writes, just count.
    """
    stats = Stats()
    after_id = 0
    batch_idx = 0

    while True:
        # Stop early if --limit reached.
        if limit is not None and stats.processed >= limit:
            logger.info(
                "limit reached: stopping (processed=%d, limit=%d)",
                stats.processed,
                limit,
            )
            break

        effective_size = batch_size
        if limit is not None:
            effective_size = min(batch_size, limit - stats.processed)
            if effective_size <= 0:
                break

        batch = _fetch_batch(db, after_id=after_id, batch_size=effective_size, source=source)
        if not batch:
            logger.info("no more rows — done")
            break

        batch_idx += 1
        for row in batch:
            _process_row(db, row, dry_run=dry_run, stats=stats)
            after_id = max(after_id, row.id)

        # Commit the batch — every 500 rows under default config. Dry-run
        # never wrote anything so the commit is a no-op (cheap).
        if not dry_run:
            db.commit()

        logger.info(
            "batch %d done: size=%d total processed=%d matched=%d "
            "house_resolved=%d house_failed=%d errors=%d",
            batch_idx,
            len(batch),
            stats.processed,
            stats.matched,
            stats.house_resolved,
            stats.house_failed,
            stats.errors,
        )

    return stats


def _log_summary(stats: Stats, db: Session, *, dry_run: bool) -> None:
    """Final per-source breakdown + overall coverage."""
    logger.info("=" * 72)
    logger.info(
        "backfill done (dry_run=%s): processed=%d matched=%d "
        "house_resolved=%d house_failed=%d skipped=%d errors=%d",
        dry_run,
        stats.processed,
        stats.matched,
        stats.house_resolved,
        stats.house_failed,
        stats.skipped,
        stats.errors,
    )
    for src, bucket in sorted(stats.by_source.items()):
        logger.info(
            "  %-10s processed=%d matched=%d house_resolved=%d house_failed=%d errors=%d",
            src,
            bucket.get("processed", 0),
            bucket.get("matched", 0),
            bucket.get("house_resolved", 0),
            bucket.get("house_failed", 0),
            bucket.get("errors", 0),
        )

    # Coverage reflects the on-disk state. In dry-run mode this still shows
    # pre-existing rows from PR I — useful to gauge what the next real run
    # would do.
    coverage = _coverage_summary(db)
    logger.info("final listing_sources coverage:")
    for src, c in sorted(coverage.items()):
        logger.info("  %-10s %d / %d (%.1f%%)", src, c["linked"], c["total"], c["pct"])


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------


def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    """argparse setup, factored out for testability."""
    p = argparse.ArgumentParser(
        description=(
            "PR J — backfill listing_sources for existing listings. Resolves "
            "canonical houses + upserts listing_sources rows for every "
            "listing that does not already have one. Idempotent on re-run."
        ),
    )
    p.add_argument(
        "--batch-size",
        type=int,
        default=500,
        help="Rows pulled per SELECT (default: 500). Each batch commits once.",
    )
    p.add_argument(
        "--limit",
        type=int,
        default=None,
        help=(
            "Cap on total rows processed across all batches. Useful for "
            "canary runs (e.g. --limit 100 before letting the full job loose)."
        ),
    )
    p.add_argument(
        "--source",
        choices=("avito", "cian", "yandex"),
        default=None,
        help=(
            "Restrict to one ext_source for staged rollout. Default: all sources mixed in id order."
        ),
    )
    p.add_argument(
        "--dry-run",
        action="store_true",
        help="Log what would be done; no DB writes.",
    )
    return p.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    """CLI entry point. Returns the number of rows processed this run."""
    args = _parse_args(argv)
    logger.info(
        "starting backfill: batch_size=%d limit=%s source=%s dry_run=%s",
        args.batch_size,
        args.limit if args.limit is not None else "all",
        args.source or "all",
        args.dry_run,
    )

    db = SessionLocal()
    try:
        stats = run_backfill(
            db,
            batch_size=args.batch_size,
            limit=args.limit,
            source=args.source,
            dry_run=args.dry_run,
        )
        _log_summary(stats, db, dry_run=args.dry_run)
        return stats.processed
    finally:
        db.close()


if __name__ == "__main__":  # pragma: no cover
    sys.exit(0 if main() >= 0 else 1)