feat(tradein): backfill 18k existing listings (PR J) (#596)

tradein-mvp/backend/scripts/README.md

@@ -148,3 +148,80 @@ uv run python -m scripts.backfill_house_coords --batch 2026-05-27_backfill
   by `audit_address_mismatch.py`).
 - `address_audit_report.sql` — psql-driven post-run summary (p50/p75/p95
   distance, top-20 outliers, per-district breakdown).
+
+---
+
+## Matching backfill (PR J)
+
+### `backfill_listing_sources.py` — retroactive matching for ~18k listings
+
+PR I (commit 7e24ccb) hooked the matching service into `save_listings()` so
+every **new** scrape now writes a `listing_sources` row + resolves a canonical
+`houses` row. This script does the same work retroactively for all
+**existing** listings — `listing_sources` only had rows from new scrapes
+post-PR I.
+
+What it does per row:
+
+1. `match_or_create_house()` (Tier 0-3) — uses `listings.house_source` /
+   `house_ext_id` when present (Avito Houses Catalog, Cian newbuilding),
+   else falls back to address/lat/lon/cadastrals.
+2. `upsert_listing_source()` with `method='backfill'`, `confidence=0.9`
+   (vs real-time `source_link` 1.0 — distinguishes the two in audits).
+3. `UPDATE listings.house_id_fk` when the row didn't already have one.
+
+```bash
+# Canary
+DATABASE_URL=postgresql+psycopg://... \
+    uv run python -m scripts.backfill_listing_sources \
+        --limit 100 --dry-run
+
+# Real run, one source at a time (staged rollout)
+uv run python -m scripts.backfill_listing_sources --source avito
+
+# Full run
+uv run python -m scripts.backfill_listing_sources --batch-size 500
+```
+
+**Idempotent / resumable** — the source query is
+`WHERE NOT EXISTS (SELECT 1 FROM listing_sources ls WHERE ls.ext_source =
+listings.source AND ls.ext_id = COALESCE(listings.source_id,
+listings.dedup_hash))`. Re-runs only pick up rows still missing from
+`listing_sources`. `upsert_listing_source` adds a second layer of safety via
+`ON CONFLICT (ext_source, ext_id) DO UPDATE`.
+
+**No network calls** — pure in-DB matching (Yandex Geocoder is blocked on
+prod, and `match_or_create_house` does not call it anyway).
+
+**Per-row SAVEPOINT** (`db.begin_nested()`) per `.claude/rules/backend.md` —
+one bad row never aborts the surrounding batch.
+
+**Expected output (PR J initial run):**
+
+| Source | Rows  | Expected matched | Notes                                          |
+|--------|-------|-------------------|------------------------------------------------|
+| avito  | 9302  | 9000+             | Many carry `house_source`/`house_ext_id`       |
+| cian   | 5158  | 5000+             | Most carry `house_source`/`house_ext_id`       |
+| yandex | 3704  | 3700+             | No `source_id` → uses `dedup_hash` as `ext_id` |
+| n1     | 264   | 264               | All have address/coords                        |
+
+**Expected duration**: rough estimate ~5-15 minutes on prod for ~18k rows
+(advisory-lock + 1-3 DB roundtrips per listing for Tier 0-3, ~500 commit
+checkpoints at default batch size). Run with `--limit 100` first to
+calibrate, then let the full job loose.
+
+Final summary in the log includes per-source coverage % so you can verify
+the run landed:
+
+```
+backfill done (dry_run=False): processed=18428 matched=18428
+  house_resolved=18200 house_failed=228 skipped=0 errors=0
+  avito      processed=9302 matched=9302 house_resolved=9290 ...
+  cian       processed=5158 matched=5158 house_resolved=5100 ...
+  yandex     processed=3704 matched=3704 house_resolved=3540 ...
+  n1         processed=264  matched=264  house_resolved=270  ...
+final listing_sources coverage:
+  avito      9302 / 9302 (100.0%)
+  cian       5158 / 5158 (100.0%)
+  ...
+```

tradein-mvp/backend/scripts/backfill_listing_sources.py

@@ -0,0 +1,615 @@
+"""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
+    kadastr_num: 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, kadastr_num, "
+    "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"],
+            kadastr_num=r["kadastr_num"],
+            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,
+                        building_cadastral_number=row.building_cadastral_number,
+                        cadastral_number=row.cadastral_number or row.kadastr_num,
+                        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'/'n1').
+            None = all sources.
+        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", "n1"),
+        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)

tradein-mvp/backend/tests/scripts/test_backfill_listing_sources.py

@@ -0,0 +1,476 @@
+"""Unit tests for PR J — backfill_listing_sources.py.
+
+Coverage:
+    - Happy path: one listing → upsert_listing_source called + house resolved.
+    - Idempotency: re-run skips listings already in listing_sources via the
+      NOT EXISTS predicate (verified by feeding the second SELECT a 0-row
+      result and confirming no hooks fire).
+    - house_source/house_ext_id linkage: when the listing carries them they
+      are passed straight to match_or_create_house as ext_source/ext_id.
+    - house match failure: upsert_listing_source still fires with
+      source_data=None so the listing_sources row is created without a
+      house linkage.
+    - --dry-run: no writes at all (no db.execute writes, no match service
+      calls).
+    - --limit: caps total rows processed even when more would otherwise be
+      returned by the next batch.
+    - _ext_id_for fallback order — source_id → dedup_hash → computed sha.
+
+No real Postgres. The DB session is a MagicMock that records execute() calls
+and short-circuits matching service calls via patch().
+"""
+
+from __future__ import annotations
+
+import os
+from contextlib import contextmanager
+from unittest.mock import MagicMock, patch
+
+# Settings requires DATABASE_URL at import time — set dummy DSN before any
+# `app.*` import (same pattern as test_audit_address_mismatch.py).
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost/test_db")
+
+import pytest
+
+from scripts.backfill_listing_sources import (
+    ListingRow,
+    Stats,
+    _build_select_sql,
+    _ext_id_for,
+    _link_listing_to_house,
+    _process_row,
+    main,
+    run_backfill,
+)
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _row(**overrides) -> ListingRow:
+    """Minimal valid ListingRow with one source_id by default."""
+    base = {
+        "id": 1,
+        "source": "avito",
+        "source_id": "100",
+        "source_url": "https://www.avito.ru/ekaterinburg/kvartiry/1-k_100",
+        "dedup_hash": "a" * 64,
+        "address": "Екатеринбург, Ленина, 1",
+        "lat": 56.84,
+        "lon": 60.60,
+        "rooms": 2,
+        "area_m2": 50.0,
+        "floor": 5,
+        "price_rub": 4_500_000,
+        "year_built": 2010,
+        "cadastral_number": None,
+        "building_cadastral_number": None,
+        "kadastr_num": None,
+        "house_source": None,
+        "house_ext_id": None,
+        "house_url": None,
+        "house_id_fk": None,
+    }
+    base.update(overrides)
+    return ListingRow(**base)
+
+
+def _make_db_mock(batches: list[list[dict]] | None = None) -> MagicMock:
+    """MagicMock Session that returns the next batch on each SELECT.
+
+    `batches` — list of row-dict lists; each subsequent call to db.execute()
+    with a `FROM listings` query returns the next one. After exhaustion an
+    empty list is returned so the driver stops.
+
+    The mock honors the :limit bind value so test_run_backfill_limit_caps_*
+    can rely on the driver's effective_size = min(batch_size, limit-processed)
+    logic without depending on real SQL LIMIT behaviour.
+
+    Coverage query (`GROUP BY l.source`) — separate side-effect for the final
+    log line; we return a single-row aggregate so the summary code path runs.
+    """
+    batches = batches or []
+    pending: list[list[dict]] = list(batches)
+
+    db = MagicMock()
+
+    @contextmanager
+    def _nested():
+        yield MagicMock()
+
+    db.begin_nested.side_effect = _nested
+    db.commit = MagicMock()
+    db.rollback = MagicMock()
+    db.close = MagicMock()
+
+    def execute_side_effect(sql, params=None):
+        sql_str = str(sql)
+        result = MagicMock()
+        if "GROUP BY l.source" in sql_str:
+            # _coverage_summary
+            result.mappings.return_value.all.return_value = [
+                {"source": "avito", "total": 10, "linked": 10},
+            ]
+            return result
+        if "FROM listings" in sql_str:
+            batch = pending.pop(0) if pending else []
+            # Honor SQL LIMIT — driver computes effective_size and binds it.
+            if params is not None and "limit" in params:
+                batch = batch[: params["limit"]]
+            result.mappings.return_value.all.return_value = batch
+            return result
+        # UPDATE listings.house_id_fk — no-op
+        if "UPDATE listings" in sql_str:
+            return result
+        return result
+
+    db.execute.side_effect = execute_side_effect
+    return db
+
+
+def _row_dict(row: ListingRow) -> dict:
+    """Convert ListingRow → dict form returned by db.execute(...).mappings()."""
+    return {
+        "id": row.id,
+        "source": row.source,
+        "source_id": row.source_id,
+        "source_url": row.source_url,
+        "dedup_hash": row.dedup_hash,
+        "address": row.address,
+        "lat": row.lat,
+        "lon": row.lon,
+        "rooms": row.rooms,
+        "area_m2": row.area_m2,
+        "floor": row.floor,
+        "price_rub": row.price_rub,
+        "year_built": row.year_built,
+        "cadastral_number": row.cadastral_number,
+        "building_cadastral_number": row.building_cadastral_number,
+        "kadastr_num": row.kadastr_num,
+        "house_source": row.house_source,
+        "house_ext_id": row.house_ext_id,
+        "house_url": row.house_url,
+        "house_id_fk": row.house_id_fk,
+    }
+
+
+@pytest.fixture
+def _patch_matching():
+    """Stub matching service for all tests."""
+    with (
+        patch("scripts.backfill_listing_sources.match_or_create_house") as m_house,
+        patch("scripts.backfill_listing_sources.upsert_listing_source") as m_link,
+    ):
+        m_house.return_value = (501, 1.0, "new")
+        m_link.return_value = None
+        yield {"house": m_house, "link": m_link}
+
+
+# ---------------------------------------------------------------------------
+# _ext_id_for — fallback order
+# ---------------------------------------------------------------------------
+
+
+def test_ext_id_prefers_source_id():
+    r = _row(source_id="abc", dedup_hash="d" * 64)
+    assert _ext_id_for(r) == "abc"
+
+
+def test_ext_id_falls_back_to_dedup_hash():
+    r = _row(source_id=None, dedup_hash="d" * 64)
+    assert _ext_id_for(r) == "d" * 64
+
+
+def test_ext_id_computed_when_both_missing():
+    """Defensive — listings.dedup_hash is NOT NULL UNIQUE, but if it's somehow
+    blank we recompute a sha256 so we never write '' as ext_id."""
+    r = _row(source_id=None, dedup_hash="")
+    out = _ext_id_for(r)
+    assert len(out) == 64  # sha256 hex
+
+
+# ---------------------------------------------------------------------------
+# _build_select_sql — NOT EXISTS predicate + optional source filter
+# ---------------------------------------------------------------------------
+
+
+def test_select_sql_excludes_listings_already_in_listing_sources():
+    sql = _build_select_sql(source=None, batch_size=500)
+    assert "NOT EXISTS" in sql
+    assert "listing_sources" in sql
+    # The NOT EXISTS body must match on COALESCE(source_id, dedup_hash) so
+    # Yandex (no source_id) listings are still uniquely identifiable.
+    assert "COALESCE(listings.source_id, listings.dedup_hash)" in sql
+    # Default — no `source = :source` clause.
+    assert ":source" not in sql
+
+
+def test_select_sql_filters_by_source_when_provided():
+    sql = _build_select_sql(source="avito", batch_size=500)
+    assert "source = :source" in sql
+
+
+# ---------------------------------------------------------------------------
+# Happy path — one listing → 1 link + 1 house resolution
+# ---------------------------------------------------------------------------
+
+
+def test_link_listing_happy_path(_patch_matching):
+    """Single listing with address → match_or_create_house + upsert called once."""
+    db = _make_db_mock()
+    row = _row(id=42, source_id="100")
+
+    house_id, resolved = _link_listing_to_house(db, row, dry_run=False)
+
+    assert house_id == 501
+    assert resolved is True
+    assert _patch_matching["house"].call_count == 1
+    assert _patch_matching["link"].call_count == 1
+
+    # upsert kwargs reflect the listing context
+    kwargs = _patch_matching["link"].call_args.kwargs
+    assert kwargs["listing_id"] == 42
+    assert kwargs["ext_source"] == "avito"
+    assert kwargs["ext_id"] == "100"
+    assert kwargs["method"] == "backfill"
+    assert kwargs["confidence"] == 0.9
+    assert kwargs["price_rub"] == 4_500_000
+    assert kwargs["area_m2"] == 50.0
+    assert kwargs["floor"] == 5
+    assert kwargs["rooms_count"] == 2
+    assert kwargs["source_data"] == {"house_id": 501}
+
+
+def test_link_listing_no_address_no_coords_skips_house_match(_patch_matching):
+    """No address + no coords → no house resolution, but listing_sources still upserted."""
+    db = _make_db_mock()
+    row = _row(
+        source_id="200",
+        address=None,
+        lat=None,
+        lon=None,
+    )
+
+    house_id, resolved = _link_listing_to_house(db, row, dry_run=False)
+
+    assert house_id is None
+    assert resolved is False
+    assert _patch_matching["house"].call_count == 0
+    assert _patch_matching["link"].call_count == 1
+    kwargs = _patch_matching["link"].call_args.kwargs
+    assert kwargs["source_data"] is None
+
+
+# ---------------------------------------------------------------------------
+# house_source / house_ext_id direct linkage
+# ---------------------------------------------------------------------------
+
+
+def test_link_listing_uses_house_source_ext_id_when_present(_patch_matching):
+    """Avito Houses Catalog row: house_source='avito', house_ext_id='3171365' →
+    those are passed verbatim to match_or_create_house."""
+    db = _make_db_mock()
+    row = _row(
+        source="avito",
+        source_id="100",
+        house_source="avito",
+        house_ext_id="3171365",
+        house_url="https://www.avito.ru/catalog/houses/foo/3171365",
+    )
+
+    _link_listing_to_house(db, row, dry_run=False)
+
+    kwargs = _patch_matching["house"].call_args.kwargs
+    assert kwargs["ext_source"] == "avito"
+    assert kwargs["ext_id"] == "3171365"
+    assert kwargs["source_url"] == "https://www.avito.ru/catalog/houses/foo/3171365"
+
+
+def test_link_listing_falls_back_to_listing_source_when_house_missing(
+    _patch_matching,
+):
+    """No house_source on listing → ext_source = listing.source, ext_id =
+    source_id/dedup_hash. Same as PR I hook."""
+    db = _make_db_mock()
+    row = _row(source="cian", source_id="999", house_source=None, house_ext_id=None)
+
+    _link_listing_to_house(db, row, dry_run=False)
+
+    kwargs = _patch_matching["house"].call_args.kwargs
+    assert kwargs["ext_source"] == "cian"
+    assert kwargs["ext_id"] == "999"
+
+
+# ---------------------------------------------------------------------------
+# House match failure still upserts listing_sources
+# ---------------------------------------------------------------------------
+
+
+def test_link_listing_house_match_failure_still_upserts_source(_patch_matching):
+    """match_or_create_house raises → listing_sources row created with house_id=None."""
+    _patch_matching["house"].side_effect = RuntimeError("normalize boom")
+
+    db = _make_db_mock()
+    row = _row(source_id="bad", address="Bad addr")
+
+    house_id, resolved = _link_listing_to_house(db, row, dry_run=False)
+
+    assert house_id is None
+    assert resolved is False
+    assert _patch_matching["house"].call_count == 1
+    assert _patch_matching["link"].call_count == 1
+    kwargs = _patch_matching["link"].call_args.kwargs
+    assert kwargs["source_data"] is None
+
+
+# ---------------------------------------------------------------------------
+# Dry run — no writes anywhere
+# ---------------------------------------------------------------------------
+
+
+def test_link_listing_dry_run_skips_all_writes(_patch_matching):
+    """--dry-run path: no match_or_create_house call, no upsert_listing_source call."""
+    db = _make_db_mock()
+    row = _row(source_id="100")
+
+    _link_listing_to_house(db, row, dry_run=True)
+
+    assert _patch_matching["house"].call_count == 0
+    assert _patch_matching["link"].call_count == 0
+
+
+# ---------------------------------------------------------------------------
+# _process_row — SAVEPOINT wrapping, stats updates
+# ---------------------------------------------------------------------------
+
+
+def test_process_row_updates_stats_on_success(_patch_matching):
+    db = _make_db_mock()
+    row = _row(source="cian", source_id="abc")
+
+    stats = Stats()
+    _process_row(db, row, dry_run=False, stats=stats)
+
+    assert stats.processed == 1
+    assert stats.matched == 1
+    assert stats.house_resolved == 1
+    assert stats.errors == 0
+    assert stats.by_source["cian"]["matched"] == 1
+    assert stats.by_source["cian"]["house_resolved"] == 1
+
+
+def test_process_row_updates_stats_on_error(_patch_matching):
+    """upsert raises inside the SAVEPOINT → counted in errors, not matched."""
+    _patch_matching["link"].side_effect = RuntimeError("DB blip")
+
+    db = _make_db_mock()
+    row = _row(source="avito", source_id="abc")
+
+    stats = Stats()
+    _process_row(db, row, dry_run=False, stats=stats)
+
+    assert stats.processed == 1
+    assert stats.errors == 1
+    assert stats.matched == 0
+    assert stats.by_source["avito"]["errors"] == 1
+
+
+# ---------------------------------------------------------------------------
+# run_backfill — batch loop + idempotency + --limit
+# ---------------------------------------------------------------------------
+
+
+def test_run_backfill_processes_single_batch(_patch_matching):
+    """One batch of 2 rows → 2 link calls + commit."""
+    rows = [_row(id=1, source_id="a"), _row(id=2, source_id="b")]
+    db = _make_db_mock(batches=[[_row_dict(r) for r in rows], []])
+
+    stats = run_backfill(db, batch_size=500, limit=None, source=None, dry_run=False)
+
+    assert stats.processed == 2
+    assert _patch_matching["link"].call_count == 2
+    # Exactly one batch committed before the empty-batch stops the loop.
+    assert db.commit.call_count == 1
+
+
+def test_run_backfill_idempotent_no_remaining(_patch_matching):
+    """When SELECT returns 0 rows (re-run after full backfill) → 0 work done."""
+    db = _make_db_mock(batches=[[]])
+
+    stats = run_backfill(db, batch_size=500, limit=None, source=None, dry_run=False)
+
+    assert stats.processed == 0
+    assert _patch_matching["link"].call_count == 0
+    assert _patch_matching["house"].call_count == 0
+
+
+def test_run_backfill_dry_run_skips_writes(_patch_matching):
+    """--dry-run: counters move, but no upsert/match calls and no commit."""
+    rows = [_row(id=1, source_id="a")]
+    db = _make_db_mock(batches=[[_row_dict(r) for r in rows], []])
+
+    stats = run_backfill(db, batch_size=500, limit=None, source=None, dry_run=True)
+
+    assert stats.processed == 1
+    assert _patch_matching["link"].call_count == 0
+    assert _patch_matching["house"].call_count == 0
+    assert db.commit.call_count == 0
+
+
+def test_run_backfill_limit_caps_processed(_patch_matching):
+    """--limit 1 stops after the first row even when more would arrive."""
+    rows = [
+        _row(id=1, source_id="a"),
+        _row(id=2, source_id="b"),
+        _row(id=3, source_id="c"),
+    ]
+    db = _make_db_mock(batches=[[_row_dict(r) for r in rows], []])
+
+    stats = run_backfill(db, batch_size=500, limit=1, source=None, dry_run=False)
+
+    # Limit truncates the batch size, so only 1 row is fetched.
+    assert stats.processed == 1
+    assert _patch_matching["link"].call_count == 1
+
+
+def test_run_backfill_advances_after_id(_patch_matching):
+    """Cursor advances past max(id) in each batch so the next SELECT uses :after_id."""
+    rows_batch1 = [_row(id=1, source_id="a"), _row(id=5, source_id="b")]
+    rows_batch2 = [_row(id=12, source_id="c")]
+    db = _make_db_mock(
+        batches=[
+            [_row_dict(r) for r in rows_batch1],
+            [_row_dict(r) for r in rows_batch2],
+            [],
+        ]
+    )
+
+    stats = run_backfill(db, batch_size=2, limit=None, source=None, dry_run=False)
+
+    assert stats.processed == 3
+    # Third SELECT should have been called with after_id >= 12.
+    # Find the SELECT calls and look at the last one's params.
+    listings_calls = [c for c in db.execute.call_args_list if "FROM listings" in str(c.args[0])]
+    assert len(listings_calls) >= 3
+    last_params = listings_calls[-1].args[1]
+    assert last_params["after_id"] >= 12
+
+
+# ---------------------------------------------------------------------------
+# main() CLI smoke — happy path with one batch
+# ---------------------------------------------------------------------------
+
+
+def test_main_cli_runs_one_batch(_patch_matching):
+    """main(--limit 1 --dry-run) returns 1 row processed."""
+    rows = [_row(id=1, source_id="a")]
+
+    class FakeSessionLocal:
+        def __call__(self):
+            return _make_db_mock(batches=[[_row_dict(r) for r in rows], []])
+
+    with patch("scripts.backfill_listing_sources.SessionLocal", FakeSessionLocal()):
+        processed = main(["--limit", "1", "--dry-run"])
+
+    assert processed == 1