feat(tradein): location-coef MVP через FDW-мост к OSM POI Птицы (#2045 PR-B) #2314

Merged
lekss361 merged 1 commit from feat/tradein-location-coef-poi into main 2026-07-03 20:55:32 +00:00
12 changed files with 1032 additions and 0 deletions

View file

@ -26,6 +26,8 @@ from app.schemas.trade_in import (
HouseAnalyticsResponse,
HouseInfoForEstimate,
IMVBenchmarkResponse,
LocationCoefFactorOut,
LocationCoefResponse,
PhotoMeta,
PlacementHistoryEntry,
PriceHistoryYearPoint,
@ -1519,6 +1521,79 @@ def get_estimate_imv_benchmark(
)
# ── Location-coef POI scoring (#2045 BE-3, LocationDrawer) ────────────────────
@router.get("/location-coef", response_model=LocationCoefResponse)
def get_location_coef(
estimate_id: UUID,
db: Annotated[Session, Depends(get_db)],
x_authenticated_user: Annotated[str | None, Header(alias="X-Authenticated-User")] = None,
radius_m: int | None = None,
) -> LocationCoefResponse:
"""Location-coefficient POI-скоринг для оценки (#2045 BE-3, LocationDrawer).
Резолвит lat/lon/median_price оценки, считает coef через
app.services.location_coef.compute_location_coef straight-line POI weighted score
(портировано из Site Finder poi_score.py) поверх локального зеркала osm_poi_ekb_local,
обновляемого scheduler'ом (source=osm_poi_ekb_refresh). result_price_rub = round(
base_price_rub * coef).
404 оценки нет / IDOR (тот же _assert_estimate_access_by_id, что и у соседних
derived-роутов). radius_m опционален (None DEFAULT_RADIUS_M=1200м, подобран для
МКД, НЕ Ptica-дефолт 2000м для участков); явное значение клэмпится в [500, 3000].
Graceful fallback (НЕ 500, НЕ сфабрикованные факторы):
- osm_poi_ekb_local пуста/не отрефрешена на этом окружении coef=1.0, factors=[],
geo_source="unavailable" (см. compute_location_coef).
- у оценки нет lat/lon (легаси/geo-fallback не сработал на POST) тот же fallback.
"""
_assert_estimate_access_by_id(db, estimate_id, x_authenticated_user)
row = db.execute(
text(
"""
SELECT lat, lon, median_price
FROM trade_in_estimates
WHERE id = CAST(:id AS uuid)
"""
),
{"id": str(estimate_id)},
).fetchone()
if row is None:
raise HTTPException(status_code=404, detail="estimate not found")
base_price_rub = int(row.median_price or 0)
if row.lat is None or row.lon is None:
logger.info("location_coef: estimate=%s has no lat/lon — unavailable fallback", estimate_id)
return LocationCoefResponse(
coef=1.0,
factors=[],
geo_source="unavailable",
base_price_rub=base_price_rub,
result_price_rub=base_price_rub,
)
from app.services.location_coef import DEFAULT_RADIUS_M, compute_location_coef
resolved_radius = DEFAULT_RADIUS_M if radius_m is None else max(500, min(radius_m, 3000))
result = compute_location_coef(db, float(row.lat), float(row.lon), radius_m=resolved_radius)
return LocationCoefResponse(
coef=result.coef,
factors=[
LocationCoefFactorOut(
poi_type=f.poi_type, name=f.name, distance_m=f.distance_m, weight=f.weight
)
for f in result.factors
],
geo_source=result.geo_source,
base_price_rub=base_price_rub,
result_price_rub=round(base_price_rub * result.coef),
)
# ── Street-level deals (rosreestr open dataset) ───────────────────────────────

View file

@ -549,3 +549,31 @@ class QuotaStatus(BaseModel):
used: int # использовано в текущем месяце
remaining: int # max(0, limit - used)
unlimited: bool # True для admin / kopylov / без заголовка
class LocationCoefFactorOut(BaseModel):
"""Один POI-фактор в ответе GET /api/v1/trade-in/location-coef (#2045 BE-3)."""
poi_type: str # категория POI (school/kindergarten/metro_stop/... — те же значения,
# что в osm_poi_ekb на стороне gendesign)
name: str | None
distance_m: float
weight: float
class LocationCoefResponse(BaseModel):
"""Ответ GET /api/v1/trade-in/location-coef (#2045 BE-3, LocationDrawer).
coef MVP-эвристика (НЕ откалибрована на реальных ценовых дельтах, см.
app/services/location_coef.py::_score_to_coef), диапазон [0.95, 1.05].
result_price_rub = round(base_price_rub * coef).
geo_source="unavailable" osm_poi_ekb_local пуста/не отрефрешена на этом окружении
(graceful fallback: coef=1.0, factors=[], НЕ 500 и НЕ сфабрикованные факторы).
"""
coef: float
factors: list[LocationCoefFactorOut]
geo_source: str
base_price_rub: int
result_price_rub: int

View file

@ -0,0 +1,200 @@
"""Location-coefficient POI scoring for trade-in estimates (#2045 BE-3, LocationDrawer).
Ported straight-line formula from Site Finder (ПТИЦА)
`backend/app/services/site_finder/poi_score.py::compute_poi_weighted_top7`:
weight = (1 / (distance_m + 100)) * CATEGORY_WEIGHTS[category]
Reads from the LOCAL mirror table `osm_poi_ekb_local` (populated by
`app/tasks/osm_poi_ekb_refresh.py` from the `gendesign_osm_poi_ekb` FDW see migrations
168-170). We deliberately do NOT query the FDW directly per-request: the same per-row cost
measured for the analogous cadastral-buildings FDW (~1.16s/row without a geom index on the
remote) would make a synchronous endpoint unusable.
Scope (MVP, #2045 BE-3):
- Straight-line distance only. The ORS routing-decay mode from Site Finder
(`compute_poi_routing_decay`) is NOT ported no ORS infrastructure in trade-in, out of
MVP scope.
- Radius tuned for apartments (1000-1500m), NOT Site Finder's 2000m default for land parcels.
- The scorecoef mapping (`_score_to_coef`) is a NEW MVP heuristic, not present in Site
Finder (there POI score is a ranking metric, not a price multiplier) see its docstring.
"""
from __future__ import annotations
import logging
from typing import Any
from pydantic import BaseModel
from sqlalchemy import text
logger = logging.getLogger(__name__)
# Веса по категории — скопированы as-is из Site Finder CATEGORY_WEIGHTS
# (backend/app/services/site_finder/poi_score.py), чтобы ranking POI был согласован
# между продуктами.
CATEGORY_WEIGHTS: dict[str, float] = {
"metro_stop": 6.0,
"school": 5.0,
"kindergarten": 4.5,
"hospital": 4.0,
"shop_mall": 4.0,
"shop_supermarket": 3.5,
"bus_stop": 4.5,
"park": 3.5,
"pharmacy": 2.5,
"tram_stop": 2.0,
"shop_small": 2.0,
"default": 1.0,
}
# Радиус подобран для КВАРТИР (МКД), а не для участков (Ptica default 2000м) —
# пешая доступность в пределах квартала/микрорайона.
DEFAULT_RADIUS_M = 1200
DEFAULT_TOP_N = 7
# Теоретический максимум суммы весов top-7 POI при идеальном расположении (d=0):
# w_i = cat_weight_i / (0 + 100) → max_sum = Σ(top7 cat_weights) / 100.
# Top-7 категорий по убыванию веса: 6.0+5.0+4.5+4.5+4.0+4.0+3.5 = 31.5 (тот же набор, что у Ptica).
_TOP7_WEIGHT_SUM: float = sum(sorted(CATEGORY_WEIGHTS.values(), reverse=True)[:7])
_MAX_STRAIGHT_SCORE: float = _TOP7_WEIGHT_SUM / 100.0 # ≈ 0.315
# coef диапазон ±5% вокруг 1.0 — heuristic v1, НЕ откалибровано на реальных ценовых дельтах
# (в отличие от Ptica, где poi_weighted_score — ранжирующая метрика, не ценовой множитель).
_COEF_BASE = 0.95
_COEF_SPREAD = 0.10
def _category_weight(category: str | None) -> float:
"""Вернуть вес категории. Если не знаем — default."""
return CATEGORY_WEIGHTS.get(category or "default", CATEGORY_WEIGHTS["default"])
class LocationCoefFactor(BaseModel):
"""Один POI-фактор в ответе location-coef."""
poi_type: str
name: str | None
distance_m: float
weight: float
class LocationCoefResult(BaseModel):
"""Результат compute_location_coef — потребляется эндпоинтом location-coef."""
coef: float
factors: list[LocationCoefFactor]
geo_source: str # "osm_poi_ekb" (норма) | "unavailable" (mirror пуста/не отрефрешена)
def _score_to_coef(poi_weighted_score: float) -> float:
"""MVP-эвристика score(0..100) → ценовой коэффициент.
coef = 0.95 + (score/100) * 0.10 диапазон [0.95, 1.05].
ВНИМАНИЕ: это НЕ откалиброванная на реальных ценовых дельтах формула первая рабочая
эвристика для MVP location-coef. Site Finder использует ту же POI-модель как ранжирующую
метрику (poi_weighted_score), а не как прямой ценовой множитель; здесь смысл другой,
поэтому маппинг введён отдельно и явно помечен как heuristic v1.
"""
return round(_COEF_BASE + (poi_weighted_score / 100.0) * _COEF_SPREAD, 4)
_NEAREST_POI_SQL = text(
"""
SELECT
p.name,
p.category,
CAST(
ST_Distance(
p.geom::geography,
ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography
) AS double precision
) AS distance_m
FROM osm_poi_ekb_local p
WHERE p.geom IS NOT NULL
AND ST_DWithin(
p.geom::geography,
ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography,
CAST(:radius_m AS double precision)
)
ORDER BY distance_m ASC
LIMIT :limit
"""
)
def compute_location_coef(
db: Any,
lat: float,
lon: float,
radius_m: int = DEFAULT_RADIUS_M,
top_n: int = DEFAULT_TOP_N,
) -> LocationCoefResult:
"""Посчитать location-coef для координат (lat, lon) по POI из osm_poi_ekb_local.
Graceful fallback: если osm_poi_ekb_local пуста (рефреш ещё не запускался на этом
окружении) возвращает coef=1.0, factors=[], geo_source="unavailable" вместо 500 или
сфабрикованных факторов. Отсутствие POI В РАДИУСЕ у непустой таблицы это легитимный
результат (coef=0.95, factors=[], geo_source="osm_poi_ekb"), не fallback.
Args:
db: SQLAlchemy Session.
lat: широта целевой квартиры.
lon: долгота целевой квартиры.
radius_m: радиус поиска в метрах (default 1200 подобран для МКД, не для участков).
top_n: количество POI, учитываемых в score (default 7).
"""
total = db.execute(text("SELECT count(*) FROM osm_poi_ekb_local")).scalar() or 0
if total == 0:
logger.warning(
"location_coef: osm_poi_ekb_local is empty (refresh job not yet run on this "
"environment) — returning unavailable fallback, no fabricated factors"
)
return LocationCoefResult(coef=1.0, factors=[], geo_source="unavailable")
rows = (
db.execute(
_NEAREST_POI_SQL,
{"lat": lat, "lon": lon, "radius_m": radius_m, "limit": top_n * 10},
)
.mappings()
.all()
)
scored: list[tuple[float, LocationCoefFactor]] = []
for row in rows:
distance_m = float(row["distance_m"])
category = row["category"] or "default"
weight = (1.0 / (distance_m + 100.0)) * _category_weight(category)
scored.append(
(
weight,
LocationCoefFactor(
poi_type=category,
name=row["name"],
distance_m=round(distance_m, 1),
weight=round(weight, 6),
),
)
)
scored.sort(key=lambda pair: pair[0], reverse=True)
top_factors = [factor for _weight, factor in scored[:top_n]]
raw_sum = sum(factor.weight for factor in top_factors)
poi_weighted_score = min(100.0, (raw_sum / _MAX_STRAIGHT_SCORE) * 100.0)
coef = _score_to_coef(poi_weighted_score)
logger.debug(
"location_coef: lat=%.5f lon=%.5f radius=%dm poi_found=%d top=%d " "score=%.1f coef=%.4f",
lat,
lon,
radius_m,
len(rows),
len(top_factors),
poi_weighted_score,
coef,
)
return LocationCoefResult(coef=coef, factors=top_factors, geo_source="osm_poi_ekb")

View file

@ -264,6 +264,19 @@ async def _job_cadastral_geo_match(
)
# ── osm_poi_ekb_refresh — sync FDW-refresh в executor ─────────────────────────
async def _job_osm_poi_ekb_refresh(
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
) -> None:
from app.tasks.osm_poi_ekb_refresh import run_osm_poi_ekb_refresh
loop = asyncio.get_event_loop()
await loop.run_in_executor(
None,
lambda: run_osm_poi_ekb_refresh(db, run_id=run_id, params=params),
)
# ── house_imv_backfill — async Avito-IMV с heartbeat, lifecycle в job ─────────
async def _job_house_imv_backfill(
db: Session, run_id: int, params: dict[str, Any], ctx: SchedulerContext
@ -369,6 +382,7 @@ def build_product_handlers(ctx: SchedulerContext) -> dict[str, Handler]:
"avito_detail_backfill": Handler(_job_avito_detail_backfill, "avito_detail_backfill"),
"yandex_detail_backfill": Handler(_job_yandex_detail_backfill, "yandex_detail_backfill"),
"cadastral_geo_match": Handler(_job_cadastral_geo_match, "cadastral_geo_match"),
"osm_poi_ekb_refresh": Handler(_job_osm_poi_ekb_refresh, "osm_poi_ekb_refresh"),
"house_imv_backfill": Handler(_job_house_imv_backfill, "house_imv_backfill"),
"house_dedup_merge": Handler(_job_house_dedup_merge, "house_dedup_merge"),
"proxy_healthcheck": Handler(

View file

@ -125,6 +125,15 @@ Sources:
DISABLED (enabled=false) deploy is neutral; enable deliberately
after a validated dry-run + manual run; weekly window 04:00-05:00
UTC quiet hour, no scraper/proxy contention)
- osm_poi_ekb_refresh run_osm_poi_ekb_refresh
(tasks/osm_poi_ekb_refresh.py, #2045 BE-3; single bulk FDW scan —
TRUNCATE osm_poi_ekb_local; INSERT ... FROM gendesign_osm_poi_ekb
(Site Finder OSM POI, EKB), building Point(4326) geom from lat/lon.
Feeds app/services/location_coef.py (GET /trade-in/location-coef,
LocationDrawer). OSM POI changes rarely daily cadence is ample.
Pure internal DB op one FDW read + local TRUNCATE+INSERT, no
HTTP/anti-bot; window 10:00-11:00 UTC после cadastral_geo_match
09:00-10:00 UTC)
"""
from __future__ import annotations
@ -1691,6 +1700,45 @@ async def trigger_cadastral_geo_match_run(db: Session, schedule_row: dict[str, A
return run_id
async def trigger_osm_poi_ekb_refresh_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
"""Создать scrape_runs + launch run_osm_poi_ekb_refresh в executor (#2045 BE-3).
Single bulk FDW scan (TRUNCATE osm_poi_ekb_local; INSERT ... FROM gendesign_osm_poi_ekb)
feeds app/services/location_coef.py. No params consumed.
Sync task (one FDW read + local TRUNCATE+INSERT, no async HTTP) run in run_in_executor
by the same pattern as trigger_cadastral_geo_match_run. run_osm_poi_ekb_refresh owns the
scrape_runs lifecycle (mark_done/mark_failed). SAFE to enable seed 170 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.osm_poi_ekb_refresh import run_osm_poi_ekb_refresh
loop = asyncio.get_event_loop()
await loop.run_in_executor(
None,
lambda: run_osm_poi_ekb_refresh(run_db, run_id=run_id, params=params),
)
except Exception:
logger.exception("scheduler: run_osm_poi_ekb_refresh crashed run_id=%d", run_id)
finally:
run_db.close()
_spawn_tracked(_run())
logger.info("scheduler: triggered osm_poi_ekb_refresh run_id=%d", run_id)
return run_id
async def trigger_house_imv_backfill_run(db: Session, schedule_row: dict[str, Any]) -> int | None:
"""Создать scrape_runs + launch backfill_house_imv в asyncio.create_task (#854).
@ -1955,6 +2003,8 @@ async def scheduler_loop() -> None:
await trigger_yandex_detail_backfill_run(db, sch)
elif source == "cadastral_geo_match":
await trigger_cadastral_geo_match_run(db, sch)
elif source == "osm_poi_ekb_refresh":
await trigger_osm_poi_ekb_refresh_run(db, sch)
elif source == "house_imv_backfill":
await trigger_house_imv_backfill_run(db, sch)
elif source == "house_dedup_merge":

View file

@ -0,0 +1,109 @@
"""OSM POI local-mirror refresh (#2045 BE-3, LocationDrawer location-coef).
Populates `osm_poi_ekb_local` (empty at deploy, migration 169) via a single bulk scan of the
`gendesign_osm_poi_ekb` FDW foreign table (migration 168 live view of gendesign's
`osm_poi_ekb` table, Site Finder's OSM POI registry, GRANTed to `tradein_fdw_reader` in PR-A
of #2045, already merged + deployed on gendesign).
WHY a local mirror (perf fact measured for the analogous gendesign_cad_buildings FDW see
`app/tasks/cadastral_geo_match.py`): a per-request FDW nearest-POI query pays a per-row FDW
round-trip (~1.16s/row without a geom index on the remote table) UNUSABLE for a synchronous
endpoint (`GET /api/v1/trade-in/location-coef`). We materialize the FDW once (single bulk
scan) into `osm_poi_ekb_local` with a real Point geom + GIST index, then
`app/services/location_coef.py` runs fast LOCAL ST_DWithin/ST_Distance queries per estimate.
Scheduler source='osm_poi_ekb_refresh' (daily OSM POI data changes rarely). Pure internal
DB op one FDW read + local TRUNCATE+INSERT, no HTTP/anti-bot.
psycopg v3: app SQL uses CAST(:x AS type), never :x::type.
"""
from __future__ import annotations
import logging
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__)
def refresh_osm_poi_ekb_local(db: Session) -> int:
"""TRUNCATE osm_poi_ekb_local; bulk INSERT from gendesign_osm_poi_ekb FDW.
Single FDW scan (no per-row round-trips). Builds a 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 osm_poi_ekb_local"))
db.execute(
text(
"""
INSERT INTO osm_poi_ekb_local (category, name, lat, lon, geom)
SELECT
category, name, lat, lon,
ST_SetSRID(ST_MakePoint(lon, lat), 4326)
FROM gendesign_osm_poi_ekb
WHERE lat IS NOT NULL
AND lon IS NOT NULL
"""
)
)
count = db.execute(text("SELECT count(*) FROM osm_poi_ekb_local")).scalar() or 0
db.commit()
logger.info("refresh_osm_poi_ekb_local: loaded %d rows from FDW", count)
return int(count)
@dataclass
class OsmPoiRefreshResult:
poi_loaded: int = 0 # rows in osm_poi_ekb_local after refresh
duration_sec: float = field(default=0.0)
def to_counters(self) -> dict[str, int]:
return {
"poi_loaded": self.poi_loaded,
"duration_sec": int(self.duration_sec),
}
def run_osm_poi_ekb_refresh(db: Session, *, run_id: int, params: dict) -> OsmPoiRefreshResult:
"""Run-lifecycle wrapper for the OSM POI local-mirror refresh (scheduler entrypoint).
Launched by the in-app scheduler (source='osm_poi_ekb_refresh') via
trigger_osm_poi_ekb_refresh_run, or manually. No params consumed single unconditional
bulk refresh (`params` accepted for scheduler-signature compatibility / future use).
Finalises scrape_runs (mark_done / mark_failed) with counters.
"""
counters: dict[str, int] = {"poi_loaded": 0}
start = time.monotonic()
try:
runs_mod.update_heartbeat(db, run_id, counters)
loaded = refresh_osm_poi_ekb_local(db)
result = OsmPoiRefreshResult(poi_loaded=loaded, duration_sec=time.monotonic() - start)
counters = result.to_counters()
runs_mod.mark_done(db, run_id, counters)
logger.info(
"run_osm_poi_ekb_refresh: run_id=%d DONE poi_loaded=%d duration=%.1fs",
run_id,
result.poi_loaded,
result.duration_sec,
)
return result
except Exception as exc:
logger.exception("run_osm_poi_ekb_refresh: 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,36 @@
-- 168_fdw_osm_poi_ekb.sql
-- FDW foreign table bridging to gendesign's v_tradein_osm_poi_ekb view for
-- location-coef / POI scoring in LocationDrawer (#2045, BE-3 final PR).
--
-- Mirrors 060_postgres_fdw_extension.sql conventions: uses the SAME SERVER
-- gendesign_remote (already created there) — this migration does NOT create a new
-- SERVER / USER MAPPING.
--
-- DEPENDENCIES:
-- - 060_postgres_fdw_extension.sql (SERVER gendesign_remote + USER MAPPING, managed by
-- backend startup with env password).
-- - gendesign: view public.v_tradein_osm_poi_ekb + GRANT SELECT to tradein_fdw_reader
-- (PR-A of #2045, already merged + deployed on gendesign).
--
-- Idempotent: DROP FOREIGN TABLE IF EXISTS + re-CREATE.
BEGIN;
DROP FOREIGN TABLE IF EXISTS gendesign_osm_poi_ekb;
CREATE FOREIGN TABLE gendesign_osm_poi_ekb (
category text,
name text,
lat double precision,
lon double precision
)
SERVER gendesign_remote
OPTIONS (schema_name 'public', table_name 'v_tradein_osm_poi_ekb');
COMMENT ON FOREIGN TABLE gendesign_osm_poi_ekb IS
'Live view of gendesign.osm_poi_ekb (Site Finder OSM POI, EKB). USER MAPPING managed by '
'backend startup (see 060). Materialized locally by the osm_poi_ekb_refresh scheduler job '
'into osm_poi_ekb_local (migration 169) — see app/tasks/osm_poi_ekb_refresh.py. '
'Feeds app/services/location_coef.py (GET /api/v1/trade-in/location-coef).';
COMMIT;

View file

@ -0,0 +1,42 @@
-- 169_osm_poi_ekb_local.sql
-- Local materialized mirror of gendesign_osm_poi_ekb (postgres_fdw foreign table, migration
-- 168) for fast location-coef / POI scoring (#2045 BE-3, LocationDrawer).
--
-- WHY (mirrors 124_cad_buildings_local.sql — same perf fact measured for the analogous
-- gendesign_cad_buildings FDW): a per-request FDW nearest-POI query would pay a per-row FDW
-- round-trip (~1.16s/row without a geom index on the remote table) — UNUSABLE for a
-- synchronous endpoint (GET /api/v1/trade-in/location-coef). Solution: materialize the FDW
-- once (single bulk scan) into a LOCAL table with a real Point geom + GIST index, then run
-- fast local ST_DWithin/ST_Distance queries per estimate.
--
-- This migration creates the EMPTY table + index only. It does NOT read the FDW — deploy
-- never depends on the gendesign DB / FDW server being reachable. The table is populated by
-- the refresh job (app/tasks/osm_poi_ekb_refresh.py), driven by the in-app scheduler
-- (source='osm_poi_ekb_refresh', seed in 170_scrape_schedules_seed_osm_poi_ekb_refresh.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 osm_poi_ekb_local (
id bigserial PRIMARY KEY,
category text,
name text,
lat double precision,
lon double precision,
geom geometry(Point, 4326)
);
COMMENT ON TABLE osm_poi_ekb_local IS
'Local mirror of gendesign_osm_poi_ekb FDW (Site Finder OSM POI registry, EKB). '
'Populated by app/tasks/osm_poi_ekb_refresh.py (single bulk FDW scan, TRUNCATE+INSERT). '
'Point geom + GIST enables fast local nearest-POI scoring for '
'app/services/location_coef.py (GET /api/v1/trade-in/location-coef, LocationDrawer). '
'Migration 168/169. Refresh driven by scheduler source=osm_poi_ekb_refresh.';
-- GIST on geom — required for ST_DWithin / ST_Distance nearest-POI scoring.
CREATE INDEX IF NOT EXISTS osm_poi_ekb_local_geom_idx
ON osm_poi_ekb_local USING GIST (geom);
COMMIT;

View file

@ -0,0 +1,46 @@
-- 170_scrape_schedules_seed_osm_poi_ekb_refresh.sql
-- Scheduler seed for the OSM POI local-mirror refresh job (#2045 BE-3, LocationDrawer).
--
-- WHAT (source='osm_poi_ekb_refresh'):
-- trigger_osm_poi_ekb_refresh_run (scheduler.py) → run_osm_poi_ekb_refresh
-- (tasks/osm_poi_ekb_refresh.py). Single bulk FDW scan: TRUNCATE osm_poi_ekb_local;
-- INSERT ... SELECT FROM gendesign_osm_poi_ekb, building a Point(4326) geom from lat/lon.
--
-- OSM POI data changes rarely (Site Finder refreshes its own osm_poi_ekb table
-- infrequently) — a daily cadence is more than sufficient freshness margin at near-zero
-- cost (pure internal DB op, no external HTTP/anti-bot).
--
-- SAFE to enable by default (enabled=true), like cadastral_geo_match (seed 125).
--
-- Window 10:00-11:00 UTC (13:00-14:00 MSK): after cadastral_geo_match (09:00-10:00 UTC);
-- pure-DB so no contention with scraper/proxy work either way.
--
-- next_run_at = tomorrow 10:00 UTC (NOT NULL — avoid deploy-time fire, same convention as
-- the cadastral_geo_match seed 125; NULL would make get_due_schedules pick it up on the
-- very next tick).
--
-- DEPENDENCIES: 052_scrape_schedules.sql (table + UNIQUE(source)), 169_osm_poi_ekb_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
(
'osm_poi_ekb_refresh',
true,
10,
11,
((CURRENT_DATE + INTERVAL '1 day') + make_interval(hours => 10)) AT TIME ZONE 'UTC',
'{}'::jsonb
)
ON CONFLICT (source) DO NOTHING;
COMMIT;

View file

@ -172,3 +172,6 @@
165_remove_n1_source.sql
166_purge_listings_phones.sql
167_drop_client_pii.sql
168_fdw_osm_poi_ekb.sql
169_osm_poi_ekb_local.sql
170_scrape_schedules_seed_osm_poi_ekb_refresh.sql

View file

@ -0,0 +1,178 @@
"""Unit tests for app.services.location_coef (#2045 BE-3, LocationDrawer).
No live Postgres needed DB is a minimal fake returning queued results (mirrors the
convention in tests/tasks/test_cadastral_geo_match.py). Covers:
- pure functions: _category_weight, _score_to_coef, normalization constants
- compute_location_coef: weighted top-N scoring, empty-mirror graceful fallback,
no-POI-in-radius (legit zero-score, NOT "unavailable")
"""
from __future__ import annotations
import os
from typing import Any
# psycopg v3 driver required; stub DATABASE_URL before any app import (settings needs a DSN).
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
from app.services import location_coef as lc
# ── Pure functions ──────────────────────────────────────────────────────────
def test_category_weight_known_categories() -> None:
assert lc._category_weight("metro_stop") == 6.0
assert lc._category_weight("school") == 5.0
assert lc._category_weight("kindergarten") == 4.5
assert lc._category_weight("hospital") == 4.0
assert lc._category_weight("shop_mall") == 4.0
assert lc._category_weight("shop_supermarket") == 3.5
assert lc._category_weight("bus_stop") == 4.5
assert lc._category_weight("park") == 3.5
assert lc._category_weight("pharmacy") == 2.5
assert lc._category_weight("tram_stop") == 2.0
assert lc._category_weight("shop_small") == 2.0
def test_category_weight_unknown_and_none_fall_back_to_default() -> None:
assert lc._category_weight("unknown_category") == 1.0
assert lc._category_weight(None) == 1.0
def test_top7_weight_sum_matches_ptica() -> None:
"""Same category set as Site Finder → identical top-7 normalization constant (31.5)."""
assert lc._TOP7_WEIGHT_SUM == 31.5
assert abs(lc._MAX_STRAIGHT_SCORE - 0.315) < 1e-9
def test_score_to_coef_bounds() -> None:
assert lc._score_to_coef(0.0) == 0.95
assert lc._score_to_coef(100.0) == 1.05
def test_score_to_coef_midpoint() -> None:
assert lc._score_to_coef(50.0) == 1.0
def test_score_to_coef_is_monotonic() -> None:
scores = [0.0, 10.0, 25.0, 50.0, 75.0, 90.0, 100.0]
coefs = [lc._score_to_coef(s) for s in scores]
assert coefs == sorted(coefs)
# ── compute_location_coef with a fake DB ─────────────────────────────────────
class _FakeResult:
def __init__(self, *, scalar_value: Any = None, mapping_rows: list[dict] | None = None):
self._scalar_value = scalar_value
self._mapping_rows = mapping_rows or []
def scalar(self) -> Any:
return self._scalar_value
def mappings(self) -> Any:
class _Mappings:
def __init__(self, rows: list[dict]) -> None:
self._rows = rows
def all(self) -> list[dict]:
return self._rows
return _Mappings(self._mapping_rows)
class _FakeDB:
"""Minimal Session stand-in: execute() returns queued results in order."""
def __init__(self, results: list[_FakeResult]) -> None:
self._results = list(results)
self.executed: list[Any] = []
def execute(self, clause: Any, params: dict | None = None) -> _FakeResult:
self.executed.append((clause, params))
return self._results.pop(0)
def test_compute_location_coef_empty_mirror_returns_unavailable() -> None:
"""osm_poi_ekb_local not yet refreshed (count=0) → unavailable, no fabricated factors."""
db = _FakeDB([_FakeResult(scalar_value=0)])
result = lc.compute_location_coef(db, lat=56.84, lon=60.6)
assert result.coef == 1.0
assert result.factors == []
assert result.geo_source == "unavailable"
# Only the count probe ran — no nearest-POI query issued against an empty mirror.
assert len(db.executed) == 1
def test_compute_location_coef_no_poi_in_radius_is_legit_zero_score() -> None:
"""Mirror populated (count>0) but nothing within radius → coef floor, NOT unavailable."""
db = _FakeDB(
[
_FakeResult(scalar_value=500), # mirror has rows elsewhere
_FakeResult(mapping_rows=[]), # nothing near this point
]
)
result = lc.compute_location_coef(db, lat=56.84, lon=60.6)
assert result.factors == []
assert result.geo_source == "osm_poi_ekb"
assert result.coef == lc._score_to_coef(0.0) == 0.95
def test_compute_location_coef_weights_and_ranks_top_n() -> None:
"""Nearer + higher-weight-category POI ranks above farther/lower-weight ones."""
rows = [
{"name": "Школа №1", "category": "school", "distance_m": 300.0},
{"name": "ТЦ Мега", "category": "shop_mall", "distance_m": 900.0},
{"name": "Метро Ботаническая", "category": "metro_stop", "distance_m": 150.0},
{"name": "Аптека", "category": "pharmacy", "distance_m": 50.0},
]
db = _FakeDB([_FakeResult(scalar_value=1000), _FakeResult(mapping_rows=rows)])
result = lc.compute_location_coef(db, lat=56.84, lon=60.6, top_n=7)
assert result.geo_source == "osm_poi_ekb"
assert len(result.factors) == 4
# metro_stop (weight 6.0) at 150m beats school (5.0) at 300m despite being closer only
# marginally — sanity check the ranking is weight-driven, not distance-only.
assert result.factors[0].poi_type == "metro_stop"
# Weights strictly descending (sorted DESC by weight before slicing to top_n).
weights = [f.weight for f in result.factors]
assert weights == sorted(weights, reverse=True)
# coef must land inside the documented [0.95, 1.05] MVP range.
assert 0.95 <= result.coef <= 1.05
def test_compute_location_coef_limits_to_top_n() -> None:
"""More than top_n candidates → only top_n factors surface in the response."""
rows = [
{"name": f"POI {i}", "category": "shop_small", "distance_m": float(100 + i * 10)}
for i in range(20)
]
db = _FakeDB([_FakeResult(scalar_value=20), _FakeResult(mapping_rows=rows)])
result = lc.compute_location_coef(db, lat=56.84, lon=60.6, top_n=7)
assert len(result.factors) == 7
def test_compute_location_coef_unknown_category_uses_default_weight() -> None:
rows = [{"name": "Неизвестный POI", "category": "some_new_osm_tag", "distance_m": 200.0}]
db = _FakeDB([_FakeResult(scalar_value=1), _FakeResult(mapping_rows=rows)])
result = lc.compute_location_coef(db, lat=56.84, lon=60.6)
assert len(result.factors) == 1
expected_weight = (1.0 / (200.0 + 100.0)) * lc.CATEGORY_WEIGHTS["default"]
assert abs(result.factors[0].weight - round(expected_weight, 6)) < 1e-9
def test_compute_location_coef_passes_radius_param() -> None:
"""radius_m is forwarded as a bound param (psycopg v3 CAST discipline, no :p::type)."""
db = _FakeDB([_FakeResult(scalar_value=1), _FakeResult(mapping_rows=[])])
lc.compute_location_coef(db, lat=56.84, lon=60.6, radius_m=1500)
_clause, params = db.executed[1]
assert params is not None
assert params["radius_m"] == 1500
def test_no_psycopg_v3_colon_colon_cast() -> None:
"""psycopg v3: never :param::type — must use CAST(:param AS type)."""
import re
assert not re.search(r":\w+::", str(lc._NEAREST_POI_SQL.text))

View file

@ -0,0 +1,251 @@
"""Tests for GET /api/v1/trade-in/location-coef (#2045 BE-3, LocationDrawer).
Mirrors the IDOR + mocked-DB conventions of test_estimate_idor.py / test_street_deals_endpoint.py
(no live Postgres needed). Covers:
- IDOR guard (owner / other pilot 404 / admin / unauthenticated 401)
- graceful fallback: osm_poi_ekb_local empty/not-refreshed, estimate has no lat/lon
- happy path: base_price_rub * coef result_price_rub, factors surfaced
- radius_m query-param clamping
"""
from __future__ import annotations
import os
import sys
from types import SimpleNamespace
from unittest.mock import MagicMock
# psycopg v3 driver required; stub DATABASE_URL before any app import.
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
# WeasyPrint requires GTK — not present in CI/Windows. Stub before any app import
# (trade_in.py imports generate_trade_in_pdf at module load).
_wp_mock = MagicMock()
sys.modules.setdefault("weasyprint", _wp_mock)
sys.modules.setdefault("weasyprint.CSS", _wp_mock)
sys.modules.setdefault("weasyprint.HTML", _wp_mock)
import pytest # noqa: E402
from fastapi import FastAPI # noqa: E402
from fastapi.testclient import TestClient # noqa: E402
_ESTIMATE_ID = "22222222-2222-2222-2222-222222222222"
@pytest.fixture(autouse=True)
def _restore_get_role():
"""Restore app.core.auth.get_role after each test (mirror test_estimate_idor)."""
from app.core import auth as auth_mod
original = auth_mod.get_role
yield
auth_mod.get_role = original
@pytest.fixture()
def trade_in_app() -> FastAPI:
"""Minimal FastAPI app mounting only the trade-in router."""
from app.api.v1 import trade_in as trade_in_module
application = FastAPI()
application.include_router(trade_in_module.router, prefix="/api/v1/trade-in")
return application
def _client_with(app: FastAPI, db_mock: MagicMock, role: str | None) -> TestClient:
"""Override get_db with *db_mock*; patch get_role to return *role* (or raise KeyError)."""
from app.core.db import get_db
def _override_db():
yield db_mock
app.dependency_overrides[get_db] = _override_db
auth_mod = sys.modules["app.core.auth"]
if role is None:
def _raise_keyerror(_u: str):
raise KeyError(_u)
auth_mod.get_role = _raise_keyerror # type: ignore[assignment]
else:
auth_mod.get_role = lambda _u: role # type: ignore[assignment]
return TestClient(app)
def _fetchone_result(row: object) -> MagicMock:
r = MagicMock()
r.fetchone.return_value = row
return r
def _scalar_result(value: object) -> MagicMock:
r = MagicMock()
r.scalar.return_value = value
return r
def _mapping_result(rows: list[dict]) -> MagicMock:
r = MagicMock()
r.mappings.return_value.all.return_value = rows
return r
def _db_with(*results: MagicMock) -> MagicMock:
db = MagicMock()
db.execute.side_effect = list(results)
return db
# ── IDOR guard ────────────────────────────────────────────────────────────────
def test_location_coef_other_pilot_gets_404(trade_in_app: FastAPI) -> None:
"""Non-owner pilot must NOT read someone else's location-coef → 404."""
db = _db_with(_fetchone_result(SimpleNamespace(created_by="victim")))
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "attacker"},
)
assert resp.status_code == 404
def test_location_coef_requires_authenticated_user(trade_in_app: FastAPI) -> None:
"""No X-Authenticated-User header → 401 (guard query already ran, header check fails)."""
db = _db_with(_fetchone_result(SimpleNamespace(created_by="kopylov")))
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
)
assert resp.status_code == 401
def test_location_coef_unknown_estimate_returns_404(trade_in_app: FastAPI) -> None:
db = _db_with(_fetchone_result(None))
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 404
def test_location_coef_admin_can_read_any(trade_in_app: FastAPI) -> None:
"""Admin reads any estimate's location-coef regardless of owner → 200."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="someone_else")), # guard
_fetchone_result(SimpleNamespace(lat=None, lon=None, median_price=5_000_000)), # target
)
client = _client_with(trade_in_app, db, role="admin")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "admin"},
)
assert resp.status_code == 200
# ── Graceful fallback ───────────────────────────────────────────────────────
def test_location_coef_no_lat_lon_returns_unavailable(trade_in_app: FastAPI) -> None:
"""Estimate without lat/lon → unavailable fallback, no osm_poi_ekb_local query at all."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=None, lon=None, median_price=5_000_000)), # target
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
body = resp.json()
assert body["geo_source"] == "unavailable"
assert body["coef"] == 1.0
assert body["factors"] == []
assert body["base_price_rub"] == 5_000_000
assert body["result_price_rub"] == 5_000_000
# Short-circuits before compute_location_coef: only guard + target queries ran.
assert db.execute.call_count == 2
def test_location_coef_empty_mirror_returns_unavailable(trade_in_app: FastAPI) -> None:
"""osm_poi_ekb_local not yet refreshed (count=0) → unavailable, no fabricated factors."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=56.84, lon=60.6, median_price=5_000_000)), # target
_scalar_result(0), # osm_poi_ekb_local count
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
body = resp.json()
assert body["geo_source"] == "unavailable"
assert body["coef"] == 1.0
assert body["factors"] == []
assert body["result_price_rub"] == body["base_price_rub"]
# ── Happy path ────────────────────────────────────────────────────────────────
def test_location_coef_happy_path_computes_result_price(trade_in_app: FastAPI) -> None:
"""POI found within radius → coef applied to base_price_rub, factors surfaced."""
poi_rows = [
{"name": "Школа №1", "category": "school", "distance_m": 300.0},
{"name": "Метро Ботаническая", "category": "metro_stop", "distance_m": 150.0},
]
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=56.84, lon=60.6, median_price=5_000_000)), # target
_scalar_result(1000), # osm_poi_ekb_local count
_mapping_result(poi_rows), # nearest POI
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
body = resp.json()
assert body["geo_source"] == "osm_poi_ekb"
assert body["base_price_rub"] == 5_000_000
assert 0.95 <= body["coef"] <= 1.05
assert body["result_price_rub"] == round(5_000_000 * body["coef"])
assert len(body["factors"]) == 2
poi_types = {f["poi_type"] for f in body["factors"]}
assert poi_types == {"school", "metro_stop"}
def test_location_coef_radius_m_clamped_to_bounds(trade_in_app: FastAPI) -> None:
"""Explicit radius_m outside [500, 3000] is clamped before hitting the DB query."""
db = _db_with(
_fetchone_result(SimpleNamespace(created_by="kopylov")), # guard
_fetchone_result(SimpleNamespace(lat=56.84, lon=60.6, median_price=5_000_000)), # target
_scalar_result(10), # osm_poi_ekb_local count
_mapping_result([]), # nearest POI query
)
client = _client_with(trade_in_app, db, role="pilot")
resp = client.get(
"/api/v1/trade-in/location-coef",
params={"estimate_id": _ESTIMATE_ID, "radius_m": 50},
headers={"X-Authenticated-User": "kopylov"},
)
assert resp.status_code == 200
# 4th execute() call is the nearest-POI query — inspect its bound radius_m param.
nearest_call = db.execute.call_args_list[3]
params = (
nearest_call.args[1] if len(nearest_call.args) > 1 else nearest_call.kwargs.get("params")
)
assert params["radius_m"] == 500