From 1d8cfb967f0b06d26b91c705ea002a466c7d41ff Mon Sep 17 00:00:00 2001 From: Light1YT Date: Wed, 27 May 2026 10:29:40 +0000 Subject: [PATCH] feat(tradein): backfill 18k existing listings (PR J) (#596) --- tradein-mvp/backend/scripts/README.md | 77 +++ .../scripts/backfill_listing_sources.py | 615 ++++++++++++++++++ tradein-mvp/backend/tests/scripts/__init__.py | 0 .../scripts/test_backfill_listing_sources.py | 476 ++++++++++++++ 4 files changed, 1168 insertions(+) create mode 100644 tradein-mvp/backend/scripts/backfill_listing_sources.py create mode 100644 tradein-mvp/backend/tests/scripts/__init__.py create mode 100644 tradein-mvp/backend/tests/scripts/test_backfill_listing_sources.py diff --git a/tradein-mvp/backend/scripts/README.md b/tradein-mvp/backend/scripts/README.md index 6eb3407e..24ef2a24 100644 --- a/tradein-mvp/backend/scripts/README.md +++ b/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%) + ... +``` diff --git a/tradein-mvp/backend/scripts/backfill_listing_sources.py b/tradein-mvp/backend/scripts/backfill_listing_sources.py new file mode 100644 index 00000000..d757d35d --- /dev/null +++ b/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) diff --git a/tradein-mvp/backend/tests/scripts/__init__.py b/tradein-mvp/backend/tests/scripts/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tradein-mvp/backend/tests/scripts/test_backfill_listing_sources.py b/tradein-mvp/backend/tests/scripts/test_backfill_listing_sources.py new file mode 100644 index 00000000..544af13c --- /dev/null +++ b/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