Проверка участка за 30 секунд: что ломается при работе с НСПД и OSM

Задача звучит просто. Пользователь вводит адрес квартиры или кадастровый номер участка, бэкенд на Python за 30 секунд опрашивает открытые реестры и считает матрицу рисков: охранные зоны ЗОУИТ, ЛЭП, зоны затопления, водоохранные зоны, придорожные полосы, санитарно-защитные зоны промышленных объектов, рельеф, экология. На выходе — структурированный PDF на 9 страниц.

На практике открытые государственные геоданные ведут себя непредсказуемо: документация расходится с реальными ответами, координаты разъезжаются, параллельные запросы вешают соединение, а WAF блокирует запросы даже с российских IP. Ниже — конкретные проблемы и решения, которые пришлось внедрить.

Схема пайплайна

Последовательность шагов:

Нормализация адреса.
Если есть кадастровый номер — прямой поиск объекта в НСПД через find. Если нет — поиск по маске адреса в реестре ОКС НСПД.
Фиксация кадастрового контура объекта.
Три независимых источника данных по контуру:
- зоны ЗОУИТ из НСПД (search_in_contour);
- граф Overpass OSM: трассы, железные дороги, свалки;
- рельеф Copernicus: уклон, риск затопления.
Скоринг в Risk Engine.
Текстовое заключение от LLM (DeepSeek v4).
Сборка PDF через WeasyPrint.

Стек: Python 3.12, FastAPI, Pydantic v2, asyncio; для геометрии — shapely и pyproj; Overpass API для оценки шума и транспортных артерий; Copernicus GLO-90 для высот и уклонов; Jinja2 + WeasyPrint для PDF. Всё это крутится в Docker Compose за Caddy с автоматическими сертификатами Let’s Encrypt на обычном VPS.

Почему asyncio.gather здесь вредит

Для слоёв Национальной системы пространственных данных и Публичной кадастровой карты используется клиент на базе библиотеки pynspd. Первое желание — запустить asyncio.gather() и параллельно забрать карточку здания, выписку по участку и слои ЗОУИТ. С НСПД это не работает по двум причинам.

WMS GetFeatureInfo по точке блокируется. Запросы к /api/aeggis/v3/.../wms по координатам (search_at_coords) получают HTTP 403 BlockedIP, причём даже из российского дата-центра Ростелекома.

Параллельные соединения рвутся. Три одновременных коннекта от одного процесса к эндпоинтам НСПД заканчиваются таймаутами и сбросом TCP-соединения.

Рабочий вариант — строго последовательные запросы внутри одного асинхронного клиента (app/pipeline/nspd_engine.py):

async def fetch_cadastral_data(cadastral_number: str, coords: tuple[float, float]):
    async with AsyncNspd() as client:
        # 1. Сначала запрашиваем точную карточку объекта
        obj = await client.find(cadastral_number)
        if not obj or not obj.geometry:
            # Fallback: не падаем с 500, возвращаем честный статус
            return {"status": "unavailable", "layers": []}

        # 2. Пересечение зон ищем через search_in_contour полигона
        contour_geojson = obj.geometry
        zones = await client.search_in_contour(
            geom=contour_geojson,
            layers=["zouit_layer", "flood_layer", "forestry_layer"]
        )
        return {"status": "available", "data": zones}

Зоны ищутся пересечением с полигоном участка. Точечный identify по координатам из пайплайна убран полностью: это снимает и ложные срабатывания, и блокировки WAF. Попутно здесь видно второе правило — если объект не найден или у него нет геометрии, функция возвращает unavailable вместо исключения и 500-й ошибки.

Для homelab-проектов, которые ходят во внешние API, вывод переносится напрямую: если сервис нестабилен под параллельной нагрузкой от одного клиента, последовательные запросы в одной сессии дают предсказуемое время ответа, а gather даёт случайные таймауты.

Геокодер ставит дом на дорогу

Геокодер Suggestions хорошо исправляет опечатки в адресах, но в частном секторе — СНТ, деревни, ИЖС — его координаты ошибаются на 50–100 метров. На живом тесте в Твери маркер дома попал в соседний кадастровый квартал, прямо на полотно автодороги. Сервис нашёл кадастровый номер дороги вместо дома клиента и строил бы отчёт по нему.

После этого появилась жёсткая иерархия источников:

Введён кадастровый номер — координаты геокодера игнорируются, полигон и центроид берутся только из ЕГРН/НСПД.
Введён текстовый адрес — геокодер нормализует название улицы и номер дома. Сам объект ищется по шаблону реестра ОКС НСПД: г. {city}, ул. {street}, д. {house}.
Координаты геокодера используются только как аварийный fallback.

Адреса из ЕГРН тоже нужно чистить: Росреестр добавляет служебные ориентиры, которые мешают поиску.

import re

def clean_egrn_address(raw_address: str) -> str:
    # Удаляем служебные ориентиры Росреестра
    garbage_patterns = [
        r"Местоположение установлено относительно ориентира.*?,",
        r"участок находится в границах.*?,",
        r"примерно в \d+ метрах по направлению.*",
    ]
    cleaned = raw_address
    for pattern in garbage_patterns:
        cleaned = re.sub(pattern, "", cleaned, flags=re.IGNORECASE)
    return cleaned.strip().strip(",")

Год постройки живёт в поле с опечаткой

При разборе паспорта многоквартирного дома поле year_built в ответах стабильно приходило пустым — "" или null. Год постройки нашёлся в поле year_commisioning, с одной s.

Строгая Pydantic-схема без алиасов в такой ситуации молча решает, что год неизвестен, и ошибку легко не заметить. Исправление — алиас плюс фабричный метод, который проверяет оба поля:

from pydantic import BaseModel, Field

class BuildingPassport(BaseModel):
    cadastral_number: str
    floors_total: int | None = None
    wall_material: str | None = None
    # Обработка опечатки в госапи
    year_built: int | None = Field(
        default=None, validation_alias="year_commisioning"
    )

    @classmethod
    def from_api(cls, raw: dict):
        year = raw.get("year_built") or raw.get("year_commisioning")
        return cls(
            cadastral_number=raw.get("cad_num", ""),
            floors_total=raw.get("floors"),
            wall_material=raw.get("walls"),
            year_built=int(year) if str(year).isdigit() else None,
        )

Заодно from_api приводит год к int только если там действительно цифры, иначе оставляет None.

Статус unavailable как контракт с LLM

Главный принцип скоринга — никаких синтетических данных. Публичный REST API ГИС ЖКХ закрыт и отвечает 404; остался только SOAP с электронной подписью. Поэтому отчёт пишет про капремонт статус unavailable и не делает выводов вроде «проблем нет».

Заключение генерирует DeepSeek. Чтобы модель не придумывала «юридическую чистоту», ей передаётся детерминированный JSON с фактами и статусом каждого слоя:

{
  "object_type": "land_plot",
  "verified_layers": {
    "zouit": {"status": "available", "objects_found": 1, "type": "ЛЭП 110 кВ"},
    "flood_risk": {"status": "available", "objects_found": 0},
    "gis_zhkh": {"status": "unavailable", "reason": "REST API deprecated"}
  }
}

Системный промпт задаёт три правила:

никаких общих советов в духе «обратитесь к юристу»;
опираться только на переданные факты;
если реестр в статусе unavailable, прямо написать, какой документ и где покупателю нужно запросить самому.

Различие между objects_found: 0 и status: unavailable здесь принципиально. Первое значит «проверили, зон затопления нет», второе — «проверить не смогли». Если свести оба случая к пустому списку, модель напишет покупателю, что всё чисто, хотя данных просто не было.

Что взять в свой проект

С капризными госAPI работайте одной сессией и последовательными запросами, пока не убедились, что сервис выдерживает параллельность.
Пространственные запросы делайте по полигону объекта: точка с погрешностью 100 метров легко попадает на соседний участок.
Если есть точный идентификатор (кадастровый номер), он главнее любых координат геокодера.
Пустое поле в ответе API проверяйте на опечатки в имени, прежде чем считать данные отсутствующими; в Pydantic это решается validation_alias.
Недоступный источник отдавайте в LLM отдельным статусом с причиной, и в промпте явно описывайте, что модель должна с ним делать.