lekss361/gendesign
1
0
Fork 0
Code Issues 57 Pull requests Projects Releases Packages Wiki Activity Actions

feat(tradein): backfill 18k existing listings → listing_sources (PR J)

Browse source 
PR I (#595) hookнул matching service в save_listings для new scrapes.
PR J — repeatable script для existing 18,428 listings (avito 9302,
cian 5158, yandex 3704, n1 264) — все они unlinked к houses.

Script:
- backfill_listing_sources.py: stream batches of 500 rows, per-row
  SAVEPOINT, idempotent via NOT EXISTS pre-check + ON CONFLICT в
  upsert_listing_source.
- Use lot.house_source/house_ext_id if scraper extract'нул (direct
  link path), otherwise match_or_create_house() (Tier 0-3 via PostGIS
  geo-proximity + address fingerprint).
- upsert_listing_source с method='backfill', confidence=0.9 (lower
  чем real-time source_link=1.0 — для forensic distinction).
- Args: --batch-size N (default 500), --limit N (canary), --dry-run.
- No network calls (in-DB matching через PostGIS + address aliases) —
  работает даже когда Yandex Geocoder заблокирован.

Tests: 19 pass в test_backfill_listing_sources.py:
- happy path, idempotency (re-run skip), house_source direct linkage,
  match failure → still upserts с NULL house_id, dry-run, --limit cap,
  cursor advance, CLI entry point.

README.md обновлён с canonical docker exec instruction.

Run post-merge через:
  ssh gendesign 'docker exec tradein-backend python -m \\
    scripts.backfill_listing_sources --batch-size 500'

Expected ~5-15 min wall-time для 18k rows (per-row Tier 0-3 matching
через advisory lock + 3 PostGIS queries). Final coverage ~80-90% (~10%
remaining без resolvable address/coords).

Closes #582 (matching gap).
This commit is contained in:
Light1YT 2026-05-27 15:29:15 +05:00
parent 7e24ccbf2c
commit e83079f07a
4 changed files with 1168 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

77
tradein-mvp/backend/scripts/README.md
View file

@ -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%)
...
```

615
tradein-mvp/backend/scripts/backfill_listing_sources.py Normal file
View file

@ -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)

0
tradein-mvp/backend/tests/scripts/__init__.py Normal file
View file

476
tradein-mvp/backend/tests/scripts/test_backfill_listing_sources.py Normal file
View file

@ -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