"""Схема улицы сделки: готовые SVG-пути для карточки лэндинга. ЗАЧЕМ ГОТОВАЯ СХЕМА, А НЕ ГЕОМЕТРИЯ. Отдать фронту GeoJSON окна 840×840 м — это десятки килобайт на КАЖДУЮ строку витрины плюс проекция в рантайме. Здесь всё считается один раз при пересчёте витрины и хранится строкой SVG-путей; порядок величины на прототипе — 4 КБ на сделку. ЧТО ЭТА СХЕМА ЗАЯВЛЯЕТ, А ЧТО НЕТ --------------------------------- * ДОМА В НЕЙ НЕТ И БЫТЬ НЕ МОЖЕТ. `deals.address` — уровня улицы («Екатеринбург, Краснолесья»), номер дома известен у 2.7% сделок. Точка дома была бы выдумана, поэтому её не рисует даже фронт: в схеме нет ни координат окна, ни констант проекции — по ней физически нельзя положить (lon, lat) в её систему координат. Это не забывчивость, а замок. * ЗДАНИЙ В НЕЙ НЕТ. Замер 2026-08-29: `cad_buildings` содержит 18 307 контуров на весь ЕКБ, в плотном центре это 51 здание на радиус 450 м, где их в разы больше. Нарисованная застройка заявляла бы полноту, которой в данных нет, и показывала бы пустые кварталы там, где стоят дома. * УЛИЦЫ — ФИЛЬТРОВАННАЯ ВЫГРУЗКА. `osm_noise_sources_ekb` собиралась как «источники шума»: 10 025 линий, 1377 различных названий. Именованные улицы покрыты хорошо, дворовые проезды и служебные отсутствуют. Схема — это именно улицы района, а не полная уличная сеть. * УЛИЦА СМАТЧИЛАСЬ НЕ ВСЕГДА. Нормализация даёт совпадение у 550 названий из 654 — это 31 410 сделок из 34 021 (92.3%, замер на проде 2026-08-29 по той же выборке, из которой набирается витрина). Оставшиеся 8% — штатный случай: схемы нет, фронт показывает район. ПРОЕКЦИЯ — та же равнопромежуточная с поправкой cos(широты), что в `scripts/export_ekb_districts_svg.py`. Второй проекции в проекте нет намеренно. Окно квадратное в метрах, поэтому viewBox тоже квадратный: 1000×1000. """ from __future__ import annotations import json import logging import math import re from dataclasses import dataclass from typing import Any from sqlalchemy import text from sqlalchemy.orm import Session logger = logging.getLogger(__name__) # Полуразмер окна в метрах. 420 м на 1000 единиц viewBox — это ~1.2 м на # единицу: улица читается как улица, а соседние кварталы попадают в кадр. RADIUS_M = 420.0 VIEWBOX = 1000.0 # Метров в градусе широты (сфера). Долгота — то же, делённое на cos(широты). M_PER_DEG_LAT = 111_320.0 # Допуск упрощения в градусах ≈ 3 м: меньше трёх единиц viewBox, то есть тоньше # линии. Смена значения обязана сопровождаться замером размера схемы. SIMPLIFY_DEG = 0.00003 # Сколько подписей соседних улиц кладём в схему. Больше — каша на карточке # 320 px и лишние байты; отбираем самые длинные в кадре. MAX_LABELS = 7 # Родовые слова, которые в `deals.address` и в OSM пишутся по-разному # («Краснолесья» ↔ «улица Краснолесья»). Снимаем с обеих сторон. _STREET_TYPES = frozenset( { "улица", "ул", "проспект", "пр", "пр-кт", "пркт", "проспкт", "переулок", "пер", "бульвар", "бул", "б-р", "шоссе", "ш", "проезд", "набережная", "наб", "тракт", "аллея", "площадь", "пл", "тупик", } ) _KEEP = re.compile(r"[^0-9a-zа-я\s-]") def normalize_street(address: str | None) -> str | None: """`'Екатеринбург, ул. Краснолесья'` → ``'краснолесья'``. Нет улицы — None. Что делает и почему именно так: * берёт часть ПОСЛЕ последней запятой — в `deals.address` город стоит первым («Екатеринбург, 8 Марта»); * ПЕРЕКРЁСТОК берёт первой улицей. «Екатеринбург, Челюскинцев/Шейнкмана» → «челюскинцев». Таких адресов в выборке ТРИ на 34 021 сделку (по одной сделке на каждый, замер 2026-08-29), и обе улицы одинаково верны: адрес и так уровня улицы, а не дома. Рисовать перекрёсток точкой значило бы заявить точность, которой в остальных 34 018 строках нет, ради трёх; * складывает ё→е: «Королёва» в адресе и «Королева» в OSM — одна улица; * снимает родовое слово (улица/проспект/…), потому что в адресе его обычно нет, а в OSM обычно есть; * цифры оставляет: «8 Марта» и «22 Партсъезда» — названия, а не номера. """ if not address: return None tail = address.rsplit(",", 1)[-1] tail = tail.split("/")[0] tail = tail.replace("ё", "е").replace("Ё", "Е").lower() tail = _KEEP.sub(" ", tail) words = [w for w in tail.split() if w.strip("-") not in _STREET_TYPES] return " ".join(words) or None # ── Запросы ────────────────────────────────────────────────────────────────── # # Обе идут в foreign table `gendesign_osm_roads_ekb` (миграция 281) над вьюхой # `v_tradein_osm_roads_ekb` чужой базы. Сервер `gendesign_remote` объявлен с # OPTIONS (extensions 'postgis'), поэтому PostGIS-операторы shippable: `&&` с # готовым конвертом уходит на ту сторону и попадает в GiST-индекс, а обратно # едет уже обрезанный GeoJSON, а не десять тысяч линий. _NAMES_SQL = text("SELECT DISTINCT name FROM gendesign_osm_roads_ekb WHERE name IS NOT NULL") _CENTER_SQL = text( """ SELECT ST_X(p) AS lon, ST_Y(p) AS lat FROM ( SELECT ST_ClosestPoint(g, ST_Centroid(g)) AS p FROM ( SELECT ST_Collect(geom) AS g FROM gendesign_osm_roads_ekb WHERE name = ANY(CAST(:names AS text[])) ) t WHERE g IS NOT NULL ) q """ ) _WINDOW_SQL = text( """ SELECT name, road_class, ST_AsGeoJSON( ST_SimplifyPreserveTopology( ST_Intersection( geom, ST_MakeEnvelope(:min_lon, :min_lat, :max_lon, :max_lat, 4326) ), CAST(:tol AS double precision) ), 6 ) AS gj FROM gendesign_osm_roads_ekb WHERE geom && ST_MakeEnvelope(:min_lon, :min_lat, :max_lon, :max_lat, 4326) """ ) # Классы из `osm_noise_sources_ekb`, которые рисуются водой. Остальные (trunk / # primary / secondary / tertiary / residential) — дороги. Разделение # перечислением, а не «всё, что не дорога»: во вьюху могут доехать новые классы, # и тогда лучше не нарисовать, чем нарисовать реку поверх улицы. _WATER_CLASSES = frozenset({"river", "canal", "stream", "lake_or_pond"}) @dataclass(frozen=True) class StreetIndex: """Нормализованное название → как эта улица называется в OSM. Строится ОДИН раз на прогон: 1377 названий на всю базу, тянуть их на каждую из двадцати сделок незачем. """ by_norm: dict[str, list[str]] def lookup(self, address: str | None) -> tuple[str, list[str]] | None: key = normalize_street(address) if key is None: return None raw = self.by_norm.get(key) return (key, raw) if raw else None def load_street_index(db: Session) -> StreetIndex: """Список названий улиц из FDW-вьюхи. Недоступна — пустой индекс. SAVEPOINT ИМЕННО ЗДЕСЬ, на месте глушения: провалившийся SELECT переводит транзакцию в aborted, и следующий запрос упал бы уже не по своей вине. Гранты на той стороне уже терялись при пересоздании объекта (#3227). """ try: with db.begin_nested(): rows = db.execute(_NAMES_SQL).all() except Exception as exc: logger.warning("улицы не резолвятся (витрина будет без схем): %s", exc) return StreetIndex(by_norm={}) by_norm: dict[str, list[str]] = {} for (name,) in rows: key = normalize_street(name) if key: by_norm.setdefault(key, []).append(name) return StreetIndex(by_norm=by_norm) def _parts(gj: dict[str, Any]) -> list[list[list[float]]]: """GeoJSON любой из встречающихся форм → список цепочек точек.""" t = gj.get("type") c = gj.get("coordinates") or [] if t == "LineString": return [c] if t == "MultiLineString": return list(c) if t == "Polygon": return list(c) if t == "MultiPolygon": return [ring for poly in c for ring in poly] if t == "GeometryCollection": return [p for g in gj.get("geometries", []) for p in _parts(g)] return [] def _length(chain: list[list[float]]) -> float: return sum(math.dist(chain[i], chain[i + 1]) for i in range(len(chain) - 1)) def build_street_scheme(db: Session, index: StreetIndex, address: str | None) -> dict | None: """Схема улицы для одной сделки, либо None — и None это ШТАТНО. None означает ровно одно: «улицы сделки в геометрии нет» (нет адреса, не сматчилось название, вьюха недоступна, у улицы нет геометрии). Витрина в этом случае отдаёт строку без схемы, а фронт показывает район — запасной вариант, который для этого и оставлен. """ found = index.lookup(address) if found is None: return None key, raw_names = found try: with db.begin_nested(): center = db.execute(_CENTER_SQL, {"names": raw_names}).mappings().first() if center is None: return None lat, lon = float(center["lat"]), float(center["lon"]) dlat = RADIUS_M / M_PER_DEG_LAT k = math.cos(math.radians(lat)) dlon = dlat / k bounds = { "min_lon": lon - dlon, "max_lon": lon + dlon, "min_lat": lat - dlat, "max_lat": lat + dlat, "tol": SIMPLIFY_DEG, } rows = db.execute(_WINDOW_SQL, bounds).mappings().all() except Exception as exc: logger.warning("схема улицы %r не собралась: %s", key, exc) return None min_lon, max_lat = bounds["min_lon"], bounds["max_lat"] # Окно квадратное в метрах, значит после поправки cos(широты) его ширина и # высота в единицах viewBox равны: масштаб один на обе оси. scale = VIEWBOX / (2 * dlat) def path(chain: list[list[float]]) -> str: out = [] for i, (plon, plat) in enumerate(chain): x = round((plon - min_lon) * k * scale, 1) y = round((max_lat - plat) * scale, 1) out.append(f"{'M' if i == 0 else 'L'}{x} {y}") return "".join(out) target: list[str] = [] roads: list[dict[str, str]] = [] water: list[str] = [] label_pool: list[tuple[float, str, list[float]]] = [] targets = set(raw_names) for row in rows: if not row["gj"]: continue chains = [c for c in _parts(json.loads(row["gj"])) if len(c) > 1] if not chains: continue cls = row["road_class"] is_target = row["name"] in targets for chain in chains: d = path(chain) if is_target: target.append(d) elif cls in _WATER_CLASSES: water.append(d) else: roads.append({"c": cls or "residential", "d": d}) if row["name"] and not is_target and cls not in _WATER_CLASSES: longest = max(chains, key=_length) label_pool.append((_length(longest), row["name"], longest[len(longest) // 2])) if not target: return None # Подписи: самые длинные в кадре, по одной на название. labels: list[dict[str, Any]] = [] seen: set[str] = set() for _, name, point in sorted(label_pool, key=lambda t: -t[0]): if name in seen: continue seen.add(name) labels.append( { "t": name, "x": round((point[0] - min_lon) * k * scale, 1), "y": round((max_lat - point[1]) * scale, 1), } ) if len(labels) >= MAX_LABELS: break return { "street": raw_names[0], "w": VIEWBOX, "h": VIEWBOX, "target": target, "roads": roads, "water": water, "labels": labels, }