613 lines
22 KiB
Python
613 lines
22 KiB
Python
"""Backfill listing_sources for ~18k existing listings — retroactive matching.
|
|
|
|
PR I (commit 7e24ccb) hooked the matching service into `save_listings()` so NEW
|
|
scraped listings now write a `listing_sources` row and resolve a canonical
|
|
`houses` row per scrape. Existing listings (~18,428: avito 9302, cian 5158,
|
|
yandex 3704, n1 264) were never matched — `listing_sources` only has rows
|
|
written since PR I shipped.
|
|
|
|
This script walks every `listings` row that does NOT already have a
|
|
`listing_sources` entry for its `(source, source_id)` (or `(source,
|
|
dedup_hash)` when `source_id` is NULL — same fallback as PR I) and performs
|
|
the same hook inline:
|
|
|
|
1. Resolve a canonical `houses` row via:
|
|
a. `house_source` / `house_ext_id` (Avito Houses Catalog, Cian
|
|
newbuilding) when the listing carries them — these are stable
|
|
source-side house identifiers that match `houses.(source,
|
|
ext_house_id)` directly.
|
|
b. Otherwise `match_or_create_house()` Tier 0-3 with the listing's
|
|
`address` / `lat` / `lon` / cadastrals.
|
|
2. `upsert_listing_source()` with method='backfill', confidence=0.9
|
|
(slightly below real-time `source_link` 1.0 so audits can distinguish).
|
|
3. UPDATE `listings.house_id_fk = matched.house_id` when the listing
|
|
didn't already have one — keeps the direct JOIN path consistent with
|
|
the matching graph.
|
|
|
|
Per-row SAVEPOINT (`db.begin_nested()`) per `.claude/rules/backend.md` so a
|
|
single bad row never aborts the whole batch.
|
|
|
|
Idempotent / resumable:
|
|
- Source query: `WHERE NOT EXISTS (SELECT 1 FROM listing_sources ls
|
|
WHERE ls.ext_source = ... AND ls.ext_id = ...)`. Re-runs skip already-
|
|
processed rows naturally.
|
|
- `upsert_listing_source` uses `ON CONFLICT (ext_source, ext_id) DO
|
|
UPDATE` — also safe.
|
|
|
|
No network calls — Yandex Geocoder is blocked on prod right now and the
|
|
matching service is purely in-database.
|
|
|
|
Usage:
|
|
DATABASE_URL=postgresql+psycopg://... \\
|
|
python -m scripts.backfill_listing_sources --batch-size 500
|
|
|
|
# Canary first
|
|
python -m scripts.backfill_listing_sources --limit 100 --dry-run
|
|
|
|
# Limit to a single source for staged rollout
|
|
python -m scripts.backfill_listing_sources --source avito --limit 1000
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import logging
|
|
import sys
|
|
from collections import defaultdict
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from sqlalchemy import text
|
|
from sqlalchemy.orm import Session
|
|
|
|
# Allow running both as `python -m scripts.backfill_listing_sources` (preferred)
|
|
# and as a stand-alone file (fallback for adhoc invocation).
|
|
try:
|
|
from app.core.db import SessionLocal # type: ignore[import-not-found]
|
|
from app.services.matching import ( # type: ignore[import-not-found]
|
|
match_or_create_house,
|
|
upsert_listing_source,
|
|
)
|
|
except ImportError: # pragma: no cover — fallback for adhoc invocation
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from app.core.db import SessionLocal
|
|
from app.services.matching import match_or_create_house, upsert_listing_source
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s %(levelname)s %(name)s %(message)s",
|
|
)
|
|
logger = logging.getLogger("backfill_listing_sources")
|
|
|
|
# Lower confidence than the real-time `source_link` (1.0) — backfilled rows
|
|
# weren't observed inline with the scrape, so audits can tell the two apart
|
|
# via `listing_sources.matched_method = 'backfill'` (the canonical filter)
|
|
# or via `confidence < 1.0` (the secondary check).
|
|
_BACKFILL_CONFIDENCE = 0.9
|
|
_BACKFILL_METHOD = "backfill"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Domain types
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@dataclass
|
|
class ListingRow:
|
|
"""One listing pulled by the source query — all fields we need for matching.
|
|
|
|
Mirrors the subset of ScrapedLot fields that match_or_create_house and
|
|
upsert_listing_source consume. Kept as a plain dataclass (no Pydantic) so
|
|
the script has zero parse overhead for 18k rows.
|
|
"""
|
|
|
|
id: int
|
|
source: str
|
|
source_id: str | None
|
|
source_url: str
|
|
dedup_hash: str
|
|
address: str | None
|
|
lat: float | None
|
|
lon: float | None
|
|
rooms: int | None
|
|
area_m2: float | None
|
|
floor: int | None
|
|
price_rub: int
|
|
year_built: int | None
|
|
cadastral_number: str | None
|
|
building_cadastral_number: str | None
|
|
house_source: str | None
|
|
house_ext_id: str | None
|
|
house_url: str | None
|
|
house_id_fk: int | None
|
|
|
|
|
|
@dataclass
|
|
class Stats:
|
|
"""Aggregate counters — printed per batch and at the end of the run."""
|
|
|
|
processed: int = 0
|
|
matched: int = 0 # listing_sources row upserted (with or without house)
|
|
house_resolved: int = 0
|
|
house_failed: int = 0
|
|
skipped: int = 0 # rare: no address/lat/lon/source_id at all
|
|
errors: int = 0
|
|
by_source: dict[str, dict[str, int]] = field(default_factory=dict)
|
|
|
|
def bump(self, source: str, key: str) -> None:
|
|
"""Increment a per-source counter (matched, house_resolved, errors...)."""
|
|
bucket = self.by_source.setdefault(source, defaultdict(int))
|
|
bucket[key] += 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Source query — stream listings without a listing_sources row
|
|
# ---------------------------------------------------------------------------
|
|
|
|
# All listings columns the matching hook consumes. Keep field list in sync
|
|
# with ListingRow.
|
|
_LISTING_COLUMNS = (
|
|
"id, source, source_id, source_url, dedup_hash, "
|
|
"address, lat, lon, rooms, area_m2, floor, price_rub, year_built, "
|
|
"cadastral_number, building_cadastral_number, "
|
|
"house_source, house_ext_id, house_url, house_id_fk"
|
|
)
|
|
|
|
|
|
def _build_select_sql(*, source: str | None, batch_size: int) -> str:
|
|
"""Streaming SELECT — skips rows already in listing_sources via NOT EXISTS.
|
|
|
|
The NOT EXISTS predicate matches the same (ext_source, ext_id) shape used
|
|
by `upsert_listing_source` so re-runs are zero-cost: any row PR I or a
|
|
previous backfill batch already inserted is excluded.
|
|
|
|
`ext_id` here mirrors the PR I hook's fallback chain: prefer
|
|
`listings.source_id`, fall back to `dedup_hash`. Yandex listings that
|
|
lack a stable `source_id` are still uniquely identifiable via
|
|
`dedup_hash` (sha256 of source + source_url + price).
|
|
"""
|
|
where_source = ""
|
|
if source is not None:
|
|
# Bind parameter — caller still passes :source. The string is just for
|
|
# SQL composition; psycopg fills :source from kwargs.
|
|
where_source = " AND source = :source "
|
|
sql = (
|
|
f"SELECT {_LISTING_COLUMNS} "
|
|
f"FROM listings "
|
|
f"WHERE id > :after_id "
|
|
f" {where_source} "
|
|
f" AND NOT EXISTS ("
|
|
f" SELECT 1 FROM listing_sources ls "
|
|
f" WHERE ls.ext_source = listings.source "
|
|
f" AND ls.ext_id = COALESCE(listings.source_id, listings.dedup_hash)"
|
|
f" ) "
|
|
f"ORDER BY id "
|
|
f"LIMIT CAST(:limit AS int)"
|
|
)
|
|
return sql
|
|
|
|
|
|
def _fetch_batch(
|
|
db: Session, *, after_id: int, batch_size: int, source: str | None
|
|
) -> list[ListingRow]:
|
|
"""Pull the next batch of unmatched listings ordered by id."""
|
|
sql = _build_select_sql(source=source, batch_size=batch_size)
|
|
params: dict[str, Any] = {"after_id": after_id, "limit": batch_size}
|
|
if source is not None:
|
|
params["source"] = source
|
|
|
|
rows = db.execute(text(sql), params).mappings().all()
|
|
return [
|
|
ListingRow(
|
|
id=r["id"],
|
|
source=r["source"],
|
|
source_id=r["source_id"],
|
|
source_url=r["source_url"],
|
|
dedup_hash=r["dedup_hash"],
|
|
address=r["address"],
|
|
lat=r["lat"],
|
|
lon=r["lon"],
|
|
rooms=r["rooms"],
|
|
area_m2=float(r["area_m2"]) if r["area_m2"] is not None else None,
|
|
floor=r["floor"],
|
|
price_rub=r["price_rub"],
|
|
year_built=r["year_built"],
|
|
cadastral_number=r["cadastral_number"],
|
|
building_cadastral_number=r["building_cadastral_number"],
|
|
house_source=r["house_source"],
|
|
house_ext_id=r["house_ext_id"],
|
|
house_url=r["house_url"],
|
|
house_id_fk=r["house_id_fk"],
|
|
)
|
|
for r in rows
|
|
]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Per-row work
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _ext_id_for(row: ListingRow) -> str:
|
|
"""Mirror PR I hook: prefer source_id, fall back to dedup_hash.
|
|
|
|
`dedup_hash` is already a sha256 hex string. Yandex listings without
|
|
stable source_id rely on this fallback — same hash on re-scrape, so
|
|
listing_sources stays unique per logical listing.
|
|
"""
|
|
if row.source_id:
|
|
return row.source_id
|
|
if row.dedup_hash:
|
|
return row.dedup_hash
|
|
# Defensive — should not happen since dedup_hash is NOT NULL UNIQUE in
|
|
# the schema, but compute one on the fly so we never write '' as ext_id.
|
|
h = hashlib.sha256(f"{row.source}|{row.source_url}|{row.price_rub}".encode()).hexdigest()
|
|
return h
|
|
|
|
|
|
def _link_listing_to_house(
|
|
db: Session, row: ListingRow, *, dry_run: bool
|
|
) -> tuple[int | None, bool]:
|
|
"""Resolve canonical house + upsert listing_sources for one listing.
|
|
|
|
Returns:
|
|
(house_id_or_None, house_resolved_bool).
|
|
`house_resolved=True` means match_or_create_house produced a non-null
|
|
id (either direct via house_source/ext_id or via Tier 0-3 fallback).
|
|
"""
|
|
ext_id = _ext_id_for(row)
|
|
|
|
# House resolution — only attempted if we have address or coords. Without
|
|
# them, match_or_create_house Tier 2 (fingerprint) trivially mismatches and
|
|
# Tier 3 (geo-proximity) is impossible — we'd be creating an island house
|
|
# row keyed off a synthetic ext_id with no real data.
|
|
house_id: int | None = None
|
|
house_resolved = False
|
|
if row.address or (row.lat is not None and row.lon is not None):
|
|
# Prefer the scraped house_source/house_ext_id (Avito Houses Catalog,
|
|
# Cian newbuilding) — those map 1:1 to `houses.(source, ext_house_id)`
|
|
# via Tier 1 (`source_exact`) and avoid fingerprint/geo lookups.
|
|
h_src = row.house_source or row.source
|
|
h_ext = row.house_ext_id or ext_id
|
|
if dry_run:
|
|
# Don't touch the DB. Pretend a house was resolved if scraped
|
|
# extras are present — used by the dry-run summary only.
|
|
house_id = row.house_id_fk
|
|
house_resolved = (
|
|
row.house_id_fk is not None
|
|
or row.house_source is not None
|
|
or (row.address is not None)
|
|
)
|
|
else:
|
|
try:
|
|
with db.begin_nested():
|
|
house_id, _conf, _method = match_or_create_house(
|
|
db,
|
|
ext_source=h_src,
|
|
ext_id=h_ext,
|
|
address=row.address,
|
|
lat=row.lat,
|
|
lon=row.lon,
|
|
year_built=row.year_built,
|
|
building_cadastral_number=row.building_cadastral_number,
|
|
cadastral_number=row.cadastral_number,
|
|
source_url=row.house_url or row.source_url,
|
|
)
|
|
house_resolved = house_id is not None
|
|
except Exception as e:
|
|
# Fall through to listing-only upsert — the listing_sources row
|
|
# is still useful (e.g. for a later backfill that can geocode).
|
|
logger.warning(
|
|
"backfill:house_match_failed listing_id=%s source=%s ext_id=%s: %s",
|
|
row.id,
|
|
row.source,
|
|
ext_id,
|
|
e,
|
|
)
|
|
house_id = None
|
|
house_resolved = False
|
|
|
|
if not dry_run:
|
|
upsert_listing_source(
|
|
db,
|
|
listing_id=row.id,
|
|
ext_source=row.source,
|
|
ext_id=str(ext_id),
|
|
method=_BACKFILL_METHOD,
|
|
confidence=_BACKFILL_CONFIDENCE,
|
|
price_rub=row.price_rub,
|
|
area_m2=row.area_m2,
|
|
floor=row.floor,
|
|
rooms_count=row.rooms,
|
|
source_url=row.source_url,
|
|
source_data={"house_id": house_id} if house_id else None,
|
|
)
|
|
|
|
# Mirror the matching graph into listings.house_id_fk so direct
|
|
# `JOIN houses ON house_id_fk = id` keeps working. Only updates when
|
|
# the row didn't already have a linkage (preserves any prior 063
|
|
# backfill).
|
|
if house_id is not None and row.house_id_fk is None:
|
|
db.execute(
|
|
text(
|
|
"UPDATE listings "
|
|
" SET house_id_fk = CAST(:hid AS bigint) "
|
|
" WHERE id = CAST(:lid AS bigint) "
|
|
" AND house_id_fk IS NULL"
|
|
),
|
|
{"hid": house_id, "lid": row.id},
|
|
)
|
|
|
|
return house_id, house_resolved
|
|
|
|
|
|
def _process_row(db: Session, row: ListingRow, *, dry_run: bool, stats: Stats) -> None:
|
|
"""Wrap _link_listing_to_house in a per-row SAVEPOINT.
|
|
|
|
Per backend.md `Bug_Pzz_Loader_Missing_Savepoint_May14`: never bare
|
|
`db.rollback()` inside a row loop — it kills the surrounding tx and the
|
|
counters get out of sync with what actually committed.
|
|
"""
|
|
stats.processed += 1
|
|
stats.bump(row.source, "processed")
|
|
|
|
# Skip listings that lack any signal we can use — without source_id we'd
|
|
# have to rely on dedup_hash + no address → effectively orphan rows in
|
|
# listing_sources. Still upsert via dedup_hash but mark as skipped from
|
|
# house-resolution counts.
|
|
if not row.source_id and not row.address and (row.lat is None or row.lon is None):
|
|
stats.skipped += 1
|
|
stats.bump(row.source, "skipped")
|
|
|
|
try:
|
|
if dry_run:
|
|
# No SAVEPOINT in dry-run — we don't touch the DB at all.
|
|
house_id, house_resolved = _link_listing_to_house(db, row, dry_run=True)
|
|
else:
|
|
with db.begin_nested():
|
|
house_id, house_resolved = _link_listing_to_house(db, row, dry_run=False)
|
|
|
|
stats.matched += 1
|
|
stats.bump(row.source, "matched")
|
|
if house_resolved:
|
|
stats.house_resolved += 1
|
|
stats.bump(row.source, "house_resolved")
|
|
else:
|
|
stats.house_failed += 1
|
|
stats.bump(row.source, "house_failed")
|
|
|
|
logger.debug(
|
|
"backfill ok listing_id=%s source=%s ext_id=%s house_id=%s",
|
|
row.id,
|
|
row.source,
|
|
_ext_id_for(row),
|
|
house_id,
|
|
)
|
|
except Exception as e:
|
|
stats.errors += 1
|
|
stats.bump(row.source, "errors")
|
|
logger.warning(
|
|
"backfill failed listing_id=%s source=%s: %s",
|
|
row.id,
|
|
row.source,
|
|
e,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Driver
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _coverage_summary(db: Session) -> dict[str, Any]:
|
|
"""Per-source coverage % after the run — useful for the final log line."""
|
|
rows = (
|
|
db.execute(
|
|
text(
|
|
"SELECT "
|
|
" l.source, "
|
|
" COUNT(*) AS total, "
|
|
" COUNT(*) FILTER ("
|
|
" WHERE EXISTS ("
|
|
" SELECT 1 FROM listing_sources ls "
|
|
" WHERE ls.ext_source = l.source "
|
|
" AND ls.ext_id = COALESCE(l.source_id, l.dedup_hash)"
|
|
" )"
|
|
" ) AS linked "
|
|
"FROM listings l "
|
|
"GROUP BY l.source "
|
|
"ORDER BY total DESC"
|
|
)
|
|
)
|
|
.mappings()
|
|
.all()
|
|
)
|
|
return {
|
|
r["source"]: {
|
|
"total": int(r["total"]),
|
|
"linked": int(r["linked"]),
|
|
"pct": (float(r["linked"]) / float(r["total"]) * 100.0) if r["total"] else 0.0,
|
|
}
|
|
for r in rows
|
|
}
|
|
|
|
|
|
def run_backfill(
|
|
db: Session,
|
|
*,
|
|
batch_size: int,
|
|
limit: int | None,
|
|
source: str | None,
|
|
dry_run: bool,
|
|
) -> Stats:
|
|
"""Main driver — streams batches and writes listing_sources rows.
|
|
|
|
Args:
|
|
db: SQLAlchemy Session.
|
|
batch_size: how many rows to fetch per SELECT. 500 keeps memory flat
|
|
and gives a checkpoint every commit.
|
|
limit: stop after processing this many total rows (--limit). None =
|
|
run to exhaustion.
|
|
source: filter to one ext_source ('avito'/'cian'/'yandex').
|
|
None = all sources (historical 'n1' rows still processed in a full run).
|
|
dry_run: skip all writes, just count.
|
|
"""
|
|
stats = Stats()
|
|
after_id = 0
|
|
batch_idx = 0
|
|
|
|
while True:
|
|
# Stop early if --limit reached.
|
|
if limit is not None and stats.processed >= limit:
|
|
logger.info(
|
|
"limit reached: stopping (processed=%d, limit=%d)",
|
|
stats.processed,
|
|
limit,
|
|
)
|
|
break
|
|
|
|
effective_size = batch_size
|
|
if limit is not None:
|
|
effective_size = min(batch_size, limit - stats.processed)
|
|
if effective_size <= 0:
|
|
break
|
|
|
|
batch = _fetch_batch(db, after_id=after_id, batch_size=effective_size, source=source)
|
|
if not batch:
|
|
logger.info("no more rows — done")
|
|
break
|
|
|
|
batch_idx += 1
|
|
for row in batch:
|
|
_process_row(db, row, dry_run=dry_run, stats=stats)
|
|
after_id = max(after_id, row.id)
|
|
|
|
# Commit the batch — every 500 rows under default config. Dry-run
|
|
# never wrote anything so the commit is a no-op (cheap).
|
|
if not dry_run:
|
|
db.commit()
|
|
|
|
logger.info(
|
|
"batch %d done: size=%d total processed=%d matched=%d "
|
|
"house_resolved=%d house_failed=%d errors=%d",
|
|
batch_idx,
|
|
len(batch),
|
|
stats.processed,
|
|
stats.matched,
|
|
stats.house_resolved,
|
|
stats.house_failed,
|
|
stats.errors,
|
|
)
|
|
|
|
return stats
|
|
|
|
|
|
def _log_summary(stats: Stats, db: Session, *, dry_run: bool) -> None:
|
|
"""Final per-source breakdown + overall coverage."""
|
|
logger.info("=" * 72)
|
|
logger.info(
|
|
"backfill done (dry_run=%s): processed=%d matched=%d "
|
|
"house_resolved=%d house_failed=%d skipped=%d errors=%d",
|
|
dry_run,
|
|
stats.processed,
|
|
stats.matched,
|
|
stats.house_resolved,
|
|
stats.house_failed,
|
|
stats.skipped,
|
|
stats.errors,
|
|
)
|
|
for src, bucket in sorted(stats.by_source.items()):
|
|
logger.info(
|
|
" %-10s processed=%d matched=%d house_resolved=%d house_failed=%d errors=%d",
|
|
src,
|
|
bucket.get("processed", 0),
|
|
bucket.get("matched", 0),
|
|
bucket.get("house_resolved", 0),
|
|
bucket.get("house_failed", 0),
|
|
bucket.get("errors", 0),
|
|
)
|
|
|
|
# Coverage reflects the on-disk state. In dry-run mode this still shows
|
|
# pre-existing rows from PR I — useful to gauge what the next real run
|
|
# would do.
|
|
coverage = _coverage_summary(db)
|
|
logger.info("final listing_sources coverage:")
|
|
for src, c in sorted(coverage.items()):
|
|
logger.info(" %-10s %d / %d (%.1f%%)", src, c["linked"], c["total"], c["pct"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CLI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
"""argparse setup, factored out for testability."""
|
|
p = argparse.ArgumentParser(
|
|
description=(
|
|
"PR J — backfill listing_sources for existing listings. Resolves "
|
|
"canonical houses + upserts listing_sources rows for every "
|
|
"listing that does not already have one. Idempotent on re-run."
|
|
),
|
|
)
|
|
p.add_argument(
|
|
"--batch-size",
|
|
type=int,
|
|
default=500,
|
|
help="Rows pulled per SELECT (default: 500). Each batch commits once.",
|
|
)
|
|
p.add_argument(
|
|
"--limit",
|
|
type=int,
|
|
default=None,
|
|
help=(
|
|
"Cap on total rows processed across all batches. Useful for "
|
|
"canary runs (e.g. --limit 100 before letting the full job loose)."
|
|
),
|
|
)
|
|
p.add_argument(
|
|
"--source",
|
|
choices=("avito", "cian", "yandex"),
|
|
default=None,
|
|
help=(
|
|
"Restrict to one ext_source for staged rollout. Default: all sources mixed in id order."
|
|
),
|
|
)
|
|
p.add_argument(
|
|
"--dry-run",
|
|
action="store_true",
|
|
help="Log what would be done; no DB writes.",
|
|
)
|
|
return p.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
"""CLI entry point. Returns the number of rows processed this run."""
|
|
args = _parse_args(argv)
|
|
logger.info(
|
|
"starting backfill: batch_size=%d limit=%s source=%s dry_run=%s",
|
|
args.batch_size,
|
|
args.limit if args.limit is not None else "all",
|
|
args.source or "all",
|
|
args.dry_run,
|
|
)
|
|
|
|
db = SessionLocal()
|
|
try:
|
|
stats = run_backfill(
|
|
db,
|
|
batch_size=args.batch_size,
|
|
limit=args.limit,
|
|
source=args.source,
|
|
dry_run=args.dry_run,
|
|
)
|
|
_log_summary(stats, db, dry_run=args.dry_run)
|
|
return stats.processed
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
if __name__ == "__main__": # pragma: no cover
|
|
sys.exit(0 if main() >= 0 else 1)
|