feat(cadastral): geo-nearest building matcher via local cad mirror

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).
This commit is contained in:
bot-backend 2026-06-18 10:18:54 +03:00
parent bf91eb0be0
commit edd187b47f
5 changed files with 894 additions and 0 deletions

View file

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

View file

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

View file

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

View file

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

View file

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