Архитектура пространственного аудита недвижимости: как подружить FastAPI, НСПД, OSM и не словить 403 на госсайтах

—

от автора

Привет, Хабр!

Последние полгода я в одиночку разрабатываю пет-проект НедвижСкан — сервис экспресс-проверки недвижимости перед сделкой.

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

пример

пример

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

Ниже расскажу, как устроен наш пайплайн на FastAPI, какие костыли пришлось изобрести для нормализации кадастровых данных и как заставить разнородные геосервисы отдавать честный статус без синтетики.

┌────────────────────────────────────────────────────────────────────────┐│                        ПАЙПЛАЙН АУДИТА ОБЪЕКТА                         │└────────────────────────────────────────────────────────────────────────┘                                    │                                    ▼                      [ Нормализация адреса ]                                    │                      Кадастровый номер найден?                                   / \                            ДА    /   \   НЕТ                                 /     \                                ▼       ▼                    [ Прямой поиск ]   [ Поиск по маске адреса ]                    [ в НСПД find  ]   [   в реестре ОКС НСПД  ]                                \       /                                 \     /                                  ▼   ▼                      [ Фиксация кадастрового контура ]                                    │               ┌────────────────────┼────────────────────┐               ▼                    ▼                    ▼     [ Зоны ЗОУИТ НСПД ]   [ Граф Overpass OSM ]   [ Рельеф Copernicus ]     (search_in_contour)   (трассы, ЖД, свалки)     (уклон, затопление)               │                    │                    │               └────────────────────┼────────────────────┘                                    │                                    ▼                         [ Скоринг Risk Engine ]                                    │                                    ▼                      [ LLM-синтез (DeepSeek v4) ]                                    │                                    ▼                        [ Адаптивный WeasyPrint PDF ]

Проблема 1. WAF, pynspd и иллюзия параллельности

Для работы со слоями Национальной системы пространственных данных (НСПД) и Публичной кадастровой карты мы используем клиент на базе библиотеки pynspd.

Кажется логичным: поднимаем asyncio.gather(), параллельно запрашиваем карточку здания, выписку по участку и пространственные слои ЗОУИТ. Но тут нас встречает суровая реальность.

  1. WMS GetFeatureInfo по координатам режется намертво. Если слать запросы к /api/aeggis/v3/.../wms по точке (search_at_coords), госсервер возвращает HTTP 403 BlockedIP даже при запросе с российского дата-центра Ростелекома.

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

Решение — строгий последовательный пайплайн в рамках единого асинхронного контекста:

# app/pipeline/nspd_engine.pyasync 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 полигона,        # а не через точечный identify по координатам        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}

Запросы зон выполняются только через полигональное пересечение контура, а не по координатам точки. Это убирает ложные срабатывания и обходит блокировки WAF.

Проблема 2. Ловушка геокодинга: когда дом улетает на 100 метров в канаву

Второй подвох поджидал в связке кодера и кадастра.

Кодер Suggestions отлично парсит опечатки в адресах, но координаты в частном секторе (СНТ, деревни, ИЖС) имеют погрешность до 50–100 метров.

На живом тесте в Твери кодер поставил маркер дома в соседний кадастровый квартал прямо на полотно автодороги. В итоге сервис нашел кадастровый номер дорожного полотна вместо дома клиента.

Пришлось внедрить жесткую иерархию приоритетов:

  1. Если пользователь ввел кадастровый номер — координаты геокодера игнорируются, полигон и центроид берутся строго из ЕГРН/НСПД.

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

  3. Координаты геокодера используются исключительно как аварийный fallback.

# Нормализация адреса и отсечение мусора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(",")

Проблема 3. Поле-призрак year_commisioning

Когда мы парсили паспорт многоквартирного дома из карточки здания, мы обнаружили, что поле year_built в официальных ответах стабильно приходит пустым ("" или null).

Оказалось, что разработчики API сохранили год постройки в поле с опечаткой: year_commisioning (с одной s).

Если валидировать схему через жесткий Pydantic без алиасов, сервис решает, что год постройки неизвестен. Решилось простым валидатором:

from pydantic import BaseModel, Fieldclass 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        )

Проблема 4. Манифест честности данных и работа с 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 — явно указать, какой документ и где покупателю нужно затребовать руками.

Что под капотом проекта

  • Бэкенд: Python 3.12, FastAPI, Pydantic v2, asyncio.

  • Геопространственный анализ: shapely, pyproj, Overpass API (OpenStreetMap) для оценки шумового загрязнения и транспортных артерий, Copernicus GLO-90 для построения уклонов и высот.

  • Генерация документов: HTML-шаблонизация Jinja2 + WeasyPrint для сборки 9-страничного PDF в типографском качестве.

  • Инфраструктура: Docker Compose, обратный прокси Caddy с автомонтированием Let’s Encrypt сертификатов на Beget VPS.

Проект доступен по адресу nedvizhscan.ru.

Коллеги, кто работает с геоинформационными сервисами и кадастром в РФ — с какими неочевидными граблями в реестрах сталкивались вы? Буду рад обсудить в комментариях архитектурные решения и особенности парсинга пространственных слоев.

ссылка на оригинал статьи https://habr.com/ru/articles/1097324/