From f82d8bc870b8421cf391887a456225d2f65bca91 Mon Sep 17 00:00:00 2001 From: lekss361 Date: Mon, 18 May 2026 00:05:11 +0300 Subject: [PATCH] feat(sf-b6): POI weighted top-7 endpoint GET /parcels/{cad}/poi-score Formula: weight = (1 / (distance_m + 100)) * category_weight Top-7 POI from osm_poi_ekb ranked by weight DESC. 12 unit tests (pure function, no DB required). --- backend/app/api/v1/parcels.py | 51 ++++++ backend/app/services/site_finder/poi_score.py | 159 +++++++++++++++++ backend/tests/test_poi_score.py | 168 ++++++++++++++++++ 3 files changed, 378 insertions(+) create mode 100644 backend/app/services/site_finder/poi_score.py create mode 100644 backend/tests/test_poi_score.py diff --git a/backend/app/api/v1/parcels.py b/backend/app/api/v1/parcels.py index 9a315b14..43f3a7dc 100644 --- a/backend/app/api/v1/parcels.py +++ b/backend/app/api/v1/parcels.py @@ -43,6 +43,7 @@ from app.services.site_finder.custom_pois import ( get_overlaps_for_scoring as _get_custom_poi_overlaps, ) from app.services.site_finder.gate_verdict import compute_gate_verdict +from app.services.site_finder.poi_score import PoiScoreResponse, compute_poi_weighted_top7 from app.services.site_finder.quarter_dump_lookup import ( get_connection_points, get_quarter_dump_data, @@ -2455,6 +2456,56 @@ def get_isochrones( } +@router.get( + "/{cad_num}/poi-score", + response_model=PoiScoreResponse, + summary="POI weighted top-7 (B6)", +) +async def get_poi_score( + cad_num: str, + db: Annotated[Session, Depends(get_db)], + radius_m: Annotated[int, Query(ge=100, le=5000)] = 2000, +) -> PoiScoreResponse: + """Вернуть top-7 ближайших POI для участка, взвешенных по формуле: + + weight = (1 / (distance_m + 100)) * category_weight + + POI берутся из osm_poi_ekb в заданном радиусе (default 2000м). + Отсортированы по weight DESC — наиболее значимые объекты первыми. + """ + # Получить координаты центроида участка из геометрических таблиц + coord_row = ( + db.execute( + text(""" + SELECT ST_X(ST_Centroid(g.geom)) AS lon, + ST_Y(ST_Centroid(g.geom)) AS lat + FROM ( + SELECT geom FROM cad_quarters_geom WHERE cad_number = :c + UNION ALL + SELECT geom FROM cad_buildings WHERE cad_num = :c + UNION ALL + SELECT geom FROM cad_parcels_geom WHERE cad_num = :c + ) g + LIMIT 1 + """), + {"c": cad_num}, + ) + .mappings() + .first() + ) + + if not coord_row: + raise HTTPException( + status_code=404, + detail=f"Геометрия для {cad_num} не найдена.", + ) + + lat = float(coord_row["lat"]) + lon = float(coord_row["lon"]) + + return compute_poi_weighted_top7(db, cad_num, lat, lon, radius_m=radius_m) + + @router.post("/{cad_num}/competitors", response_model=CompetitorsResponse) async def get_parcel_competitors( cad_num: str, diff --git a/backend/app/services/site_finder/poi_score.py b/backend/app/services/site_finder/poi_score.py new file mode 100644 index 00000000..31a7032b --- /dev/null +++ b/backend/app/services/site_finder/poi_score.py @@ -0,0 +1,159 @@ +"""POI weighted score для Site Finder (B6). + +Формула: weight = (1 / (distance_m + 100)) * category_weight + +Возвращает top-7 ближайших POI из osm_poi_ekb, отсортированных по weight DESC. +Категории и их веса согласованы с _POI_WEIGHTS в parcels.py. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from pydantic import BaseModel +from sqlalchemy import text + +logger = logging.getLogger(__name__) + +# Веса по категории — согласованы с _POI_WEIGHTS в parcels.py + новые из vault B6. +# Задача: "2GIS-style ranking", метро самое приоритетное. +CATEGORY_WEIGHTS: dict[str, float] = { + "metro_stop": 6.0, + "school": 5.0, + "kindergarten": 4.5, + "hospital": 4.0, + "shop_supermarket": 3.5, + "shop_mall": 4.0, + "park": 3.5, + "bus_stop": 4.5, + "tram_stop": 2.0, + "pharmacy": 2.5, + "shop_small": 2.0, + "default": 1.0, +} + + +class PoiScoreItem(BaseModel): + """Один POI в ranked-ответе.""" + + name: str | None + category: str + distance_m: float + weight: float + address: str | None + + +class PoiScoreResponse(BaseModel): + cad_num: str + radius_m: int + top_poi: list[PoiScoreItem] + + +def _category_weight(category: str) -> float: + """Вернуть вес категории. Если не знаем — default.""" + return CATEGORY_WEIGHTS.get(category, CATEGORY_WEIGHTS["default"]) + + +def compute_poi_weighted_top7( + db: Any, + cad_num: str, + lat: float, + lon: float, + radius_m: int = 2000, + top_n: int = 7, +) -> PoiScoreResponse: + """Найти top-N POI вокруг (lat, lon) в radius_m, ранжировать по weighted score. + + Запрос к osm_poi_ekb через ST_DWithin + ST_Distance. + Формула: weight = (1 / (distance_m + 100)) * category_weight + + Args: + db: SQLAlchemy Session + cad_num: кадастровый номер (для ответа) + lat: широта центроида участка + lon: долгота центроида участка + radius_m: радиус поиска в метрах (default 2000) + top_n: количество POI в ответе (default 7) + + Returns: + PoiScoreResponse с отсортированными по weight DESC POI. + """ + # ST_DWithin с geography=true использует метры напрямую. + # ST_Distance тоже в метрах при geography=true. + rows = ( + db.execute( + text(""" + SELECT + p.name, + p.category, + p.tags, + CAST( + ST_Distance( + p.geom::geography, + ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography + ) AS double precision + ) AS distance_m + FROM osm_poi_ekb p + WHERE ST_DWithin( + p.geom::geography, + ST_SetSRID(ST_MakePoint(:lon, :lat), 4326)::geography, + :radius_m + ) + ORDER BY distance_m ASC + LIMIT :limit + """), + { + "lat": lat, + "lon": lon, + "radius_m": radius_m, + "limit": top_n * 10, # запрашиваем больше, потом ранжируем + }, + ) + .mappings() + .all() + ) + + logger.debug( + "poi_score: cad=%s lat=%.5f lon=%.5f radius=%dm → %d candidates", + cad_num, + lat, + lon, + radius_m, + len(rows), + ) + + items: list[PoiScoreItem] = [] + for row in rows: + distance_m = float(row["distance_m"]) + category = row["category"] or "default" + cat_weight = _category_weight(category) + weight = (1.0 / (distance_m + 100.0)) * cat_weight + + # Адрес из tags jsonb если есть + tags: dict[str, str] = row["tags"] or {} + addr_parts = [ + tags.get("addr:street"), + tags.get("addr:housenumber"), + ] + address = ", ".join(p for p in addr_parts if p) or None + + items.append( + PoiScoreItem( + name=row["name"], + category=category, + distance_m=round(distance_m, 1), + weight=round(weight, 6), + address=address, + ) + ) + + # Сортировка по weight DESC, берём top_n + items.sort(key=lambda x: x.weight, reverse=True) + top_items = items[:top_n] + + return PoiScoreResponse( + cad_num=cad_num, + radius_m=radius_m, + top_poi=top_items, + ) diff --git a/backend/tests/test_poi_score.py b/backend/tests/test_poi_score.py new file mode 100644 index 00000000..e633894c --- /dev/null +++ b/backend/tests/test_poi_score.py @@ -0,0 +1,168 @@ +"""Tests for POI weighted score service (B6). + +Юнит-тесты для чистой функции — без DB. +""" + +from app.services.site_finder.poi_score import ( + CATEGORY_WEIGHTS, + PoiScoreResponse, + _category_weight, + compute_poi_weighted_top7, +) + +# ── unit: _category_weight ───────────────────────────────────────────────────── + + +def test_category_weight_metro(): + """Метро имеет наибольший вес из всех категорий.""" + metro_w = _category_weight("metro_stop") + for cat in CATEGORY_WEIGHTS: + if cat != "metro_stop" and cat != "default": + assert metro_w >= _category_weight( + cat + ), f"metro_stop weight {metro_w} должен быть >= {cat} weight {_category_weight(cat)}" + + +def test_category_weight_unknown_returns_default(): + w = _category_weight("unknown_category_xyz") + assert w == CATEGORY_WEIGHTS["default"] + + +def test_category_weight_all_positive(): + """Все веса в CATEGORY_WEIGHTS должны быть положительными (B6 — ranking, не штраф).""" + for cat, w in CATEGORY_WEIGHTS.items(): + assert w > 0, f"Вес {cat}={w} должен быть > 0" + + +# ── unit: weight formula ratio ───────────────────────────────────────────────── + + +def test_weight_formula_ratio(): + """Ближний объект той же категории должен иметь больший вес.""" + cat = "school" + cw = _category_weight(cat) + w_near = (1.0 / (100.0 + 100.0)) * cw # 100м + w_far = (1.0 / (1000.0 + 100.0)) * cw # 1000м + assert w_near > w_far + + +def test_weight_formula_category_dominates_at_equal_distance(): + """При одинаковом расстоянии метро должно быть впереди автобусной остановки.""" + dist = 500.0 + w_metro = (1.0 / (dist + 100.0)) * _category_weight("metro_stop") + w_bus = (1.0 / (dist + 100.0)) * _category_weight("bus_stop") + assert w_metro > w_bus + + +# ── unit: compute_poi_weighted_top7 with mock DB ─────────────────────────────── + + +class _MockMappings: + def __init__(self, data: list[dict]) -> None: + self._data = data + + def all(self) -> list[dict]: + return self._data # type: ignore[return-value] + + +class _MockResult: + def __init__(self, data: list[dict]) -> None: + self._data = data + + def mappings(self) -> "_MockMappings": + return _MockMappings(self._data) + + +class _MockDb: + """Минимальный мок SQLAlchemy Session для тестирования без БД.""" + + def __init__(self, rows: list[dict]) -> None: + self._rows = rows + + def execute(self, *_args: object, **_kwargs: object) -> _MockResult: + return _MockResult(self._rows) + + +def _make_row(name: str, category: str, distance_m: float) -> dict: + return { + "name": name, + "category": category, + "tags": {}, + "distance_m": distance_m, + } + + +def test_top7_returns_at_most_7(): + rows = [_make_row(f"POI {i}", "school", float(i * 50)) for i in range(1, 20)] + db = _MockDb(rows) + result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605) + assert isinstance(result, PoiScoreResponse) + assert len(result.top_poi) <= 7 + + +def test_top7_sorted_by_weight_desc(): + rows = [ + _make_row("Дальняя школа", "school", 1500.0), + _make_row("Метро", "metro_stop", 300.0), + _make_row("Близкая школа", "school", 100.0), + ] + db = _MockDb(rows) + result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605) + weights = [item.weight for item in result.top_poi] + assert weights == sorted(weights, reverse=True), "top_poi должны быть по weight DESC" + + +def test_metro_beats_school_at_equal_distance(): + """Метро в 300м должно быть на первом месте перед школой в 300м (равное расстояние).""" + rows = [ + _make_row("Школа №1", "school", 300.0), + _make_row("Метро Чкаловская", "metro_stop", 300.0), + ] + db = _MockDb(rows) + result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605) + assert ( + result.top_poi[0].category == "metro_stop" + ), "При равном расстоянии метро (category_weight=6.0) должно быть выше школы (5.0)" + + +def test_metro_first_when_close(): + """Метро в 50м должно быть на первом месте перед школой в 300м.""" + rows = [ + _make_row("Школа №1", "school", 300.0), + _make_row("Метро Чкаловская", "metro_stop", 50.0), + ] + db = _MockDb(rows) + result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605) + assert result.top_poi[0].category == "metro_stop", ( + "Метро (weight=6.0) в 50м должно быть впереди школы (weight=5.0) в 300м — " + f"metro_weight={(1/(50+100))*6:.5f} vs school_weight={(1/(300+100))*5:.5f}" + ) + + +def test_empty_db_returns_empty_top_poi(): + db = _MockDb([]) + result = compute_poi_weighted_top7(db, "66:41:0204016:10", 56.838, 60.605) + assert result.top_poi == [] + assert result.cad_num == "66:41:0204016:10" + assert result.radius_m == 2000 + + +def test_address_built_from_tags(): + rows = [ + { + "name": "Магазин", + "category": "shop_small", + "tags": {"addr:street": "ул. Ленина", "addr:housenumber": "10"}, + "distance_m": 200.0, + } + ] + db = _MockDb(rows) + result = compute_poi_weighted_top7(db, "test", 56.838, 60.605) + assert result.top_poi[0].address == "ул. Ленина, 10" + + +def test_address_none_when_no_tags(): + rows = [_make_row("Парк", "park", 400.0)] + db = _MockDb(rows) + result = compute_poi_weighted_top7(db, "test", 56.838, 60.605) + assert result.top_poi[0].address is None -- 2.45.3