diff --git a/backend/app/api/v1/parcels.py b/backend/app/api/v1/parcels.py
index 9a315b14..f8c407d3 100644
--- a/backend/app/api/v1/parcels.py
+++ b/backend/app/api/v1/parcels.py
@@ -30,6 +30,7 @@ from app.schemas.parcel import (
RiskZone,
)
from app.services.exporters.layout_tz_pdf import render_layout_tz_pdf
+from app.services.exporters.snapshot_pdf import generate_snapshot_pdf
from app.services.site_finder.best_layouts import get_best_layouts
from app.services.site_finder.cadastre_fetch import (
cad_exists_in_db,
@@ -43,6 +44,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,
@@ -1701,6 +1703,217 @@ def analyze_parcel(
except Exception as e:
logger.warning("parcel_meta query failed for %s: %s", cad_num, e)
+ # B5-1) EGRN block — расширенные данные из cad_parcels (SF-B5)
+ egrn_block: dict[str, Any] = {}
+ try:
+ egrn_row = (
+ db.execute(
+ text("""
+ SELECT cost_value AS cadastral_value_rub,
+ cost_index AS cost_index_per_m2,
+ land_record_category_type AS land_category,
+ permitted_use_established_by_document AS permitted_use_text,
+ cost_registration_date AS last_egrn_update_date,
+ land_record_area AS area_m2,
+ ownership_type,
+ right_type,
+ status,
+ readable_address,
+ registration_date
+ FROM cad_parcels
+ WHERE cad_num = CAST(:c AS text)
+ LIMIT 1
+ """),
+ {"c": cad_num},
+ )
+ .mappings()
+ .first()
+ )
+ if egrn_row:
+ _cad_val = (
+ float(egrn_row["cadastral_value_rub"])
+ if egrn_row["cadastral_value_rub"] is not None
+ else None
+ )
+ _area_m2 = float(egrn_row["area_m2"]) if egrn_row["area_m2"] is not None else None
+ _idx = egrn_row["cost_index_per_m2"]
+ _cad_per_m2: float | None = None
+ if _idx is not None:
+ _cad_per_m2 = float(_idx)
+ elif _cad_val is not None and _area_m2 and _area_m2 > 0:
+ _cad_per_m2 = round(_cad_val / _area_m2, 2)
+ egrn_block = {
+ "cadastral_value_rub": _cad_val,
+ "cadastral_value_per_m2": _cad_per_m2,
+ "land_category": egrn_row["land_category"],
+ "permitted_use_text": egrn_row["permitted_use_text"],
+ "last_egrn_update_date": (
+ egrn_row["last_egrn_update_date"].isoformat()
+ if egrn_row["last_egrn_update_date"] is not None
+ else None
+ ),
+ "area_m2": _area_m2,
+ "ownership_type": egrn_row["ownership_type"],
+ "right_type": egrn_row["right_type"],
+ "parcel_status": egrn_row["status"],
+ "address": egrn_row["readable_address"],
+ "registration_date": (
+ egrn_row["registration_date"].isoformat()
+ if egrn_row["registration_date"] is not None
+ else None
+ ),
+ }
+ except Exception as e:
+ logger.warning("egrn_block query failed for %s: %s", cad_num, e)
+
+ # B5-2) Encumbrance block — ЗОУИТ из cad_zouit (SF-B5)
+ encumbrance_block: dict[str, Any] = {
+ "has_zouit": False,
+ "zouit_types": [],
+ "zouit_count": 0,
+ }
+ try:
+ zouit_rows = (
+ db.execute(
+ text("""
+ SELECT type_zone, name_by_doc
+ FROM cad_zouit
+ WHERE ST_Intersects(geom, ST_GeomFromText(:wkt, 4326))
+ ORDER BY id
+ """),
+ {"wkt": geom_wkt},
+ )
+ .mappings()
+ .all()
+ )
+ if zouit_rows:
+ _zouit_types = list({r["type_zone"] for r in zouit_rows if r["type_zone"]})
+ encumbrance_block = {
+ "has_zouit": True,
+ "zouit_types": _zouit_types,
+ "zouit_count": len(zouit_rows),
+ }
+ except Exception as e:
+ logger.warning("encumbrance_block query failed for %s: %s", cad_num, e)
+
+ # B5-3) Red lines block — пересечение с cad_red_lines (SF-B5)
+ red_lines_block: dict[str, Any] = {"intersects": False, "count": 0}
+ try:
+ rl_row = (
+ db.execute(
+ text("""
+ SELECT COUNT(*) AS cnt
+ FROM cad_red_lines
+ WHERE ST_Intersects(
+ geom::geometry,
+ ST_GeomFromText(:wkt, 4326)
+ )
+ """),
+ {"wkt": geom_wkt},
+ )
+ .mappings()
+ .first()
+ )
+ if rl_row:
+ _rl_cnt = int(rl_row["cnt"])
+ red_lines_block = {
+ "intersects": _rl_cnt > 0,
+ "count": _rl_cnt,
+ }
+ except Exception as e:
+ logger.warning("red_lines_block query failed for %s: %s", cad_num, e)
+
+ # B5-4) Metro placeholder — заполнится после merge 22h metro scraper
+ metro_block: dict[str, Any] = {"nearest_top3": None}
+
+ # B5-5) District price ranges из objective_lots (SF-B5)
+ district_price_block: dict[str, Any] = {
+ "district_price_per_m2_min": None,
+ "district_price_per_m2_max": None,
+ "district_price_per_m2_median": None,
+ "district_price_sample_size": None,
+ }
+ if district_row and district_row["district_name"]:
+ try:
+ dp_row = (
+ db.execute(
+ text("""
+ SELECT
+ MIN(price_per_m2_rub) AS price_min,
+ MAX(price_per_m2_rub) AS price_max,
+ PERCENTILE_CONT(0.5) WITHIN GROUP (
+ ORDER BY price_per_m2_rub
+ ) AS price_median,
+ COUNT(*) AS sample_size
+ FROM objective_lots
+ WHERE district = CAST(:dn AS text)
+ AND price_per_m2_rub IS NOT NULL
+ AND price_per_m2_rub BETWEEN 30000 AND 600000
+ """),
+ {"dn": district_row["district_name"]},
+ )
+ .mappings()
+ .first()
+ )
+ if dp_row and dp_row["sample_size"] and int(dp_row["sample_size"]) > 0:
+ district_price_block = {
+ "district_price_per_m2_min": (
+ round(float(dp_row["price_min"])) if dp_row["price_min"] else None
+ ),
+ "district_price_per_m2_max": (
+ round(float(dp_row["price_max"])) if dp_row["price_max"] else None
+ ),
+ "district_price_per_m2_median": (
+ round(float(dp_row["price_median"])) if dp_row["price_median"] else None
+ ),
+ "district_price_sample_size": int(dp_row["sample_size"]),
+ }
+ except Exception as e:
+ logger.warning("district_price_block query failed for %s: %s", cad_num, e)
+
+ # B5-6) Risk indicators — flood_zone из cad_risk_zones + noise_score + geology proxy (SF-B5)
+ risks_block: dict[str, Any] = {
+ "flood_zone": False,
+ "noise_score": round(noise_score, 2),
+ "geology_risk_label": None,
+ }
+ try:
+ flood_row = (
+ db.execute(
+ text("""
+ SELECT COUNT(*) AS cnt
+ FROM cad_risk_zones
+ WHERE ST_Intersects(
+ geom::geometry,
+ ST_GeomFromText(:wkt, 4326)
+ )
+ AND (risk_type ILIKE '%flood%' OR risk_type ILIKE '%подтоп%'
+ OR risk_type ILIKE '%затоп%')
+ """),
+ {"wkt": geom_wkt},
+ )
+ .mappings()
+ .first()
+ )
+ _flood = bool(flood_row and int(flood_row["cnt"]) > 0)
+ # Geology proxy через hydrology flood_risk_flag (уже посчитан выше)
+ _geo_flood = hydrology.get("flood_risk_flag", False) if hydrology else False
+ _has_flood = _flood or _geo_flood
+ # geology_risk_label: high если flooding, medium если шум > 65дБ, иначе low
+ if _has_flood:
+ _geo_label: str | None = "high"
+ elif noise_db_max >= 65.0:
+ _geo_label = "medium"
+ else:
+ _geo_label = "low"
+ risks_block = {
+ "flood_zone": _has_flood,
+ "noise_score": round(noise_score, 2),
+ "geology_risk_label": _geo_label,
+ }
+ except Exception as e:
+ logger.warning("risks_block query failed for %s: %s", cad_num, e)
+
# 10) Market trend — динамика цен ДДУ в радиусе 3 км за 6 vs предыдущие 6 месяцев
market_trend: dict[str, Any] | None = None
try:
@@ -2299,6 +2512,16 @@ def analyze_parcel(
},
# #254: custom POI scoring — user-defined points (via X-Session-Id header).
"custom_poi_score_items": custom_poi_items,
+ # SF-B5: EGRN + encumbrance + red_lines + metro + district prices + risks
+ "egrn": egrn_block,
+ "encumbrance": encumbrance_block,
+ "red_lines": red_lines_block,
+ "metro": metro_block,
+ "district_price_per_m2_min": district_price_block["district_price_per_m2_min"],
+ "district_price_per_m2_max": district_price_block["district_price_per_m2_max"],
+ "district_price_per_m2_median": district_price_block["district_price_per_m2_median"],
+ "district_price_sample_size": district_price_block["district_price_sample_size"],
+ "risks": risks_block,
}
@@ -2455,6 +2678,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,
@@ -2532,3 +2805,189 @@ async def get_parcel_best_layouts_pdf(
except Exception as exc:
logger.error("best_layouts PDF endpoint failed for %s: %s", cad_num, exc)
raise HTTPException(status_code=500, detail="Internal server error") from exc
+
+
+@router.get(
+ "/{cad_num}/snapshot.pdf",
+ summary="1-page PDF snapshot участка (НСПД + POI + конкуренты)",
+)
+def parcel_snapshot_pdf(
+ cad_num: str,
+ db: Annotated[Session, Depends(get_db)],
+) -> Response:
+ """Генерирует одностраничный PDF-снимок участка (A4).
+
+ Содержимое:
+ - Header: кадастровый номер, адрес, район, площадь
+ - Block 1: 5 KPI (площадь, кадастровая стоимость, категория, ВРИ, дата обновления)
+ - Block 2: Топ-7 POI по взвешенному баллу (из osm_poi_ekb, радиус 1 км)
+ - Block 3: Топ-5 конкурентов (из domrf_kn_objects, радиус 3 км)
+ - Footer: gendsgn.ru + дата генерации
+
+ Не является официальной выпиской ЕГРН — только аналитические данные НСПД.
+ Генерация <2 сек. Открывается в Adobe Reader / Chrome.
+ """
+ # 1) Получить метаданные участка из cad_parcels
+ parcel_row = (
+ db.execute(
+ text("""
+ SELECT readable_address AS address,
+ land_record_area AS area_m2,
+ land_record_category_type AS land_category,
+ permitted_use_established_by_document AS vri,
+ cost_value AS cadastral_cost,
+ updated_at AS last_update
+ FROM cad_parcels
+ WHERE cad_num = CAST(:c AS text)
+ LIMIT 1
+ """),
+ {"c": cad_num},
+ )
+ .mappings()
+ .first()
+ )
+
+ if not parcel_row:
+ raise HTTPException(
+ status_code=404,
+ detail=f"Участок {cad_num} не найден в БД. Используйте POST /analyze для загрузки.",
+ )
+
+ # 2) Получить геометрию (WKT) для POI / competitor queries
+ geom_row = (
+ db.execute(
+ text("""
+ SELECT ST_AsText(COALESCE(
+ (SELECT geom FROM cad_parcels_geom WHERE cad_num = CAST(:c AS text) LIMIT 1),
+ (SELECT geom FROM cad_parcels WHERE cad_num = CAST(:c AS text) LIMIT 1)
+ )) AS wkt
+ """),
+ {"c": cad_num},
+ )
+ .mappings()
+ .first()
+ )
+ geom_wkt: str | None = geom_row["wkt"] if geom_row else None
+
+ # 3) POI в радиусе 1 км (только если есть геометрия)
+ poi_rows: list[dict[str, Any]] = []
+ if geom_wkt:
+ poi_rows = [
+ dict(r)
+ for r in db.execute(
+ text("""
+ SELECT category,
+ name,
+ ST_Distance(
+ p.geom::geography,
+ ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography
+ ) AS distance_m
+ FROM osm_poi_ekb p
+ WHERE ST_DWithin(
+ p.geom::geography,
+ ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
+ 1000
+ )
+ ORDER BY distance_m ASC
+ LIMIT 50
+ """),
+ {"wkt": geom_wkt},
+ )
+ .mappings()
+ .all()
+ ]
+
+ # 4) Конкуренты в радиусе 3 км (только если есть геометрия)
+ competitor_rows: list[dict[str, Any]] = []
+ if geom_wkt:
+ competitor_rows = [
+ dict(r)
+ for r in db.execute(
+ text("""
+ WITH latest_obj AS (
+ SELECT DISTINCT ON (obj_id) *
+ FROM domrf_kn_objects
+ WHERE latitude IS NOT NULL
+ ORDER BY obj_id, snapshot_date DESC NULLS LAST
+ )
+ SELECT obj_id,
+ comm_name,
+ dev_name,
+ obj_class,
+ flat_count,
+ ST_Distance(
+ ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
+ ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography
+ ) AS distance_m
+ FROM latest_obj o
+ WHERE ST_DWithin(
+ ST_SetSRID(ST_MakePoint(o.longitude, o.latitude), 4326)::geography,
+ ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,
+ 3000
+ )
+ ORDER BY flat_count DESC NULLS LAST
+ LIMIT 20
+ """),
+ {"wkt": geom_wkt},
+ )
+ .mappings()
+ .all()
+ ]
+
+ # 5) Получить district (через пересечение с ekb_districts если есть геом)
+ district: str | None = None
+ if geom_wkt:
+ district_row = (
+ db.execute(
+ text("""
+ SELECT d.district_name
+ FROM ekb_districts d
+ WHERE ST_Contains(d.geom, ST_Centroid(ST_GeomFromText(:wkt, 4326)))
+ LIMIT 1
+ """),
+ {"wkt": geom_wkt},
+ )
+ .mappings()
+ .first()
+ )
+ if district_row:
+ district = district_row["district_name"]
+
+ # 6) Форматировать last_update
+ raw_update = parcel_row["last_update"]
+ last_update_str: str | None = None
+ if raw_update is not None:
+ try:
+ last_update_str = raw_update.strftime("%d.%m.%Y")
+ except AttributeError:
+ last_update_str = str(raw_update)[:10]
+
+ # 7) Сгенерировать PDF
+ try:
+ pdf_bytes = generate_snapshot_pdf(
+ cad_num=cad_num,
+ address=parcel_row["address"],
+ district=district,
+ area_m2=float(parcel_row["area_m2"]) if parcel_row["area_m2"] is not None else None,
+ cadastral_cost_rub=(
+ float(parcel_row["cadastral_cost"])
+ if parcel_row["cadastral_cost"] is not None
+ else None
+ ),
+ land_category=parcel_row["land_category"],
+ vri=parcel_row["vri"],
+ last_update=last_update_str,
+ poi_rows=poi_rows,
+ competitor_rows=competitor_rows,
+ competitors_limit=5,
+ )
+ except Exception as exc:
+ logger.error("snapshot PDF generation failed for %s: %s", cad_num, exc)
+ raise HTTPException(status_code=500, detail="Ошибка генерации PDF") from exc
+
+ cad_safe = cad_num.replace(":", "-")
+ return Response(
+ content=pdf_bytes,
+ media_type="application/pdf",
+ headers={"Content-Disposition": f'attachment; filename="snapshot-{cad_safe}.pdf"'},
+ )
diff --git a/backend/app/api/v1/pilot.py b/backend/app/api/v1/pilot.py
new file mode 100644
index 00000000..7ff2b4e4
--- /dev/null
+++ b/backend/app/api/v1/pilot.py
@@ -0,0 +1,74 @@
+"""Pilot request lead-gen endpoint.
+
+POST /api/v1/pilot/request — принимает заявку на пилот (лид с лендинга или страницы анализа),
+сохраняет в таблицу pilot_requests.
+Telegram-уведомление — TODO (creds не настроены, см. #307 SF-B3).
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Annotated, Any, Literal
+
+from fastapi import APIRouter, Depends, Request
+from pydantic import BaseModel, EmailStr, Field
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.core.db import get_db
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+class PilotRequestInput(BaseModel):
+ name: str = Field(min_length=2, max_length=200)
+ phone: str | None = Field(default=None, max_length=50)
+ email: EmailStr | None = None
+ company: str | None = Field(default=None, max_length=200)
+ message: str | None = Field(default=None, max_length=2000)
+ source: Literal["landing", "analyze_page", "other"] = "landing"
+
+
+@router.post("/request")
+async def create_pilot_request(
+ payload: PilotRequestInput,
+ request: Request,
+ db: Annotated[Session, Depends(get_db)],
+) -> dict[str, Any]:
+ """Сохраняет заявку на пилот в pilot_requests."""
+ user_agent = request.headers.get("user-agent")
+
+ row = (
+ db.execute(
+ text(
+ """
+ INSERT INTO pilot_requests (name, phone, email, company, message, source, user_agent)
+ VALUES (:name, :phone, :email, :company, :message, :source, :user_agent)
+ RETURNING CAST(id AS text), created_at
+ """
+ ),
+ {
+ "name": payload.name,
+ "phone": payload.phone,
+ "email": str(payload.email) if payload.email else None,
+ "company": payload.company,
+ "message": payload.message,
+ "source": payload.source,
+ "user_agent": user_agent,
+ },
+ )
+ .mappings()
+ .one()
+ )
+
+ db.commit()
+
+ logger.info("pilot_request saved id=%s source=%s", row["id"], payload.source)
+
+ return {
+ "id": row["id"],
+ "created_at": row["created_at"].isoformat(),
+ "status": "received",
+ }
diff --git a/backend/app/main.py b/backend/app/main.py
index c22dcbe8..07071a70 100644
--- a/backend/app/main.py
+++ b/backend/app/main.py
@@ -26,6 +26,7 @@ from app.api.v1 import (
landing,
parcels,
photos,
+ pilot,
trade_in,
users,
)
@@ -102,6 +103,7 @@ app.include_router(
)
app.include_router(trade_in.router, prefix="/api/v1/trade-in", tags=["trade-in"])
app.include_router(landing.router, prefix="/api/v1", tags=["landing"])
+app.include_router(pilot.router, prefix="/api/v1/pilot", tags=["pilot"])
app.include_router(users.router, prefix="/api/v1", tags=["users"])
diff --git a/backend/app/services/exporters/snapshot_pdf.py b/backend/app/services/exporters/snapshot_pdf.py
new file mode 100644
index 00000000..756e25a6
--- /dev/null
+++ b/backend/app/services/exporters/snapshot_pdf.py
@@ -0,0 +1,204 @@
+"""Генерация одностраничного PDF-снимка кадастрового участка.
+
+Использует WeasyPrint + Jinja2. Шрифты — DejaVu Sans из системы (Dockerfile)
+или из пакета weasyprint (font fallback). Шаблон: app/templates/parcel_snapshot.html.
+"""
+
+import datetime
+import logging
+import pathlib
+from typing import Any
+
+from jinja2 import Environment, FileSystemLoader, select_autoescape
+
+logger = logging.getLogger(__name__)
+
+# Путь к директории шаблонов (относительно этого файла — 2 уровня вверх, затем templates)
+_TEMPLATE_DIR = pathlib.Path(__file__).parent.parent / "templates"
+
+# Системные пути DejaVu Sans (Ubuntu/Debian Docker-образ + Alpine резерв)
+_DEJAVU_CANDIDATES: list[str] = [
+ "/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf",
+ "/usr/share/fonts/dejavu/DejaVuSans.ttf",
+ "/usr/share/fonts/TTF/DejaVuSans.ttf",
+]
+
+_CATEGORY_RU: dict[str, str] = {
+ "school": "Школа",
+ "kindergarten": "Детский сад",
+ "pharmacy": "Аптека",
+ "hospital": "Больница",
+ "shop_mall": "ТЦ",
+ "shop_supermarket": "Супермаркет",
+ "shop_small": "Магазин",
+ "park": "Парк",
+ "bus_stop": "Автобус",
+ "metro_stop": "Метро",
+ "tram_stop": "Трамвай",
+}
+
+# Веса POI-категорий — должны совпадать с _POI_WEIGHTS в parcels.py.
+# Дублированы здесь чтобы exporter не импортировал из api-слоя.
+_POI_WEIGHTS: dict[str, float] = {
+ "school": 1.5,
+ "kindergarten": 1.5,
+ "pharmacy": 0.8,
+ "hospital": 0.6,
+ "shop_mall": 1.2,
+ "shop_supermarket": 1.0,
+ "shop_small": 0.5,
+ "park": 1.8,
+ "bus_stop": 0.3,
+ "metro_stop": 1.5,
+ "tram_stop": -0.5,
+}
+
+_WALK_SPEED_M_PER_MIN: float = 80.0 # ~5 км/ч
+
+
+def _find_font_url() -> str:
+ """Вернуть file:// URL для DejaVu Sans или пустую строку (system fallback).
+
+ WeasyPrint умеет сам находить системные шрифты через fonttools/fontconfig,
+ поэтому пустая строка допустима — шрифт тогда подбирается CSS generic.
+ """
+ for path in _DEJAVU_CANDIDATES:
+ if pathlib.Path(path).exists():
+ return f"file://{path}"
+ logger.warning(
+ "snapshot_pdf: DejaVu Sans не найден в стандартных путях — используем system fallback"
+ )
+ return ""
+
+
+def _format_cost(value: float | None) -> str:
+ """Форматировать кадастровую стоимость в читаемый вид (млн/тыс ₽)."""
+ if value is None:
+ return "—"
+ if value >= 1_000_000:
+ return f"{value / 1_000_000:.1f} млн ₽"
+ if value >= 1_000:
+ return f"{value / 1_000:.0f} тыс ₽"
+ return f"{value:.0f} ₽"
+
+
+def _build_poi_items(poi_rows: list[dict[str, Any]], limit: int = 7) -> list[dict[str, Any]]:
+ """Вычислить weighted_score для каждого POI и вернуть топ-N отсортированных.
+
+ Формула: weighted_score = weight * max(0, 1 - distance_m / 1000)
+ Отрицательные вклады (трамвай) — не включаем в топ-список.
+ """
+ items: list[dict[str, Any]] = []
+ for p in poi_rows:
+ cat: str = p.get("category", "")
+ w = _POI_WEIGHTS.get(cat, 0.0)
+ distance_m = float(p.get("distance_m") or 0)
+ decay = max(0.0, 1.0 - distance_m / 1000.0)
+ score = round(w * decay, 2)
+ if score <= 0:
+ continue
+ walk_min = max(1, round(distance_m / _WALK_SPEED_M_PER_MIN))
+ items.append(
+ {
+ "category_ru": _CATEGORY_RU.get(cat, cat),
+ "name": p.get("name") or "",
+ "distance_m": round(distance_m),
+ "walk_min": walk_min,
+ "weighted_score": score,
+ }
+ )
+ items.sort(key=lambda x: x["weighted_score"], reverse=True)
+ return items[:limit]
+
+
+def generate_snapshot_pdf(
+ *,
+ cad_num: str,
+ address: str | None,
+ district: str | None,
+ area_m2: float | None,
+ cadastral_cost_rub: float | None,
+ land_category: str | None,
+ vri: str | None,
+ last_update: str | None,
+ poi_rows: list[dict[str, Any]],
+ competitor_rows: list[dict[str, Any]],
+ competitors_limit: int = 5,
+) -> bytes:
+ """Сгенерировать PDF-снимок участка (1 страница A4).
+
+ Аргументы:
+ cad_num: кадастровый номер.
+ address: адрес из cad_parcels.
+ district: район города.
+ area_m2: площадь в кв. м (конвертируем в га для отображения).
+ cadastral_cost_rub: кадастровая стоимость в рублях.
+ land_category: категория земель.
+ vri: вид разрешённого использования.
+ last_update: строка даты последнего обновления данных.
+ poi_rows: сырые строки из osm_poi_ekb (category, name, distance_m).
+ competitor_rows: строки конкурентов из domrf_kn_objects.
+ competitors_limit: сколько конкурентов выводить (3-5 по ТЗ).
+
+ Возвращает: bytes PDF-документа.
+ """
+ # WeasyPrint импортируем локально — тяжёлый; не нужен при импорте модуля
+ try:
+ from weasyprint import HTML
+ except ImportError as exc:
+ raise RuntimeError(
+ "WeasyPrint не установлен. Добавь 'weasyprint>=62.0' в pyproject.toml."
+ ) from exc
+
+ env = Environment(
+ loader=FileSystemLoader(str(_TEMPLATE_DIR)),
+ autoescape=select_autoescape(["html"]),
+ )
+ template = env.get_template("parcel_snapshot.html")
+
+ area_ha = f"{area_m2 / 10_000:.2f}" if area_m2 else "—"
+ poi_items = _build_poi_items(poi_rows, limit=7)
+
+ # Конкуренты — берём топ N ближайших (уже отсортированы по flat_count DESC;
+ # переупорядочиваем по distance_m для удобства чтения)
+ competitors_display = sorted(
+ competitor_rows[:competitors_limit],
+ key=lambda r: float(r.get("distance_m") or 0),
+ )
+ competitors_ctx: list[dict[str, Any]] = [
+ {
+ "comm_name": r.get("comm_name"),
+ "dev_name": r.get("dev_name"),
+ "obj_class": r.get("obj_class"),
+ "flat_count": r.get("flat_count"),
+ "distance_m": round(float(r.get("distance_m") or 0)),
+ }
+ for r in competitors_display
+ ]
+
+ generated_at = datetime.datetime.now(tz=datetime.UTC).strftime("%d.%m.%Y %H:%M UTC")
+
+ html_str = template.render(
+ cad_num=cad_num,
+ address=address,
+ district=district,
+ area_ha=area_ha,
+ cadastral_cost=_format_cost(cadastral_cost_rub),
+ land_category=land_category,
+ vri=vri,
+ last_update=last_update or "—",
+ poi_items=poi_items,
+ competitors=competitors_ctx,
+ generated_at=generated_at,
+ font_url=_find_font_url(),
+ )
+
+ logger.info(
+ "snapshot_pdf: rendering PDF for %s (%d POI, %d competitors)",
+ cad_num,
+ len(poi_items),
+ len(competitors_ctx),
+ )
+
+ pdf_bytes: bytes = HTML(string=html_str, base_url=str(_TEMPLATE_DIR)).write_pdf()
+ return pdf_bytes
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/app/templates/parcel_snapshot.html b/backend/app/templates/parcel_snapshot.html
new file mode 100644
index 00000000..b8e9ae74
--- /dev/null
+++ b/backend/app/templates/parcel_snapshot.html
@@ -0,0 +1,240 @@
+
+
+
+
+ Карточка участка {{ cad_num }}
+
+
+
+
+
+
+
+
+Основные характеристики
+
+
+
Площадь
+
{{ area_ha }} га
+
+
+
Кадастровая стоимость
+
{{ cadastral_cost }}
+
+
+
Категория земель
+
{{ land_category or '—' }}
+
+
+
ВРИ
+
{{ vri or '—' }}
+
+
+
Последнее обновление
+
{{ last_update or '—' }}
+
+
Ближайшая инфраструктура (топ-7 по взвешенному баллу)
+{% if poi_items %}
+
+
+
+ | Категория |
+ Название |
+ Расстояние |
+ Пешком |
+ Балл |
+
+
+
+ {% for poi in poi_items %}
+
+ | {{ poi.category_ru }} |
+ {{ poi.name or '—' }} |
+ {{ poi.distance_m }} м |
+ {{ poi.walk_min }} мин |
+ {{ poi.weighted_score }} |
+
+ {% endfor %}
+
+
+{% else %}
+POI в радиусе 1 км не найдены.
+{% endif %}
+
+
+Конкуренты в радиусе 3 км (топ {{ competitors|length }})
+{% if competitors %}
+
+
+
+ | ЖК / Объект |
+ Застройщик |
+ Класс |
+ Квартир |
+ Расстояние |
+
+
+
+ {% for c in competitors %}
+
+ | {{ c.comm_name or '—' }} |
+ {{ c.dev_name or '—' }} |
+ {{ c.obj_class or '—' }} |
+ {{ c.flat_count or '—' }} |
+ {{ c.distance_m }} м |
+
+ {% endfor %}
+
+
+{% else %}
+Конкурентов в радиусе 3 км не обнаружено.
+{% endif %}
+
+
+ Не является выпиской из ЕГРН. Данные носят аналитический характер.
+ Для официальной выписки: rosreestr.gov.ru
+
+
+
+
+
+
+
diff --git a/backend/pyproject.toml b/backend/pyproject.toml
index dff4278e..1bd36e47 100644
--- a/backend/pyproject.toml
+++ b/backend/pyproject.toml
@@ -22,6 +22,7 @@ dependencies = [
"tenacity>=9.0.0",
"pillow>=10.4.0",
"weasyprint>=62.0",
+ "jinja2>=3.1.0",
"ezdxf>=1.3.0",
"openpyxl>=3.1.0",
"pandas>=2.2.0",
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
diff --git a/backend/uv.lock b/backend/uv.lock
index 439b0a93..d1643ad0 100644
--- a/backend/uv.lock
+++ b/backend/uv.lock
@@ -568,6 +568,7 @@ dependencies = [
{ name = "geopandas" },
{ name = "httpx" },
{ name = "ijson" },
+ { name = "jinja2" },
{ name = "numpy" },
{ name = "openpyxl" },
{ name = "pandas" },
@@ -608,6 +609,7 @@ requires-dist = [
{ name = "geopandas", specifier = ">=1.0.0" },
{ name = "httpx", specifier = ">=0.27.0" },
{ name = "ijson", specifier = ">=3.2.0" },
+ { name = "jinja2", specifier = ">=3.1.0" },
{ name = "numpy", specifier = ">=2.0.0" },
{ name = "openpyxl", specifier = ">=3.1.0" },
{ name = "pandas", specifier = ">=2.2.0" },
@@ -871,6 +873,18 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" },
]
+[[package]]
+name = "jinja2"
+version = "3.1.6"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+ { name = "markupsafe" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/df/bf/f7da0350254c0ed7c72f3e33cef02e048281fec7ecec5f032d4aac52226b/jinja2-3.1.6.tar.gz", hash = "sha256:0137fb05990d35f1275a587e9aee6d56da821fc83491a0fb838183be43f66d6d", size = 245115, upload-time = "2025-03-05T20:05:02.478Z" }
+wheels = [
+ { url = "https://files.pythonhosted.org/packages/62/a1/3d680cbfd5f4b8f15abc1d571870c5fc3e594bb582bc3b64ea099db13e56/jinja2-3.1.6-py3-none-any.whl", hash = "sha256:85ece4451f492d0c13c5dd7c13a64681a86afae63a5f347908daf103ce6d2f67", size = 134899, upload-time = "2025-03-05T20:05:00.369Z" },
+]
+
[[package]]
name = "joblib"
version = "1.5.3"
diff --git a/data/sql/118_pilot_requests.sql b/data/sql/118_pilot_requests.sql
new file mode 100644
index 00000000..7bdb6e63
--- /dev/null
+++ b/data/sql/118_pilot_requests.sql
@@ -0,0 +1,21 @@
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS pilot_requests (
+ id uuid PRIMARY KEY DEFAULT gen_random_uuid(),
+ name text NOT NULL,
+ phone text,
+ email text,
+ company text,
+ message text,
+ source text, -- 'landing', 'analyze_page', etc
+ user_agent text,
+ created_at timestamptz NOT NULL DEFAULT NOW(),
+ notified_at timestamptz -- when Telegram sent
+);
+
+CREATE INDEX IF NOT EXISTS pilot_requests_created_idx
+ ON pilot_requests (created_at DESC);
+
+COMMENT ON TABLE pilot_requests IS '#307 SF-B3 lead generation, опционально notified в Telegram.';
+
+COMMIT;