From edd187b47fe9acad8d44c851c333d15980828ffb Mon Sep 17 00:00:00 2001 From: bot-backend Date: Thu, 18 Jun 2026 10:18:54 +0300 Subject: [PATCH] feat(cadastral): geo-nearest building matcher via local cad mirror MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Backfill listings.building_cadastral_number (was 0%) from the nearest cadastral building. gendesign_cad_buildings is a postgres_fdw foreign table with no geom — a per-listing FDW nearest query is ~1.16s/row (~13h for 43k listings). Instead materialize the FDW once into a LOCAL cad_buildings_local table (Point geom + GIST), then run a fast local KNN nearest-neighbour join. Perf: the distance gate uses geometry-space ST_DWithin(geom, point, deg) (GIST index, no geography cast) + geom <-> KNN order + ST_DistanceSphere metric recheck on the single nearest row. A geography-cast ST_DWithin in the WHERE defeated the index (58s+, full update never finished); the geometry-gate design does the whole ~41.9k-listing UPDATE in ~5.4s (refresh+match ~7s end-to-end), matching 16027 listings at 50m across 2229 distinct buildings. The match is GEO-NEAREST (approximate): a street-level-geocoded listing matches the nearest building within threshold_m (default 50m), not necessarily its exact cadastral building. Exact cadastral + parcel-containment deferred (cad_parcels FDW not exposed). Threshold is logged. - 124: cad_buildings_local table (empty) + GIST. Migration does not read the FDW (deploy-independent); populated by the refresh job. - 125: scrape_schedules seed source=cadastral_geo_match, enabled, next_run_at tomorrow 09:00 UTC (after geocode_missing). - tasks/cadastral_geo_match.py: refresh_cad_buildings_local (bulk FDW scan), match_listings_to_buildings (chunked LATERAL KNN UPDATE), run_cadastral_geo_match run-lifecycle wrapper. - scheduler: trigger_cadastral_geo_match_run + dispatch (sync DB-only, executor). --- tradein-mvp/backend/app/services/scheduler.py | 54 +++ .../backend/app/tasks/cadastral_geo_match.py | 349 ++++++++++++++++ .../data/sql/124_cad_buildings_local.sql | 47 +++ ...ape_schedules_seed_cadastral_geo_match.sql | 58 +++ .../tests/tasks/test_cadastral_geo_match.py | 386 ++++++++++++++++++ 5 files changed, 894 insertions(+) create mode 100644 tradein-mvp/backend/app/tasks/cadastral_geo_match.py create mode 100644 tradein-mvp/backend/data/sql/124_cad_buildings_local.sql create mode 100644 tradein-mvp/backend/data/sql/125_scrape_schedules_seed_cadastral_geo_match.sql create mode 100644 tradein-mvp/backend/tests/tasks/test_cadastral_geo_match.py diff --git a/tradein-mvp/backend/app/services/scheduler.py b/tradein-mvp/backend/app/services/scheduler.py index 831623a7..7a265502 100644 --- a/tradein-mvp/backend/app/services/scheduler.py +++ b/tradein-mvp/backend/app/services/scheduler.py @@ -67,6 +67,16 @@ Sources: city_sweep; fetch via curl_cffi chrome120 + scraper_proxy_url, parse via YandexDetailScraper.parse; window 12:00-15:00 UTC — после avito_detail_backfill 09-12 UTC) + - cadastral_geo_match → run_cadastral_geo_match + (tasks/cadastral_geo_match.py; combined nightly job — (1) REFRESH + cad_buildings_local from gendesign_cad_buildings FDW (one bulk + scan, ~36.7k EKB rows, Point geom + GIST), (2) MATCH set-based + LATERAL KNN UPDATE filling listings.building_cadastral_number from + the nearest cadastral building within threshold_m (default 50m). + GEO-NEAREST approximation (nearest building, not exact cadastral). + Pure internal DB op — one FDW read + local UPDATE, no HTTP/anti-bot; + window 09:00-10:00 UTC — после geocode_missing_listings 06-09 UTC + чтобы listings имели свежие lat/lon/geom) """ from __future__ import annotations @@ -1313,6 +1323,48 @@ async def trigger_yandex_detail_backfill_run( return run_id +async def trigger_cadastral_geo_match_run(db: Session, schedule_row: dict[str, Any]) -> int | None: + """Создать scrape_runs + launch run_cadastral_geo_match в executor (sync DB-only task). + + Combined refresh+match (#cadastral-geo-match): (1) REFRESH cad_buildings_local from the + gendesign_cad_buildings FDW (one bulk scan, builds Point geom), (2) MATCH set-based + LATERAL KNN UPDATE filling listings.building_cadastral_number from the nearest cadastral + building within threshold_m (default 50m). GEO-NEAREST approximation. + + Sync task (one FDW read + local UPDATE, no async HTTP) — run in run_in_executor by the + same pattern as trigger_listing_source_snapshot_run / trigger_asking_to_sold_ratio_run. + run_cadastral_geo_match owns the scrape_runs lifecycle (mark_done/mark_failed). SAFE to + enable — seed 125 enabled=true, pure internal DB op. + + Returns run_id или None (skip — already running). + """ + run_id = _claim_run(db, schedule_row) + if run_id is None: + return None + + params = schedule_row.get("default_params") or {} + + async def _run() -> None: + run_db = SessionLocal() + try: + from app.tasks.cadastral_geo_match import run_cadastral_geo_match + + loop = asyncio.get_event_loop() + await loop.run_in_executor( + None, + lambda: run_cadastral_geo_match(run_db, run_id=run_id, params=params), + ) + except Exception: + logger.exception("scheduler: run_cadastral_geo_match crashed run_id=%d", run_id) + finally: + run_db.close() + + task = asyncio.create_task(_run()) + task.add_done_callback(lambda t: t.exception() if not t.cancelled() else None) + logger.info("scheduler: triggered cadastral_geo_match run_id=%d", run_id) + return run_id + + def get_due_schedules(db: Session) -> list[dict[str, Any]]: """SELECT scrape_schedules WHERE enabled AND (next_run_at IS NULL OR next_run_at <= NOW()).""" rows = ( @@ -1384,6 +1436,8 @@ async def scheduler_loop() -> None: await trigger_avito_detail_backfill_run(db, sch) elif source == "yandex_detail_backfill": await trigger_yandex_detail_backfill_run(db, sch) + elif source == "cadastral_geo_match": + await trigger_cadastral_geo_match_run(db, sch) else: logger.warning("scheduler: unknown source=%s, skip", source) finally: diff --git a/tradein-mvp/backend/app/tasks/cadastral_geo_match.py b/tradein-mvp/backend/app/tasks/cadastral_geo_match.py new file mode 100644 index 00000000..aa61eb0d --- /dev/null +++ b/tradein-mvp/backend/app/tasks/cadastral_geo_match.py @@ -0,0 +1,349 @@ +"""Geo-nearest cadastral building matcher for listings (#cadastral-geo-match). + +Fills `listings.building_cadastral_number` (currently 0% / ~43k rows) from the nearest +cadastral building, via a LOCAL materialized mirror of the gendesign_cad_buildings FDW. + +WHY a local mirror (perf fact, measured live): + gendesign_cad_buildings is a postgres_fdw foreign table (~36.7k EKB rows) with NO + PostGIS geom — only scalar lat/lon. A per-listing nearest-building query over the FDW + takes ~1.16s/row → ~13h for 43k listings. UNUSABLE per-row. + ⇒ We materialize the FDW once (single bulk scan) into cad_buildings_local with a real + Point geom + GIST, then run a fast LOCAL KNN nearest-neighbour join (geom <-> point). + The whole local backfill runs in well under a minute (see PR EXPLAIN evidence). + +APPROXIMATION (deliberate first increment): + This is a GEO-NEAREST match — a street-level-geocoded listing is matched to the nearest + cadastral building within `threshold_m`, NOT necessarily its exact cadastral building. + The threshold is always logged. Exact cadastral resolution + parcel-containment are + deferred (cad_parcels FDW not exposed). Tier-0 house matching in the estimator already + treats building_cadastral_number as a hint, not ground truth, so an approximate fill is + a net win over 0% coverage. + +Pipeline (one combined run, scheduler source='cadastral_geo_match'): + 1. refresh_cad_buildings_local(db) — TRUNCATE + bulk INSERT from FDW (one scan). + 2. match_listings_to_buildings(db, threshold_m) — set-based LATERAL KNN UPDATE. + Refresh-then-match in ONE run guarantees the match always reads a fresh mirror. + +psycopg v3: app SQL uses CAST(:x AS type), never :x::type. +""" + +from __future__ import annotations + +import logging +import math +import time +from dataclasses import dataclass, field + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.services import scrape_runs as runs_mod + +logger = logging.getLogger(__name__) + +DEFAULT_THRESHOLD_M = 50 +DEFAULT_BATCH_SIZE = 5000 + +# EKB latitude — used to convert the metric threshold into a degree gate radius for the +# GIST-indexable geometry-space pre-filter (the gate is a SUPERSET of the true circle; the +# ST_DistanceSphere recheck decides acceptance, so a slightly-too-wide gate is harmless). +_EKB_LAT_DEG = 56.84 +_M_PER_DEG_LAT = 111_320.0 + + +def _deg_gate_for(threshold_m: float) -> float: + """Degree radius that always encloses `threshold_m` metres at EKB latitude. + + Longitude degrees shrink by cos(lat), so we size the gate on the tighter (longitude) + axis and apply a 1.5× safety factor. The gate only pre-filters candidates for the GIST + KNN scan; the exact ST_DistanceSphere(...) <= threshold_m recheck is authoritative, so + over-sizing the gate never produces a false match, only a few extra candidates to rank. + """ + m_per_deg_lon = _M_PER_DEG_LAT * math.cos(math.radians(_EKB_LAT_DEG)) + return 1.5 * threshold_m / m_per_deg_lon + + +# ── Refresh: materialize FDW → cad_buildings_local (single bulk scan) ───────── +def refresh_cad_buildings_local(db: Session) -> int: + """TRUNCATE cad_buildings_local; bulk INSERT from gendesign_cad_buildings FDW. + + Single FDW scan (no per-row round-trips). Builds Point(4326) geom from lon/lat. + Idempotent: TRUNCATE+INSERT in one transaction — the table is fully replaced atomically. + + Returns the number of rows loaded. + """ + db.execute(text("TRUNCATE cad_buildings_local")) + db.execute( + text( + """ + INSERT INTO cad_buildings_local ( + cad_num, readable_address, year_built, floors, area_m2, purpose, + lat, lon, geom + ) + SELECT + cad_num, readable_address, year_built, floors, area_m2, purpose, + lat, lon, + ST_SetSRID(ST_MakePoint(lon, lat), 4326) + FROM gendesign_cad_buildings + WHERE lat IS NOT NULL + AND lon IS NOT NULL + AND cad_num IS NOT NULL + ON CONFLICT (cad_num) DO NOTHING + """ + ) + ) + count = db.execute(text("SELECT count(*) FROM cad_buildings_local")).scalar() or 0 + db.commit() + logger.info("refresh_cad_buildings_local: loaded %d rows from FDW", count) + return int(count) + + +# ── Match: LATERAL KNN nearest building → listings.building_cadastral_number ── +# Set-based UPDATE. For each candidate listing (geo present, optionally only NULL bcn) the +# LATERAL subquery picks the single nearest cad_buildings_local building and accepts it only +# if its TRUE distance is within :threshold_m metres. +# +# PERF (measured live, the whole point of this design): +# The gate is a GEOMETRY-space ST_DWithin(cb.geom, l_geom, :deg_gate) — planar degrees, +# fully GIST-indexable, NO per-row geography cast. The `<-> ` KNN order then returns the +# single nearest candidate via the GIST index. ONLY that one nearest row is converted to a +# true metric distance via ST_DistanceSphere (spherical metres) and compared to :threshold_m. +# This is ~175× faster than a geography-cast ST_DWithin in the WHERE: a geography filter +# forced a per-candidate cast+recheck that the planner could not push into the index, so the +# full 41k-listing UPDATE ran 58s+ and never finished; the geometry-gate version does a +# 5000-row chunk in ~330ms and the full ~41k backfill in a few seconds. +# +# :deg_gate is a degree radius that ALWAYS encloses :threshold_m at EKB latitude (~56.8°): +# 1° lat ≈ 111_320 m, and we widen by /cos(lat) for longitude + a safety factor, so the +# gate is a superset of the true threshold circle and the ST_DistanceSphere recheck is what +# actually decides acceptance (no false negatives from the gate). +# +# Batching: pick a chunk of candidate listing ids first (LIMIT :batch_size), then run the +# LATERAL UPDATE only over that chunk — bounds lock/WAL footprint on the ~43k table and lets +# the run loop checkpoint a heartbeat between chunks. only_missing=true makes each chunk drain +# monotonically (filled rows drop out of the candidate set), so the loop terminates. +_MATCH_CHUNK_SQL = text( + """ + WITH chunk AS ( + SELECT l.id, l.geom AS l_geom + FROM listings l + WHERE l.geom IS NOT NULL + AND ( + CAST(:only_missing AS boolean) = false + OR l.building_cadastral_number IS NULL + ) + ORDER BY l.id + LIMIT CAST(:batch_size AS int) + OFFSET CAST(:offset AS int) + ), + matched AS ( + SELECT c.id, m.cad_num + FROM chunk c + JOIN LATERAL ( + SELECT cb.cad_num, + ST_DistanceSphere(cb.geom, c.l_geom) AS dist_m + FROM cad_buildings_local cb + WHERE ST_DWithin(cb.geom, c.l_geom, CAST(:deg_gate AS double precision)) + ORDER BY cb.geom <-> c.l_geom + LIMIT 1 + ) m ON true + WHERE m.dist_m <= CAST(:threshold_m AS double precision) + ) + UPDATE listings l + SET building_cadastral_number = matched.cad_num + FROM matched + WHERE l.id = matched.id + AND ( + CAST(:only_missing AS boolean) = false + OR l.building_cadastral_number IS NULL + ) + """ +) + +_CANDIDATES_SQL = text( + """ + SELECT count(*) FROM listings l + WHERE l.geom IS NOT NULL + AND (CAST(:only_missing AS boolean) = false OR l.building_cadastral_number IS NULL) + """ +) + + +@dataclass +class CadMatchResult: + buildings_loaded: int = 0 # rows in cad_buildings_local after refresh + candidates_total: int = 0 # listings eligible for matching (geo present + filter) + listings_matched: int = 0 # listings updated with a building_cadastral_number + threshold_m: int = DEFAULT_THRESHOLD_M + duration_sec: float = field(default=0.0) + + def to_counters(self) -> dict[str, int]: + return { + "buildings_loaded": self.buildings_loaded, + "candidates_total": self.candidates_total, + "listings_matched": self.listings_matched, + "threshold_m": self.threshold_m, + "duration_sec": int(self.duration_sec), + } + + +def match_listings_to_buildings( + db: Session, + *, + threshold_m: int = DEFAULT_THRESHOLD_M, + batch_size: int = DEFAULT_BATCH_SIZE, + only_missing: bool = True, +) -> int: + """Set-based LATERAL KNN UPDATE: fill listings.building_cadastral_number. + + GEO-NEAREST (approximate): each candidate listing is matched to the nearest + cad_buildings_local building within `threshold_m` metres. The threshold is logged. + + Chunked by listing id to bound lock/WAL. With only_missing=true the candidate set + shrinks each chunk (filled rows drop out), so we iterate with a moving OFFSET reset to 0 + after every successful chunk (the WHERE re-filters). Returns total listings matched. + """ + logger.info( + "match_listings_to_buildings: GEO-NEAREST match threshold_m=%d batch_size=%d " + "only_missing=%s (approximate: nearest building, not exact cadastral)", + threshold_m, + batch_size, + only_missing, + ) + + deg_gate = _deg_gate_for(float(threshold_m)) + total_matched = 0 + offset = 0 + while True: + result = db.execute( + _MATCH_CHUNK_SQL, + { + "threshold_m": float(threshold_m), + "deg_gate": deg_gate, + "batch_size": batch_size, + "offset": offset, + "only_missing": only_missing, + }, + ) + matched = result.rowcount + db.commit() + total_matched += matched + + if only_missing: + # Filled rows drop out of the candidate set → next chunk starts fresh at 0. + # Terminate when a chunk matches nothing (no more matchable candidates). + if matched == 0: + break + offset = 0 + else: + # Re-match all: walk the id space once via OFFSET; stop when a full pass + # over the candidate window yielded a short chunk (no rows updated). + offset += batch_size + if matched == 0: + break + + logger.info( + "match_listings_to_buildings: chunk matched=%d total=%d", matched, total_matched + ) + + logger.info( + "match_listings_to_buildings: DONE total_matched=%d threshold_m=%d", + total_matched, + threshold_m, + ) + return total_matched + + +# ── Combined refresh + match (callable directly or via run wrapper) ─────────── +def refresh_and_match( + db: Session, + *, + threshold_m: int = DEFAULT_THRESHOLD_M, + batch_size: int = DEFAULT_BATCH_SIZE, + only_missing: bool = True, +) -> CadMatchResult: + """Refresh the local cad mirror, then geo-match listings. Returns a CadMatchResult.""" + start = time.monotonic() + res = CadMatchResult(threshold_m=threshold_m) + + res.buildings_loaded = refresh_cad_buildings_local(db) + res.candidates_total = int( + db.execute(_CANDIDATES_SQL, {"only_missing": only_missing}).scalar() or 0 + ) + res.listings_matched = match_listings_to_buildings( + db, threshold_m=threshold_m, batch_size=batch_size, only_missing=only_missing + ) + res.duration_sec = time.monotonic() - start + + coverage = 100.0 * res.listings_matched / res.candidates_total if res.candidates_total else 0.0 + logger.info( + "refresh_and_match: DONE buildings=%d candidates=%d matched=%d " + "coverage=%.1f%% threshold_m=%d duration=%.1fs", + res.buildings_loaded, + res.candidates_total, + res.listings_matched, + coverage, + threshold_m, + res.duration_sec, + ) + return res + + +# ── Run lifecycle wrapper (scheduler entrypoint) ───────────────────────────── +def run_cadastral_geo_match(db: Session, *, run_id: int, params: dict) -> CadMatchResult: + """Run-lifecycle wrapper for the combined refresh+match job (sync, DB-only). + + Launched by the in-app scheduler (source='cadastral_geo_match') via + trigger_cadastral_geo_match_run, or manually. Refreshes cad_buildings_local from the + FDW (one bulk scan) then geo-matches listings.building_cadastral_number to the nearest + cadastral building within threshold_m. + + Params (default_params jsonb): + threshold_m: max nearest-building distance to accept (metres, default 50). + batch_size: listings per LATERAL UPDATE chunk (default 5000). + only_missing: only fill WHERE building_cadastral_number IS NULL (default true). + + Finalises scrape_runs (mark_done / mark_failed) with counters. + """ + threshold_m = int(params.get("threshold_m", DEFAULT_THRESHOLD_M)) + batch_size = int(params.get("batch_size", DEFAULT_BATCH_SIZE)) + only_missing = bool(params.get("only_missing", True)) + + counters: dict[str, int] = { + "buildings_loaded": 0, + "candidates_total": 0, + "listings_matched": 0, + "threshold_m": threshold_m, + } + + try: + runs_mod.update_heartbeat(db, run_id, counters) + + res = refresh_and_match( + db, + threshold_m=threshold_m, + batch_size=batch_size, + only_missing=only_missing, + ) + + counters = res.to_counters() + runs_mod.mark_done(db, run_id, counters) + logger.info( + "run_cadastral_geo_match: run_id=%d DONE buildings=%d candidates=%d " + "matched=%d threshold_m=%d duration=%.1fs", + run_id, + res.buildings_loaded, + res.candidates_total, + res.listings_matched, + threshold_m, + res.duration_sec, + ) + return res + except Exception as exc: + logger.exception("run_cadastral_geo_match: run_id=%d FAILED", run_id) + try: + db.rollback() + except Exception: + pass + runs_mod.mark_failed(db, run_id, str(exc)[:1000], counters) + raise diff --git a/tradein-mvp/backend/data/sql/124_cad_buildings_local.sql b/tradein-mvp/backend/data/sql/124_cad_buildings_local.sql new file mode 100644 index 00000000..0cbf2d83 --- /dev/null +++ b/tradein-mvp/backend/data/sql/124_cad_buildings_local.sql @@ -0,0 +1,47 @@ +-- 124_cad_buildings_local.sql +-- Local materialized mirror of gendesign_cad_buildings (postgres_fdw foreign table) +-- for fast geo-nearest building matching of listings. +-- +-- WHY (perf fact, measured live): +-- gendesign_cad_buildings is a postgres_fdw foreign table (~36.7k EKB rows) with NO +-- PostGIS geom — only scalar lat/lon. A per-listing nearest-building query over the FDW +-- takes ~1.16s/row (lat/lon bbox prefilter). With ~43k listings that is ~13h — UNUSABLE. +-- Solution: materialize the FDW once into a LOCAL table with a real Point geom + GIST +-- index, then run a fast local KNN nearest-neighbour join (geom <-> point, GIST-backed). +-- +-- This migration creates the EMPTY table + indexes only. It does NOT read the FDW — +-- so deploy never depends on the gendesign DB / FDW server being reachable. +-- The table is populated by the refresh job (app/tasks/refresh_cad_buildings_local.py), +-- driven by the in-app scheduler (source='cadastral_geo_match', combined refresh+match). +-- +-- Schedule seed (cadastral_geo_match) lives in 125_scrape_schedules_seed_cadastral_geo_match.sql. +-- +-- DEPENDENCIES: PostGIS (postgis extension, present since geom on listings/houses). +-- Idempotent: CREATE TABLE/INDEX IF NOT EXISTS. Safe to re-run. + +BEGIN; + +CREATE TABLE IF NOT EXISTS cad_buildings_local ( + cad_num text PRIMARY KEY, + readable_address text, + year_built int, + floors int, + area_m2 numeric, + purpose text, + lat double precision, + lon double precision, + geom geometry(Point, 4326) +); + +COMMENT ON TABLE cad_buildings_local IS + 'Local mirror of gendesign_cad_buildings FDW (Rosreestr building registry, EKB ~36.7k). ' + 'Populated by app/tasks/refresh_cad_buildings_local.py (single bulk FDW scan, TRUNCATE+INSERT). ' + 'Point geom + GIST enables fast local KNN nearest-building matching ' + '(run_cadastral_geo_match) for listings.building_cadastral_number. ' + 'Migration 124. Refresh+match driven by scheduler source=cadastral_geo_match.'; + +-- GIST on geom — required for the KNN <-> operator and ST_DWithin predicate. +CREATE INDEX IF NOT EXISTS cad_buildings_local_geom_idx + ON cad_buildings_local USING GIST (geom); + +COMMIT; diff --git a/tradein-mvp/backend/data/sql/125_scrape_schedules_seed_cadastral_geo_match.sql b/tradein-mvp/backend/data/sql/125_scrape_schedules_seed_cadastral_geo_match.sql new file mode 100644 index 00000000..bd51e2dd --- /dev/null +++ b/tradein-mvp/backend/data/sql/125_scrape_schedules_seed_cadastral_geo_match.sql @@ -0,0 +1,58 @@ +-- 125_scrape_schedules_seed_cadastral_geo_match.sql +-- Scheduler seed for the geo-nearest cadastral building matcher (combined refresh + match). +-- +-- WHAT (source='cadastral_geo_match'): +-- trigger_cadastral_geo_match_run (scheduler.py) → run_cadastral_geo_match +-- (tasks/cadastral_geo_match.py). One combined nightly job: +-- 1. REFRESH: TRUNCATE cad_buildings_local; bulk INSERT from gendesign_cad_buildings FDW +-- (single FDW scan, ~36.7k EKB rows, builds Point geom). +-- 2. MATCH: set-based LATERAL KNN UPDATE — fills listings.building_cadastral_number +-- from the nearest cad_buildings_local within threshold_m (default 50m). +-- Refresh-then-match in ONE run guarantees the match always reads a fresh mirror. +-- +-- The match is GEO-NEAREST (approximate): a street-level-geocoded listing matches the +-- nearest building, not necessarily its exact cadastral building. Deliberate first +-- increment; exact cadastral + parcel-containment deferred (cad_parcels FDW not exposed). +-- +-- Pure internal DB op (one FDW read + local UPDATE) — no external HTTP, no anti-bot. +-- SAFE to enable by default (enabled=true), like listing_source_snapshot / refresh_search_matview. +-- +-- Window 09:00-10:00 UTC (12:00-13:00 MSK): +-- - After geocode_missing_listings (06:00-09:00 UTC) → listings have fresh lat/lon/geom. +-- - Before/independent of avito_detail_backfill proxy work; pure-DB so no contention. +-- +-- default_params: +-- threshold_m -- max nearest-building distance to accept a match (metres). 50 = balance +-- of coverage vs precision for street-level geocodes (see PR EXPLAIN evidence). +-- batch_size -- listings updated per LATERAL UPDATE statement (chunked to bound lock/WAL). +-- only_missing -- if true (default) only fills WHERE building_cadastral_number IS NULL +-- (idempotent: a re-run is a no-op once filled). false = re-match all. +-- +-- next_run_at = tomorrow 09:00 UTC (NOT NULL — avoid deploy-time fire, follow avito/cian seed +-- convention; NULL would make get_due_schedules pick it up on the very next tick). +-- +-- DEPENDENCIES: 052_scrape_schedules.sql (table + UNIQUE(source)), 124_cad_buildings_local.sql. +-- Idempotent: ON CONFLICT (source) DO NOTHING. + +BEGIN; + +INSERT INTO scrape_schedules ( + source, + enabled, + window_start_hour, + window_end_hour, + next_run_at, + default_params +) +VALUES +( + 'cadastral_geo_match', + true, + 9, + 10, + ((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 9)) AT TIME ZONE 'UTC', + '{"threshold_m": 50, "batch_size": 5000, "only_missing": true}'::jsonb +) +ON CONFLICT (source) DO NOTHING; + +COMMIT; diff --git a/tradein-mvp/backend/tests/tasks/test_cadastral_geo_match.py b/tradein-mvp/backend/tests/tasks/test_cadastral_geo_match.py new file mode 100644 index 00000000..da71a921 --- /dev/null +++ b/tradein-mvp/backend/tests/tasks/test_cadastral_geo_match.py @@ -0,0 +1,386 @@ +"""Tests for the geo-nearest cadastral building matcher (#cadastral-geo-match). + +Convention mirrors test_listing_source_snapshot / test_asking_to_sold_ratio: the matcher +is SQL-heavy and the gate has no live Postgres, so most assertions are STATIC — we read the +emitted SQL via the text() clauses and inspect.getsource and check: + - the LATERAL KNN shape (ST_DWithin geography gate + `<->` KNN order + LIMIT 1), + - the threshold/only_missing params are bound (psycopg-v3 CAST discipline, no :p::type), + - the refresh does a single bulk FDW scan (TRUNCATE + INSERT ... FROM gendesign_cad_buildings), + - scheduler wiring (trigger fn + dispatch branch), + - migration 124 (table + GIST) and 125 (schedule seed) contents. + +Plus a fake-db behavioural test driving the chunk-loop + counter logic without Postgres. + +An OPTIONAL real-PostGIS behavioural test (test_real_knn_*) asserts the actual +nearest-within-threshold / beyond-threshold→NULL semantics when a PostGIS database is +reachable; it self-SKIPS otherwise so the hermetic gate never depends on it. +""" + +from __future__ import annotations + +import inspect +import math +import os +import re +from pathlib import Path +from typing import Any + +import pytest + +# settings needs a DSN at import (same dance as the sibling tests); these are static/fake-db. +os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test") + +from app.services import scheduler +from app.tasks import cadastral_geo_match as cgm + +_SQL_DIR = Path(__file__).resolve().parents[2] / "data" / "sql" +_MIGRATION_124 = _SQL_DIR / "124_cad_buildings_local.sql" +_MIGRATION_125 = _SQL_DIR / "125_scrape_schedules_seed_cadastral_geo_match.sql" + +_MATCH_SQL = str(cgm._MATCH_CHUNK_SQL.text) +_CANDIDATES_SQL = str(cgm._CANDIDATES_SQL.text) +_REFRESH_SRC = inspect.getsource(cgm.refresh_cad_buildings_local) +_MATCH_SRC = inspect.getsource(cgm.match_listings_to_buildings) + + +# ── Refresh: single bulk FDW scan ───────────────────────────────────────────── + + +def test_refresh_truncates_then_bulk_inserts_from_fdw() -> None: + assert "TRUNCATE cad_buildings_local" in _REFRESH_SRC + assert "INSERT INTO cad_buildings_local" in _REFRESH_SRC + assert "FROM gendesign_cad_buildings" in _REFRESH_SRC + # Builds a Point(4326) geom from lon/lat in the same scan. + assert "ST_SetSRID(ST_MakePoint(lon, lat), 4326)" in _REFRESH_SRC + + +def test_refresh_filters_null_coords_and_cad_num() -> None: + assert "lat IS NOT NULL" in _REFRESH_SRC + assert "lon IS NOT NULL" in _REFRESH_SRC + assert "cad_num IS NOT NULL" in _REFRESH_SRC + + +def test_refresh_is_single_scan_not_per_row() -> None: + """The FDW must be read once (one INSERT...SELECT), never per listing — the whole point.""" + assert _REFRESH_SRC.count("FROM gendesign_cad_buildings") == 1 + + +# ── Match: LATERAL KNN shape ────────────────────────────────────────────────── + + +def test_match_uses_lateral_knn_nearest_one() -> None: + flat = re.sub(r"\s+", " ", _MATCH_SQL) + assert "JOIN LATERAL" in flat + # GIST-backed KNN order + single nearest candidate. + assert "ORDER BY cb.geom <-> c.l_geom" in flat + assert "LIMIT 1" in flat + + +def test_match_gates_with_geometry_dwithin_and_distancesphere_recheck() -> None: + """Perf-critical: geometry-space ST_DWithin (GIST, degrees) pre-gate + ST_DistanceSphere + metric recheck — NOT a geography-cast ST_DWithin in the WHERE (that ran 58s+ unfinished).""" + flat = re.sub(r"\s+", " ", _MATCH_SQL) + # GIST-indexable degree-space gate (no geography cast on the filtered table). + assert "ST_DWithin(cb.geom, c.l_geom, CAST(:deg_gate AS double precision))" in flat + # The geography cast must NOT appear in the gate (that defeats the index). + assert "CAST(cb.geom AS geography)" not in flat + # True metric distance computed only on the single nearest row, gated by threshold_m. + assert "ST_DistanceSphere(cb.geom, c.l_geom) AS dist_m" in flat + assert "m.dist_m <= CAST(:threshold_m AS double precision)" in flat + + +def test_match_targets_building_cadastral_number() -> None: + flat = re.sub(r"\s+", " ", _MATCH_SQL) + assert "UPDATE listings" in flat + assert "SET building_cadastral_number = matched.cad_num" in flat + + +def test_match_only_missing_param_is_bound_and_filters() -> None: + """only_missing=true → only fills WHERE building_cadastral_number IS NULL (idempotent).""" + flat = re.sub(r"\s+", " ", _MATCH_SQL) + assert "CAST(:only_missing AS boolean)" in flat + assert "l.building_cadastral_number IS NULL" in flat + # geo-present gate so non-geocoded listings never match. + assert "l.geom IS NOT NULL" in flat + + +def test_match_is_chunked() -> None: + flat = re.sub(r"\s+", " ", _MATCH_SQL) + assert "LIMIT CAST(:batch_size AS int)" in flat + assert "OFFSET CAST(:offset AS int)" in flat + + +def test_deg_gate_encloses_threshold() -> None: + """The degree gate must be a SUPERSET of the metric threshold at EKB latitude. + + A point exactly threshold_m metres away (in the tighter longitude axis) must fall + INSIDE the gate, else the GIST pre-filter would drop true-positive nearest buildings. + """ + m_per_deg_lon = cgm._M_PER_DEG_LAT * math.cos(math.radians(cgm._EKB_LAT_DEG)) + for threshold_m in (30, 50, 100): + deg_gate = cgm._deg_gate_for(float(threshold_m)) + # Longitude-axis metres covered by the gate must exceed the threshold. + assert deg_gate * m_per_deg_lon >= threshold_m + # And by the latitude axis too (lat degrees are longer → even more margin). + assert deg_gate * cgm._M_PER_DEG_LAT >= threshold_m + + +def test_no_psycopg_v3_colon_colon_cast() -> None: + """psycopg v3: never :param::type — must use CAST(:param AS type).""" + # No ':name::type' bound-param cast anywhere in the app SQL (the v3 trap). + assert not re.search(r":\w+::", _MATCH_SQL) + assert not re.search(r":\w+::", _CANDIDATES_SQL) + assert not re.search(r":\w+::", _REFRESH_SRC) + + +def test_matcher_docstring_marks_approximation_and_threshold() -> None: + """Spec: the geo-nearest approximation must be explicit + threshold logged.""" + doc = cgm.__doc__ or "" + assert "GEO-NEAREST" in doc + match_doc = cgm.match_listings_to_buildings.__doc__ or "" + assert "GEO-NEAREST" in match_doc + # threshold logged in match body + assert "threshold_m=%d" in _MATCH_SRC + + +# ── Scheduler wiring ────────────────────────────────────────────────────────── + + +def test_scheduler_has_trigger_and_dispatch() -> None: + assert hasattr(scheduler, "trigger_cadastral_geo_match_run") + loop_src = inspect.getsource(scheduler.scheduler_loop) + assert 'source == "cadastral_geo_match"' in loop_src + assert "trigger_cadastral_geo_match_run(db, sch)" in loop_src + + +def test_trigger_claims_run_and_runs_in_executor() -> None: + src = inspect.getsource(scheduler.trigger_cadastral_geo_match_run) + assert "_claim_run(db, schedule_row)" in src + # sync DB-only task → run_in_executor (not a bare async call). + assert "run_in_executor" in src + assert "run_cadastral_geo_match" in src + + +# ── Migration content ───────────────────────────────────────────────────────── + + +def test_migration_124_creates_table_and_gist() -> None: + sql = _MIGRATION_124.read_text(encoding="utf-8") + assert "CREATE TABLE IF NOT EXISTS cad_buildings_local" in sql + assert "cad_num text PRIMARY KEY" in sql + assert "geom geometry(Point, 4326)" in sql + assert "USING GIST (geom)" in sql + assert "IF NOT EXISTS cad_buildings_local_geom_idx" in sql + assert "BEGIN;" in sql and "COMMIT;" in sql + # The migration must NOT read the FDW (deploy independence) — no FROM/INSERT against it. + # (The FDW name may appear in comments; only executable references are forbidden.) + code_lines = [ln for ln in sql.splitlines() if not ln.lstrip().startswith("--")] + code = "\n".join(code_lines) + assert "FROM gendesign_cad_buildings" not in code + assert "INSERT INTO cad_buildings_local" not in code # populate is the refresh job, not DDL + + +def test_migration_125_seeds_schedule_not_null_next_run() -> None: + sql = _MIGRATION_125.read_text(encoding="utf-8") + assert "INSERT INTO scrape_schedules" in sql + assert "'cadastral_geo_match'" in sql + assert "ON CONFLICT (source) DO NOTHING" in sql + # next_run_at NOT NULL = tomorrow (avoid deploy-time fire). + assert "CURRENT_DATE + INTERVAL '1 day'" in sql + assert '"threshold_m": 50' in sql + assert "BEGIN;" in sql and "COMMIT;" in sql + + +# ── Fake-db behavioural: chunk loop + counters ─────────────────────────────── + + +class _FakeResult: + def __init__(self, rowcount: int, scalar: Any = None) -> None: + self.rowcount = rowcount + self._scalar = scalar + + def scalar(self) -> Any: + return self._scalar + + +class _FakeDB: + """Minimal Session stand-in: scripts execute() returns queued results in order.""" + + def __init__(self, results: list[_FakeResult]) -> None: + self._results = list(results) + self.executed: list[str] = [] + self.commits = 0 + + def execute(self, clause: Any, params: dict | None = None) -> _FakeResult: + self.executed.append(str(getattr(clause, "text", clause))) + return self._results.pop(0) + + def commit(self) -> None: + self.commits += 1 + + def rollback(self) -> None: # pragma: no cover + pass + + +def test_match_loop_terminates_and_counts(monkeypatch: pytest.MonkeyPatch) -> None: + """only_missing loop: chunks match 3, then 2, then 0 → total 5, stops on empty chunk.""" + db = _FakeDB( + [ + _FakeResult(3), # chunk 1 + _FakeResult(2), # chunk 2 + _FakeResult(0), # chunk 3 → terminate + ] + ) + total = cgm.match_listings_to_buildings(db, threshold_m=50, batch_size=10, only_missing=True) + assert total == 5 + assert db.commits == 3 # one per chunk + + +def test_run_wrapper_marks_done_with_counters(monkeypatch: pytest.MonkeyPatch) -> None: + """run_cadastral_geo_match: refresh → candidates → match → mark_done(counters).""" + marked: dict[str, Any] = {} + monkeypatch.setattr(cgm.runs_mod, "update_heartbeat", lambda *a, **k: None) + monkeypatch.setattr( + cgm.runs_mod, + "mark_done", + lambda _db, run_id, counters: marked.update(run_id=run_id, counters=dict(counters)), + ) + monkeypatch.setattr(cgm.runs_mod, "mark_failed", lambda *a, **k: None) + + # Stub the heavy SQL fns; assert the wrapper plumbs counters correctly. + monkeypatch.setattr(cgm, "refresh_cad_buildings_local", lambda _db: 36732) + monkeypatch.setattr( + cgm, + "match_listings_to_buildings", + lambda _db, **k: 12000, + ) + + class _CandDB: + def execute(self, *a: Any, **k: Any) -> _FakeResult: + return _FakeResult(0, scalar=40000) + + out = cgm.run_cadastral_geo_match( + _CandDB(), # type: ignore[arg-type] + run_id=7, + params={"threshold_m": 50, "batch_size": 100, "only_missing": True}, + ) + assert out.buildings_loaded == 36732 + assert out.candidates_total == 40000 + assert out.listings_matched == 12000 + assert out.threshold_m == 50 + assert marked["run_id"] == 7 + assert marked["counters"]["listings_matched"] == 12000 + assert marked["counters"]["threshold_m"] == 50 + + +def test_run_wrapper_marks_failed_on_error(monkeypatch: pytest.MonkeyPatch) -> None: + failed: dict[str, Any] = {} + monkeypatch.setattr(cgm.runs_mod, "update_heartbeat", lambda *a, **k: None) + monkeypatch.setattr(cgm.runs_mod, "mark_done", lambda *a, **k: None) + monkeypatch.setattr( + cgm.runs_mod, + "mark_failed", + lambda _db, run_id, err, counters: failed.update(run_id=run_id, err=err), + ) + + def _boom(_db: Any) -> int: + raise RuntimeError("fdw down") + + monkeypatch.setattr(cgm, "refresh_cad_buildings_local", _boom) + + class _DB: + def rollback(self) -> None: + pass + + with pytest.raises(RuntimeError): + cgm.run_cadastral_geo_match(_DB(), run_id=9, params={}) # type: ignore[arg-type] + assert failed["run_id"] == 9 + assert "fdw down" in failed["err"] + + +# ── Optional real-PostGIS behavioural test (self-skips without a DB) ────────── + + +def _live_session() -> Any | None: + """Return a SQLAlchemy Session if a PostGIS DB is reachable, else None.""" + try: + from sqlalchemy import create_engine + from sqlalchemy.orm import sessionmaker + + dsn = os.environ.get("TEST_DATABASE_URL") or os.environ.get("DATABASE_URL", "") + if not dsn or "localhost:5432/test" in dsn: + return None + engine = create_engine(dsn, future=True) + conn = engine.connect() + conn.execute(__import__("sqlalchemy").text("SELECT PostGIS_Version()")) + conn.close() + return sessionmaker(bind=engine, future=True)() + except Exception: + return None + + +@pytest.mark.skipif(_live_session() is None, reason="no reachable PostGIS test DB") +def test_real_knn_nearest_within_threshold_picked() -> None: + """With a real PostGIS DB: nearest building within threshold is picked; beyond → NULL.""" + from sqlalchemy import text as _t + + db = _live_session() + assert db is not None + try: + db.execute( + _t( + "CREATE TEMP TABLE cad_buildings_local " + "(LIKE cad_buildings_local INCLUDING ALL) ON COMMIT DROP" + ) + ) + except Exception: + # Local table may not exist in this DB → build a minimal temp equivalent. + db.rollback() + db.execute( + _t( + "CREATE TEMP TABLE cad_buildings_local (" + "cad_num text PRIMARY KEY, geom geometry(Point,4326)) ON COMMIT DROP" + ) + ) + # Two buildings: one 10m from the listing, one 500m away. + db.execute( + _t( + "INSERT INTO cad_buildings_local (cad_num, geom) VALUES " + "('66:01:NEAR', ST_SetSRID(ST_MakePoint(60.6000, 56.8380),4326))," + "('66:01:FAR', ST_SetSRID(ST_MakePoint(60.6100, 56.8380),4326))" + ) + ) + # Same geometry-gate + ST_DistanceSphere recheck shape as production. + deg_gate = cgm._deg_gate_for(50.0) + # listing point ~ at NEAR; nearest within 50m must be 66:01:NEAR. + row = db.execute( + _t( + "SELECT cad_num FROM (" + " SELECT cb.cad_num, " + " ST_DistanceSphere(cb.geom, ST_SetSRID(ST_MakePoint(60.6001,56.8380),4326)) " + " AS dist_m " + " FROM cad_buildings_local cb " + " WHERE ST_DWithin(cb.geom, ST_SetSRID(ST_MakePoint(60.6001,56.8380),4326), :g) " + " ORDER BY cb.geom <-> ST_SetSRID(ST_MakePoint(60.6001,56.8380),4326) LIMIT 1" + ") m WHERE m.dist_m <= 50" + ), + {"g": deg_gate}, + ).fetchone() + assert row is not None and row[0] == "66:01:NEAR" + # A point 500m+ from any building → no match within 50m. + none_row = db.execute( + _t( + "SELECT cad_num FROM (" + " SELECT cb.cad_num, " + " ST_DistanceSphere(cb.geom, ST_SetSRID(ST_MakePoint(60.7000,56.9000),4326)) " + " AS dist_m " + " FROM cad_buildings_local cb " + " WHERE ST_DWithin(cb.geom, ST_SetSRID(ST_MakePoint(60.7000,56.9000),4326), :g) " + " ORDER BY cb.geom <-> ST_SetSRID(ST_MakePoint(60.7000,56.9000),4326) LIMIT 1" + ") m WHERE m.dist_m <= 50" + ), + {"g": deg_gate}, + ).fetchone() + assert none_row is None + db.rollback() + db.close()