feat(tradein): Phase 1.3 backfill scripts -- fingerprints + house/listing_sources
Multi-Source Integration Phase 1.3. Manual operator scripts for legacy data
migration. NOT auto-applied -- run explicitly via uv/psql.
scripts/:
- backfill_001_address_fingerprints.py -- Python, populates houses.address_fingerprint
for rows where NULL (uses matching/normalize.address_fingerprint from PR #470)
- backfill_002_house_sources.sql -- INSERT INTO house_sources FROM houses,
ON CONFLICT (ext_source, ext_id) DO NOTHING (idempotent)
- backfill_003_listing_sources.sql -- INSERT INTO listing_sources FROM listings,
same pattern
- README.md -- order + usage
All idempotent -- re-running safe. confidence=1.0, matched_method='backfill'.
This commit is contained in:
parent
b26e0ff7c3
commit
b32628e262
4 changed files with 207 additions and 0 deletions
27
tradein-mvp/scripts/README.md
Normal file
27
tradein-mvp/scripts/README.md
Normal file
|
|
@ -0,0 +1,27 @@
|
|||
# tradein-mvp scripts -- backfill
|
||||
|
||||
One-shot manual operator scripts. **NOT auto-applied by deploy.yml** (which only runs `tradein-mvp/backend/data/sql/NN_*.sql`).
|
||||
|
||||
## Order
|
||||
|
||||
Run in numbered sequence after PR #470 (matching module) merged:
|
||||
|
||||
```bash
|
||||
cd tradein-mvp/backend
|
||||
export DATABASE_URL="postgresql+psycopg://user:pass@host:5432/tradein"
|
||||
|
||||
# 001 -- backfill houses.address_fingerprint (Python, idempotent)
|
||||
uv run python ../scripts/backfill_001_address_fingerprints.py
|
||||
|
||||
# 002 -- backfill house_sources link rows (SQL, idempotent)
|
||||
psql "${DATABASE_URL/postgresql+psycopg/postgresql}" -f ../scripts/backfill_002_house_sources.sql
|
||||
|
||||
# 003 -- backfill listing_sources link rows (SQL, idempotent)
|
||||
psql "${DATABASE_URL/postgresql+psycopg/postgresql}" -f ../scripts/backfill_003_listing_sources.sql
|
||||
```
|
||||
|
||||
All scripts idempotent -- re-running is safe (ON CONFLICT DO NOTHING on the SQL; row-level UPDATE WHERE NULL on the Python).
|
||||
|
||||
## Other scripts
|
||||
|
||||
- `upload-cian-cookies.sh` -- one-shot Cian session cookie upload (see inline docs)
|
||||
86
tradein-mvp/scripts/backfill_001_address_fingerprints.py
Normal file
86
tradein-mvp/scripts/backfill_001_address_fingerprints.py
Normal file
|
|
@ -0,0 +1,86 @@
|
|||
"""Backfill houses.address_fingerprint for legacy rows.
|
||||
|
||||
Usage:
|
||||
cd tradein-mvp/backend
|
||||
DATABASE_URL=postgresql://... uv run python ../scripts/backfill_001_address_fingerprints.py
|
||||
|
||||
Idempotent -- safe to re-run. Updates only rows with NULL address_fingerprint.
|
||||
Logs progress every 100 rows.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os
|
||||
import sys
|
||||
|
||||
import psycopg
|
||||
|
||||
# Allow running from repo root via `python tradein-mvp/scripts/...`
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "backend"))
|
||||
|
||||
from app.services.matching.normalize import (
|
||||
address_fingerprint,
|
||||
normalize_address,
|
||||
)
|
||||
|
||||
logging.basicConfig(
|
||||
format="%(asctime)s [%(levelname)s] %(message)s",
|
||||
level=logging.INFO,
|
||||
)
|
||||
logger = logging.getLogger("backfill_001")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
dsn = os.environ.get("DATABASE_URL")
|
||||
if not dsn:
|
||||
logger.error("DATABASE_URL env var required")
|
||||
return 1
|
||||
|
||||
# Strip SQLAlchemy dialect prefix if present (tradein-mvp uses postgresql+psycopg://)
|
||||
dsn = dsn.replace("postgresql+psycopg://", "postgresql://", 1)
|
||||
|
||||
updated = 0
|
||||
skipped = 0
|
||||
with psycopg.connect(dsn) as conn:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
"SELECT id, address, lat, lon FROM houses "
|
||||
"WHERE address_fingerprint IS NULL AND address IS NOT NULL"
|
||||
)
|
||||
rows = cur.fetchall()
|
||||
logger.info("Found %d houses missing fingerprint", len(rows))
|
||||
|
||||
for i, (house_id, address, lat, lon) in enumerate(rows, 1):
|
||||
normalized = normalize_address(address)
|
||||
if normalized is None:
|
||||
skipped += 1
|
||||
continue
|
||||
fp = address_fingerprint(normalized, lat, lon)
|
||||
if fp is None:
|
||||
skipped += 1
|
||||
continue
|
||||
cur.execute(
|
||||
"UPDATE houses SET address_fingerprint = %s WHERE id = %s",
|
||||
(fp, house_id),
|
||||
)
|
||||
updated += 1
|
||||
if i % 100 == 0:
|
||||
conn.commit()
|
||||
logger.info(
|
||||
"Progress: %d/%d updated=%d skipped=%d",
|
||||
i,
|
||||
len(rows),
|
||||
updated,
|
||||
skipped,
|
||||
)
|
||||
|
||||
conn.commit()
|
||||
|
||||
logger.info(
|
||||
"Done: updated=%d skipped=%d total=%d", updated, skipped, updated + skipped
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
42
tradein-mvp/scripts/backfill_002_house_sources.sql
Normal file
42
tradein-mvp/scripts/backfill_002_house_sources.sql
Normal file
|
|
@ -0,0 +1,42 @@
|
|||
-- backfill_002_house_sources.sql
|
||||
-- Purpose: Populate house_sources from legacy houses.(source, ext_house_id) pairs.
|
||||
-- Idempotent -- ON CONFLICT DO NOTHING.
|
||||
-- Run AFTER backfill_001 (fingerprint backfill is independent but logically prior).
|
||||
--
|
||||
-- Column notes (verified against 009_houses.sql + 029_extend_matching_valuation_dynamics.sql):
|
||||
-- houses.last_scraped_at -- timestamptz, NOT NULL DEFAULT NOW() (009_houses.sql)
|
||||
-- house_sources.last_seen_at -- added in 029 (nullable timestamptz)
|
||||
-- house_sources columns: id, house_id, ext_source, ext_id, confidence, matched_method,
|
||||
-- matched_at, ext_url, raw_payload, last_seen_at
|
||||
--
|
||||
-- Usage:
|
||||
-- psql "$DATABASE_URL" -f tradein-mvp/scripts/backfill_002_house_sources.sql
|
||||
|
||||
BEGIN;
|
||||
|
||||
INSERT INTO house_sources (
|
||||
house_id,
|
||||
ext_source,
|
||||
ext_id,
|
||||
confidence,
|
||||
matched_method,
|
||||
matched_at,
|
||||
last_seen_at
|
||||
)
|
||||
SELECT
|
||||
h.id,
|
||||
h.source,
|
||||
h.ext_house_id,
|
||||
1.0 AS confidence,
|
||||
'backfill' AS matched_method,
|
||||
NOW() AS matched_at,
|
||||
h.last_scraped_at AS last_seen_at
|
||||
FROM houses h
|
||||
WHERE h.ext_house_id IS NOT NULL
|
||||
ON CONFLICT (ext_source, ext_id) DO NOTHING;
|
||||
|
||||
\echo 'house_sources backfill complete'
|
||||
SELECT count(*) AS total_house_sources FROM house_sources;
|
||||
SELECT ext_source, count(*) AS rows FROM house_sources GROUP BY ext_source ORDER BY ext_source;
|
||||
|
||||
COMMIT;
|
||||
52
tradein-mvp/scripts/backfill_003_listing_sources.sql
Normal file
52
tradein-mvp/scripts/backfill_003_listing_sources.sql
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
-- backfill_003_listing_sources.sql
|
||||
-- Purpose: Populate listing_sources from legacy listings.(source, source_id) pairs.
|
||||
-- Idempotent -- ON CONFLICT DO NOTHING.
|
||||
--
|
||||
-- Column notes (verified against 002_core_tables.sql + 011_listings_alter.sql
|
||||
-- + 029_extend_matching_valuation_dynamics.sql):
|
||||
-- listings.area_m2 -- numeric(8,2), defined in 002_core_tables.sql (NOT total_area)
|
||||
-- listings.price_rub -- bigint NOT NULL (002_core_tables.sql)
|
||||
-- listings.floor -- int (002_core_tables.sql)
|
||||
-- listings.rooms -- int (002_core_tables.sql, NOT rooms_count)
|
||||
-- listings.scraped_at -- timestamptz NOT NULL DEFAULT NOW() (002_core_tables.sql)
|
||||
-- listing_sources columns (029): price_rub, area_m2, floor, rooms_count, raw_payload, last_seen_at
|
||||
--
|
||||
-- Usage:
|
||||
-- psql "$DATABASE_URL" -f tradein-mvp/scripts/backfill_003_listing_sources.sql
|
||||
|
||||
BEGIN;
|
||||
|
||||
INSERT INTO listing_sources (
|
||||
listing_id,
|
||||
ext_source,
|
||||
ext_id,
|
||||
confidence,
|
||||
matched_method,
|
||||
matched_at,
|
||||
last_scraped_at,
|
||||
price_rub,
|
||||
area_m2,
|
||||
floor,
|
||||
rooms_count
|
||||
)
|
||||
SELECT
|
||||
l.id,
|
||||
l.source,
|
||||
l.source_id,
|
||||
1.0 AS confidence,
|
||||
'backfill' AS matched_method,
|
||||
NOW() AS matched_at,
|
||||
l.scraped_at AS last_scraped_at,
|
||||
l.price_rub AS price_rub,
|
||||
l.area_m2 AS area_m2,
|
||||
l.floor AS floor,
|
||||
l.rooms AS rooms_count
|
||||
FROM listings l
|
||||
WHERE l.source_id IS NOT NULL
|
||||
ON CONFLICT (ext_source, ext_id) DO NOTHING;
|
||||
|
||||
\echo 'listing_sources backfill complete'
|
||||
SELECT count(*) AS total_listing_sources FROM listing_sources;
|
||||
SELECT ext_source, count(*) AS rows FROM listing_sources GROUP BY ext_source ORDER BY ext_source;
|
||||
|
||||
COMMIT;
|
||||
Loading…
Add table
Reference in a new issue