feat(sf): ingest функц.зон генплана ЕКБ-2045 из ГИСОГД-СО WFS → ekb_genplan_functional_zone (#1137)

Co-authored-by: lekss361 <lekss361@gendsgn.local>
Co-committed-by: lekss361 <lekss361@gendsgn.local>

backend/app/services/scrapers/gisogd_so_client.py

@@ -0,0 +1,214 @@
+"""ГИСОГД-СО WFS-клиент — функциональные зоны генплана ЕКБ-2045 (#1058).
+
+Источник: ``gisogd66.midural.ru`` — региональная ГИС органов государственной власти
+Свердловской области. Открытый WFS без авторизации.
+
+Два метода:
+  ``resolve_functional_zone_layer()`` — резолвит актуальное typeName слоя «Функциональные
+  зоны» из каталога слоёв. Суффикс-хеш в имени слоя меняется при обновлении генплана,
+  поэтому хардкодить нельзя — резолвим динамически. Fallback-хардкод если каталог недоступен.
+
+  ``fetch_functional_zones()`` — WFS GetFeature → список feature-dict'ов GeoJSON
+  (FeatureCollection), ~6436 фич, Polygon, EPSG:4326.
+
+TLS: midural.ru использует российский CA (по аналогии с #242 ekburg_permits). При SSLError
+используем verify=False с explicit-комментом — без этого соединение рвётся на prod VPS.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+# ── Константы ─────────────────────────────────────────────────────────────────
+
+GISOGD_BASE = "https://gisogd66.midural.ru"
+CATALOG_URL = f"{GISOGD_BASE}/api/catalogs/agate_ekbgo"
+WFS_URL = f"{GISOGD_BASE}/map/tiles/agate_ekbgo/wfs"
+
+# Fallback-хардкод если каталог недоступен (последний известный хеш, 2026-06-07).
+_LAYER_FALLBACK = "agate_ekbgo:agate_ekbgo_geo_urban_10_functional_zone_f6c5926"
+
+# Имя узла в каталоге (case-insensitive поиск).
+_CATALOG_NODE_NAME = "Функциональные зоны"
+
+# Параметры запроса к WFS.
+_WFS_OUTPUT_FORMAT = "application/json"
+_WFS_VERSION = "2.0.0"
+
+# Щедрый таймаут — ответ ~6436 фич (полный слой одним запросом, ~несколько МБ).
+_WFS_TIMEOUT = 120
+_CATALOG_TIMEOUT = 20
+
+# Размер страницы для постраничного fallback (если целиком не забирается).
+_PAGE_SIZE = 2000
+
+
+# ── Сетевые helpers (module-level — мокаются в тестах) ───────────────────────
+
+
+def _http_get_json(url: str, params: dict[str, str] | None = None, *, timeout: int) -> Any:
+    """GET → JSON. verify=False: midural.ru — российский CA (аналог #242 ekburg_permits).
+
+    Сначала пробуем verify=True; при SSLError fallback на verify=False с предупреждением.
+    """
+    try:
+        resp = httpx.get(url, params=params, timeout=timeout, verify=True)
+        resp.raise_for_status()
+        return resp.json()
+    except httpx.ConnectError as exc:
+        # SSLError обёрнут в ConnectError у httpx — переключаемся на verify=False.
+        logger.warning(
+            "gisogd_so_client: SSL verify=True failed (%s), retry with verify=False", exc
+        )
+    resp = httpx.get(url, params=params, timeout=timeout, verify=False)
+    resp.raise_for_status()
+    return resp.json()
+
+
+# ── Каталог → резолв слоя ────────────────────────────────────────────────────
+
+
+def _find_layer_in_tree(node: Any) -> str | None:
+    """Рекурсивный обход JSON-дерева каталога. Ищет узел с name==_CATALOG_NODE_NAME.
+
+    Возвращает ``value`` найденного узла (typeName слоя) или None.
+    """
+    if not isinstance(node, dict):
+        return None
+    name = node.get("name") or ""
+    if name.strip() == _CATALOG_NODE_NAME:
+        value = node.get("value")
+        return str(value) if value else None
+    # Рекурсивно по children (разные ключи в каталоге: children / items / layers).
+    for key in ("children", "items", "layers", "nodes"):
+        children = node.get(key)
+        if isinstance(children, list):
+            for child in children:
+                found = _find_layer_in_tree(child)
+                if found:
+                    return found
+    return None
+
+
+def resolve_functional_zone_layer() -> str:
+    """Резолвит actуальное typeName слоя «Функциональные зоны» из каталога ГИСОГД-СО.
+
+    Суффикс-хеш в имени слоя меняется при обновлении генплана → не хардкодить.
+    При недоступности каталога возвращает fallback-хардкод.
+
+    Returns:
+        typeName вида ``agate_ekbgo:agate_ekbgo_geo_urban_10_functional_zone_<hash>``.
+    """
+    try:
+        catalog = _http_get_json(CATALOG_URL, timeout=_CATALOG_TIMEOUT)
+    except Exception as exc:
+        logger.warning(
+            "gisogd_so_client: каталог недоступен (%s), использую fallback layer=%s",
+            exc,
+            _LAYER_FALLBACK,
+        )
+        return _LAYER_FALLBACK
+
+    layer = _find_layer_in_tree(catalog)
+    if layer:
+        logger.info("gisogd_so_client: резолвлен слой из каталога: %s", layer)
+        return layer
+
+    logger.warning(
+        "gisogd_so_client: узел '%s' не найден в каталоге, использую fallback layer=%s",
+        _CATALOG_NODE_NAME,
+        _LAYER_FALLBACK,
+    )
+    return _LAYER_FALLBACK
+
+
+# ── WFS GetFeature → список фич ──────────────────────────────────────────────
+
+
+def _wfs_params(type_name: str, start_index: int = 0, count: int | None = None) -> dict[str, str]:
+    """Строит параметры WFS GetFeature-запроса."""
+    params: dict[str, str] = {
+        "service": "WFS",
+        "version": _WFS_VERSION,
+        "request": "GetFeature",
+        "typeNames": type_name,
+        "outputFormat": _WFS_OUTPUT_FORMAT,
+    }
+    if count is not None:
+        params["count"] = str(count)
+    if start_index > 0:
+        params["startIndex"] = str(start_index)
+    return params
+
+
+def fetch_functional_zones(type_name: str | None = None) -> list[dict[str, Any]]:
+    """Загружает все фичи слоя функциональных зон из ГИСОГД-СО WFS.
+
+    Сначала пробует забрать весь слой одним запросом (~6436 фич). При таймауте
+    автоматически переключается на постраничный режим (count=_PAGE_SIZE).
+
+    Args:
+        type_name: typeName WFS-слоя. None → вызывает resolve_functional_zone_layer().
+
+    Returns:
+        Список feature-dict'ов (GeoJSON Feature): ``{"geometry": ..., "properties": ...}``.
+        Пусто если WFS недоступен или вернул 0 фич.
+    """
+    if type_name is None:
+        type_name = resolve_functional_zone_layer()
+
+    # Попытка 1: весь слой одним запросом.
+    try:
+        data = _http_get_json(WFS_URL, _wfs_params(type_name), timeout=_WFS_TIMEOUT)
+        features: list[dict[str, Any]] = (data or {}).get("features") or []
+        if features:
+            logger.info("gisogd_so_client: WFS загружено фич=%d (один запрос)", len(features))
+            return features
+    except httpx.TimeoutException:
+        logger.warning("gisogd_so_client: WFS таймаут одним запросом, переключаюсь на страницы")
+    except Exception as exc:
+        logger.error("gisogd_so_client: WFS ошибка одним запросом: %s", exc)
+        return []
+
+    # Попытка 2: постраничная загрузка.
+    features = []
+    start = 0
+    while True:
+        try:
+            data = _http_get_json(
+                WFS_URL,
+                _wfs_params(type_name, start_index=start, count=_PAGE_SIZE),
+                timeout=_WFS_TIMEOUT,
+            )
+        except Exception as exc:
+            logger.error("gisogd_so_client: WFS постраничная ошибка start=%d: %s", start, exc)
+            break
+        page: list[dict[str, Any]] = (data or {}).get("features") or []
+        if not page:
+            break
+        features.extend(page)
+        logger.info(
+            "gisogd_so_client: WFS страница start=%d, фич на странице=%d, итого=%d",
+            start,
+            len(page),
+            len(features),
+        )
+        if len(page) < _PAGE_SIZE:
+            break
+        start += _PAGE_SIZE
+
+    logger.info("gisogd_so_client: WFS итого загружено фич=%d (постраничная)", len(features))
+    return features
+
+
+__all__ = [
+    "CATALOG_URL",
+    "WFS_URL",
+    "fetch_functional_zones",
+    "resolve_functional_zone_layer",
+]

backend/app/services/site_finder/functional_zone_lookup.py

@@ -0,0 +1,68 @@
+"""Lookup функциональных зон генплана ЕКБ-2045 для analyze_parcel (#1058).
+
+Читает ``ekb_genplan_functional_zone`` (м.139) и отдаёт зоны, пересекающие участок.
+Graceful: таблица ещё не задеплоена / ошибка БД / пустой wkt → возвращает [].
+
+Analyze-врезка (вшивание в build_ird_analyze_block / analyze_parcel) — отдельный
+follow-up после merge #1131, чтобы не конфликтить с ird-analyze PR.
+
+Зеркалит ird_overlay_lookup.py (стиль, graceful handling, CAST psycopg v3).
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.exc import OperationalError, ProgrammingError
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+# Spatial overlap участка: ST_Intersects через GIST idx_ekb_genplan_fz_geom.
+# Участок передаётся как WKT в 4326. CAST psycopg v3 (НИКОГДА :x::type).
+_ZONE_OVERLAP_SQL = text(
+    """
+    SELECT zone_type, status, area_ha, zone_class_code
+    FROM ekb_genplan_functional_zone
+    WHERE ST_Intersects(geom, ST_GeomFromText(CAST(:parcel_wkt AS text), 4326))
+    ORDER BY zone_type, status
+    """
+)
+
+
+def parcel_functional_zones(db: Session, parcel_wkt: str | None) -> list[dict[str, Any]]:
+    """Функциональные зоны генплана, пересекающие участок.
+
+    Args:
+        db: SQLAlchemy-сессия.
+        parcel_wkt: геометрия участка в WKT (EPSG:4326). None → пустой результат.
+
+    Returns:
+        Список dict: [{zone_type, status, area_ha, zone_class_code}].
+        Пусто при: нет wkt / таблица не задеплоена / нет пересечений / ошибка БД.
+    """
+    if not parcel_wkt:
+        return []
+    try:
+        rows = db.execute(_ZONE_OVERLAP_SQL, {"parcel_wkt": parcel_wkt}).mappings().all()
+    except (OperationalError, ProgrammingError) as exc:
+        # Таблица ещё не задеплоена или ошибка БД — graceful degrade (analyze не падает).
+        logger.warning(
+            "parcel_functional_zones: ekb_genplan_functional_zone недоступна, skip: %s", exc
+        )
+        return []
+
+    return [
+        {
+            "zone_type": r["zone_type"],
+            "status": r["status"],
+            "area_ha": float(r["area_ha"]) if r["area_ha"] is not None else None,
+            "zone_class_code": r["zone_class_code"],
+        }
+        for r in rows
+    ]
+
+
+__all__ = ["parcel_functional_zones"]

backend/app/workers/beat_schedule.py

@@ -393,4 +393,15 @@ def build_beat_schedule() -> dict:
         "options": {"queue": "celery"},
     }
 
+    # Функциональные зоны генплана ЕКБ-2045 (#1058): ГИСОГД-СО WFS →
+    # ekb_genplan_functional_zone (м.139). ~6436 полигонов, EPSG:4326. Генплан
+    # меняется редко (раз в год+), еженедельно избыточно но дёшево — WFS-фетч +
+    # upsert по globalid идемпотентен. Понедельник 06:30 МСК — после supply-layers
+    # (06:00). Техническая, не в job_settings.
+    schedule["genplan-zones-sync-weekly"] = {
+        "task": "tasks.genplan_zones_sync.sync_functional_zones",
+        "schedule": _parse_cron("30 6 * * mon"),
+        "options": {"queue": "celery"},
+    }
+
     return schedule

backend/app/workers/celery_app.py

@@ -70,6 +70,7 @@ celery_app = Celery(
         "app.workers.tasks.planning_harvest",
         "app.workers.tasks.zone_regulation_refresh",
         "app.workers.tasks.reservation_ingest",
+        "app.workers.tasks.genplan_zones_sync",
         "app.workers.tasks.ekb_ppt_tep_sync",
     ],
 )

backend/app/workers/tasks/genplan_zones_sync.py

@@ -0,0 +1,147 @@
+"""Celery task: sync функциональных зон генплана ЕКБ-2045 из ГИСОГД-СО WFS (#1058).
+
+Загружает ~6436 полигонов функциональных зон из WFS ГИСОГД-СО и UPSERT-ит их в
+``ekb_genplan_functional_zone`` (м.139) по ключу ``globalid`` (UUID объекта).
+
+Зеркалит конвенции ird_harvest.py / supply_layers_refresh.py:
+  • SessionLocal() + try/finally close; logger (не print).
+  • SAVEPOINT per-row (``with db.begin_nested():``) — одна битая фича не валит батч.
+  • CAST(:x AS type) — НИКОГДА :x::type (psycopg v3).
+  • Геометрия: ST_SetSRID(ST_GeomFromGeoJSON, 4326) — без transform (уже 4326).
+  • ON CONFLICT (globalid) DO UPDATE: обновляет все поля кроме id/fetched_at (latest wins).
+  • commit один раз в конце.
+
+Beat: «genplan-zones-sync-weekly» (понедельник 06:30 МСК) — после supply-layers-refresh.
+Admin trigger: при необходимости ручного запуска можно сделать POST-эндпоинт.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+from datetime import UTC, datetime
+from typing import Any
+
+from sqlalchemy import text
+from sqlalchemy.orm import Session
+
+from app.core.db import SessionLocal
+from app.services.scrapers.gisogd_so_client import fetch_functional_zones
+from app.workers.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+# UPSERT одной фичи. geom = ST_SetSRID(ST_GeomFromGeoJSON, 4326) — данные уже в 4326.
+# ON CONFLICT обновляет все аналитические поля; fetched_at = excluded (latest wins).
+# CAST(:x AS type) — psycopg v3 (НИКОГДА :x::type).
+_UPSERT_SQL = text(
+    """
+    INSERT INTO ekb_genplan_functional_zone (
+        globalid, zone_type, zone_class_code, status,
+        area_ha, doc_ref, geom, raw_props, fetched_at
+    ) VALUES (
+        CAST(:globalid AS text),
+        CAST(:zone_type AS text),
+        CAST(:zone_class_code AS text),
+        CAST(:status AS text),
+        CAST(:area_ha AS numeric),
+        CAST(:doc_ref AS text),
+        ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geojson AS text)), 4326),
+        CAST(:raw_props AS jsonb),
+        CAST(:fetched_at AS timestamptz)
+    )
+    ON CONFLICT (globalid) DO UPDATE SET
+        zone_type       = EXCLUDED.zone_type,
+        zone_class_code = EXCLUDED.zone_class_code,
+        status          = EXCLUDED.status,
+        area_ha         = EXCLUDED.area_ha,
+        doc_ref         = EXCLUDED.doc_ref,
+        geom            = EXCLUDED.geom,
+        raw_props       = EXCLUDED.raw_props,
+        fetched_at      = EXCLUDED.fetched_at
+    """
+)
+
+
+def _extract_params(feature: dict[str, Any], fetched_at: str) -> dict[str, Any] | None:
+    """Извлекает параметры UPSERT из GeoJSON-фичи.
+
+    Возвращает None если фича не пригодна для записи (нет geometry или globalid).
+    """
+    geom = feature.get("geometry")
+    props: dict[str, Any] = feature.get("properties") or {}
+
+    globalid = props.get("urban10_globalid") or props.get("globalid")
+    if not geom or not globalid:
+        return None
+
+    # zone_type: приоритет full_name, fallback urban10_classid_v.
+    zone_type = props.get("full_name") or props.get("urban10_classid_v")
+    zone_class_code_raw = props.get("urban10_classid")
+    zone_class_code = str(zone_class_code_raw) if zone_class_code_raw is not None else None
+
+    area_raw = props.get("urban10_area")
+    try:
+        area_ha = float(area_raw) if area_raw is not None else None
+    except (TypeError, ValueError):
+        area_ha = None
+
+    return {
+        "globalid": str(globalid),
+        "zone_type": str(zone_type) if zone_type is not None else None,
+        "zone_class_code": zone_class_code,
+        "status": props.get("urban10_status_v"),
+        "area_ha": area_ha,
+        "doc_ref": props.get("info_set_key_ref"),
+        "geojson": json.dumps(geom),
+        "raw_props": json.dumps(props, ensure_ascii=False),
+        "fetched_at": fetched_at,
+    }
+
+
+def _upsert_feature(db: Session, params: dict[str, Any]) -> bool:
+    """UPSERT одной фичи под SAVEPOINT. True если записана, False при ошибке."""
+    try:
+        with db.begin_nested():
+            db.execute(_UPSERT_SQL, params)
+        return True
+    except Exception as exc:
+        logger.warning(
+            "genplan_zones_sync: upsert failed globalid=%s: %s",
+            params.get("globalid"),
+            exc,
+        )
+        return False
+
+
+@celery_app.task(name="tasks.genplan_zones_sync.sync_functional_zones")
+def sync_functional_zones() -> dict[str, int]:
+    """Загружает функц.зоны из ГИСОГД-СО WFS → UPSERT ekb_genplan_functional_zone.
+
+    Returns:
+        ``{"zones": N}`` — количество успешно записанных зон.
+    """
+    fetched_at = datetime.now(UTC).isoformat()
+    db = SessionLocal()
+    n_zones = 0
+    n_skip = 0
+    try:
+        logger.info("genplan_zones_sync: старт загрузки функциональных зон ГИСОГД-СО")
+        features = fetch_functional_zones()
+        logger.info("genplan_zones_sync: получено фич=%d", len(features))
+
+        for feat in features:
+            params = _extract_params(feat, fetched_at)
+            if params is None:
+                n_skip += 1
+                continue
+            if _upsert_feature(db, params):
+                n_zones += 1
+            else:
+                n_skip += 1
+
+        db.commit()
+        logger.info("genplan_zones_sync: готово, zones=%d skip=%d", n_zones, n_skip)
+        return {"zones": n_zones}
+    finally:
+        db.close()

backend/tests/services/scrapers/test_gisogd_so_client.py

@@ -0,0 +1,189 @@
+"""Тесты gisogd_so_client (#1058) — каталог-резолв + WFS fetch.
+
+Сеть не дёргается: мокаем _http_get_json на module-level.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from app.services.scrapers import gisogd_so_client
+from app.services.scrapers.gisogd_so_client import (
+    _LAYER_FALLBACK,
+    _find_layer_in_tree,
+    fetch_functional_zones,
+    resolve_functional_zone_layer,
+)
+
+# ── _find_layer_in_tree ───────────────────────────────────────────────────────
+
+_LAYER_KNOWN = "agate_ekbgo:agate_ekbgo_geo_urban_10_functional_zone_f6c5926"
+_LAYER_ABC = "agate_ekbgo:agate_ekbgo_geo_urban_10_functional_zone_abc123"
+_LAYER_XYZ = "agate_ekbgo:agate_ekbgo_geo_urban_10_functional_zone_xyz789"
+
+
+def test_find_layer_direct_match() -> None:
+    """Узел с name == 'Функциональные зоны' → возвращает value."""
+    node = {"name": "Функциональные зоны", "value": _LAYER_KNOWN}
+    result = _find_layer_in_tree(node)
+    assert result == _LAYER_KNOWN
+
+
+def test_find_layer_in_nested_children() -> None:
+    """Вложенный узел в children → резолвится рекурсивно."""
+    catalog = {
+        "name": "root",
+        "children": [
+            {
+                "name": "Зонирование",
+                "children": [
+                    {"name": "Функциональные зоны", "value": _LAYER_ABC},
+                ],
+            },
+        ],
+    }
+    result = _find_layer_in_tree(catalog)
+    assert result == _LAYER_ABC
+
+
+def test_find_layer_not_found_returns_none() -> None:
+    """Узел не найден → None."""
+    catalog = {"name": "Другой слой", "value": "some:layer", "children": []}
+    result = _find_layer_in_tree(catalog)
+    assert result is None
+
+
+def test_find_layer_non_dict_node_returns_none() -> None:
+    """Нода не dict → None (безопасный обход)."""
+    assert _find_layer_in_tree(None) is None  # type: ignore[arg-type]
+    assert _find_layer_in_tree("string") is None  # type: ignore[arg-type]
+    assert _find_layer_in_tree([]) is None  # type: ignore[arg-type]
+
+
+# ── resolve_functional_zone_layer ─────────────────────────────────────────────
+
+
+def test_resolve_layer_from_catalog(monkeypatch: Any) -> None:
+    """Каталог доступен → корректный layer из дерева."""
+    catalog = {
+        "name": "root",
+        "children": [
+            {"name": "Функциональные зоны", "value": _LAYER_XYZ},
+        ],
+    }
+    monkeypatch.setattr(gisogd_so_client, "_http_get_json", lambda *a, **kw: catalog)
+
+    result = resolve_functional_zone_layer()
+    assert result == _LAYER_XYZ
+
+
+def test_resolve_layer_fallback_on_catalog_error(monkeypatch: Any) -> None:
+    """Каталог недоступен → fallback-хардкод."""
+
+    def _raise(*a: Any, **kw: Any) -> None:
+        raise Exception("connection refused")
+
+    monkeypatch.setattr(gisogd_so_client, "_http_get_json", _raise)
+
+    result = resolve_functional_zone_layer()
+    assert result == _LAYER_FALLBACK
+
+
+def test_resolve_layer_fallback_node_not_found(monkeypatch: Any) -> None:
+    """Каталог пришёл но нужного узла нет → fallback-хардкод."""
+    catalog = {
+        "name": "root",
+        "children": [{"name": "Другой слой", "value": "some:other:layer"}],
+    }
+    monkeypatch.setattr(gisogd_so_client, "_http_get_json", lambda *a, **kw: catalog)
+
+    result = resolve_functional_zone_layer()
+    assert result == _LAYER_FALLBACK
+
+
+# ── fetch_functional_zones ────────────────────────────────────────────────────
+
+
+def _make_fc(n: int) -> dict[str, Any]:
+    """FeatureCollection с n фичами."""
+    return {
+        "type": "FeatureCollection",
+        "features": [
+            {
+                "geometry": {"type": "Polygon", "coordinates": []},
+                "properties": {"urban10_globalid": f"uuid-{i}"},
+            }
+            for i in range(n)
+        ],
+    }
+
+
+def test_fetch_returns_features_single_request(monkeypatch: Any) -> None:
+    """Успешный одиночный WFS-запрос → список фич."""
+    fc = _make_fc(6436)
+    calls: list[dict[str, Any]] = []
+
+    def _mock_get(url: str, params: dict[str, str] | None = None, **kw: Any) -> Any:
+        calls.append({"url": url, "params": params})
+        return fc
+
+    monkeypatch.setattr(gisogd_so_client, "_http_get_json", _mock_get)
+
+    features = fetch_functional_zones("agate_ekbgo:test_layer")
+    assert len(features) == 6436
+    assert len(calls) == 1  # один запрос
+    assert calls[0]["params"]["typeNames"] == "agate_ekbgo:test_layer"
+
+
+def test_fetch_pagination_on_timeout(monkeypatch: Any) -> None:
+    """Первый запрос — TimeoutException → постраничная загрузка."""
+    import httpx
+
+    first_call = True
+
+    def _mock_get(url: str, params: dict[str, str] | None = None, **kw: Any) -> Any:
+        nonlocal first_call
+        if first_call:
+            first_call = False
+            raise httpx.TimeoutException("timeout")
+        # Постраничные: 3 фичи на первой странице, пустая на второй.
+        start = int((params or {}).get("startIndex", 0))
+        if start == 0:
+            return _make_fc(3)
+        return {"type": "FeatureCollection", "features": []}
+
+    monkeypatch.setattr(gisogd_so_client, "_http_get_json", _mock_get)
+
+    features = fetch_functional_zones("agate_ekbgo:test_layer")
+    assert len(features) == 3
+
+
+def test_fetch_empty_on_wfs_error(monkeypatch: Any) -> None:
+    """WFS не отвечает → пустой список, без падения."""
+
+    def _raise(*a: Any, **kw: Any) -> None:
+        raise Exception("connection error")
+
+    monkeypatch.setattr(gisogd_so_client, "_http_get_json", _raise)
+
+    features = fetch_functional_zones("agate_ekbgo:test_layer")
+    assert features == []
+
+
+def test_fetch_resolves_layer_when_none(monkeypatch: Any) -> None:
+    """type_name=None → вызывается resolve_functional_zone_layer перед WFS."""
+    resolved: list[str] = []
+
+    def _mock_resolve() -> str:
+        resolved.append("called")
+        return "agate_ekbgo:resolved_layer"
+
+    def _mock_get(url: str, params: dict[str, str] | None = None, **kw: Any) -> Any:
+        return _make_fc(2)
+
+    monkeypatch.setattr(gisogd_so_client, "resolve_functional_zone_layer", _mock_resolve)
+    monkeypatch.setattr(gisogd_so_client, "_http_get_json", _mock_get)
+
+    features = fetch_functional_zones(type_name=None)
+    assert resolved == ["called"]
+    assert len(features) == 2

backend/tests/services/test_functional_zone_lookup.py

@@ -0,0 +1,98 @@
+"""Тесты parcel_functional_zones (#1058) — read-side lookup функциональных зон генплана."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy.exc import OperationalError, ProgrammingError
+
+from app.services.site_finder.functional_zone_lookup import parcel_functional_zones
+
+
+class _Result:
+    def __init__(self, rows: list[dict[str, Any]]) -> None:
+        self._rows = rows
+
+    def mappings(self) -> _Result:
+        return self
+
+    def all(self) -> list[dict[str, Any]]:
+        return self._rows
+
+
+class _FakeDB:
+    def __init__(
+        self,
+        rows: list[dict[str, Any]] | None = None,
+        raise_exc: Exception | None = None,
+    ) -> None:
+        self._rows = rows or []
+        self._raise = raise_exc
+
+    def execute(self, sql: Any, params: dict[str, Any] | None = None) -> _Result:
+        if self._raise is not None:
+            raise self._raise
+        return _Result(self._rows)
+
+
+_WKT = "POLYGON((60.5 56.8, 60.6 56.8, 60.6 56.9, 60.5 56.9, 60.5 56.8))"
+
+
+def test_returns_zones_for_parcel() -> None:
+    """Обычный случай: строки из БД → список dict с нужными ключами."""
+    rows = [
+        {
+            "zone_type": "Жилые зоны",
+            "status": "Существующий",
+            "area_ha": 15.3,
+            "zone_class_code": "Ж",
+        },
+        {
+            "zone_type": "Общественные зоны",
+            "status": "Планируемый",
+            "area_ha": None,
+            "zone_class_code": "О",
+        },
+    ]
+    result = parcel_functional_zones(_FakeDB(rows), _WKT)
+
+    assert len(result) == 2
+    first = result[0]
+    assert first["zone_type"] == "Жилые зоны"
+    assert first["status"] == "Существующий"
+    assert first["area_ha"] == 15.3
+    assert first["zone_class_code"] == "Ж"
+    second = result[1]
+    assert second["area_ha"] is None  # NULL → None (не падает)
+
+
+def test_empty_when_no_wkt() -> None:
+    """parcel_wkt=None → пустой список без запроса к БД."""
+    result = parcel_functional_zones(_FakeDB([{"zone_type": "Жилые зоны"}]), None)
+    assert result == []
+
+
+def test_empty_when_no_overlap() -> None:
+    """БД вернула 0 строк → пустой список."""
+    result = parcel_functional_zones(_FakeDB([]), _WKT)
+    assert result == []
+
+
+def test_graceful_when_table_missing_operational_error() -> None:
+    """OperationalError (таблица не задеплоена) → [] без падения."""
+    db = _FakeDB(raise_exc=OperationalError("stmt", {}, Exception("no such table")))
+    result = parcel_functional_zones(db, _WKT)
+    assert result == []
+
+
+def test_graceful_when_table_missing_programming_error() -> None:
+    """ProgrammingError (e.g. relation не существует) → [] без падения."""
+    db = _FakeDB(raise_exc=ProgrammingError("stmt", {}, Exception("relation does not exist")))
+    result = parcel_functional_zones(db, _WKT)
+    assert result == []
+
+
+def test_empty_string_wkt_treated_as_falsy() -> None:
+    """Пустая строка wkt → пустой список (falsy guard)."""
+    result = parcel_functional_zones(_FakeDB([]), "")
+    assert result == []

backend/tests/workers/test_genplan_zones_sync.py

@@ -0,0 +1,201 @@
+"""Тесты sync_functional_zones (#1058) — ГИСОГД-СО WFS → ekb_genplan_functional_zone.
+
+Сеть/БД не дёргаются: мокаем fetch_functional_zones + SessionLocal, проверяем
+маппинг properties → колонки, globalid-дедуп, upsert-идемпотентность.
+"""
+
+from __future__ import annotations
+
+from contextlib import contextmanager
+from typing import Any
+
+from app.workers.tasks import genplan_zones_sync
+
+# ── Фейковые данные ──────────────────────────────────────────────────────────
+
+
+def _make_feature(
+    globalid: str = "aaaa-1111",
+    full_name: str | None = "Жилые зоны",
+    classid: str | None = "Ж",
+    status: str | None = "Существующий",
+    area_ha: float | None = 12.5,
+    doc_ref: str | None = "REF-001",
+) -> dict[str, Any]:
+    return {
+        "geometry": {
+            "type": "Polygon",
+            "coordinates": [[[60.5, 56.8], [60.6, 56.8], [60.5, 56.8]]],
+        },
+        "properties": {
+            "urban10_globalid": globalid,
+            "full_name": full_name,
+            "urban10_classid_v": "Жилые зоны (fallback)",
+            "urban10_classid": classid,
+            "urban10_status_v": status,
+            "urban10_area": area_ha,
+            "info_set_key_ref": doc_ref,
+        },
+    }
+
+
+# ── Фейковая БД ──────────────────────────────────────────────────────────────
+
+
+class _FakeDB:
+    def __init__(self) -> None:
+        self.executed: list[tuple[Any, dict[str, Any] | None]] = []
+        self.committed = False
+        self.closed = False
+
+    @contextmanager
+    def begin_nested(self) -> Any:
+        yield
+
+    def execute(self, sql: Any, params: dict[str, Any] | None = None) -> Any:
+        self.executed.append((sql, params))
+        return None
+
+    def commit(self) -> None:
+        self.committed = True
+
+    def close(self) -> None:
+        self.closed = True
+
+
+# ── Тесты ────────────────────────────────────────────────────────────────────
+
+
+def test_sync_maps_properties_to_columns(monkeypatch: Any) -> None:
+    """Проверяем маппинг properties → параметры UPSERT."""
+    feat = _make_feature(
+        globalid="uuid-abc",
+        full_name="Жилые зоны",
+        classid="Ж",
+        status="Существующий",
+        area_ha=10.5,
+        doc_ref="DOC-777",
+    )
+    db = _FakeDB()
+    monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: [feat])
+    monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
+
+    result = genplan_zones_sync.sync_functional_zones()
+
+    assert result == {"zones": 1}
+    assert db.committed is True
+    assert db.closed is True
+    assert len(db.executed) == 1
+    _, params = db.executed[0]
+    assert params is not None
+    assert params["globalid"] == "uuid-abc"
+    assert params["zone_type"] == "Жилые зоны"
+    assert params["zone_class_code"] == "Ж"
+    assert params["status"] == "Существующий"
+    assert params["area_ha"] == 10.5
+    assert params["doc_ref"] == "DOC-777"
+    assert isinstance(params["geojson"], str)
+    assert isinstance(params["raw_props"], str)
+
+
+def test_sync_fallback_to_classid_v_when_no_full_name(monkeypatch: Any) -> None:
+    """full_name=None → zone_type берётся из urban10_classid_v."""
+    feat = _make_feature(full_name=None)
+    db = _FakeDB()
+    monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: [feat])
+    monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
+
+    genplan_zones_sync.sync_functional_zones()
+
+    _, params = db.executed[0]
+    assert params is not None
+    assert params["zone_type"] == "Жилые зоны (fallback)"
+
+
+def test_sync_skips_feature_without_globalid(monkeypatch: Any) -> None:
+    """Фича без globalid → пропускается (defensive), не вызывает UPSERT."""
+    feat: dict[str, Any] = {
+        "geometry": {"type": "Polygon", "coordinates": []},
+        "properties": {"full_name": "Жилые зоны"},  # нет urban10_globalid
+    }
+    db = _FakeDB()
+    monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: [feat])
+    monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
+
+    result = genplan_zones_sync.sync_functional_zones()
+
+    assert result == {"zones": 0}
+    assert db.executed == []
+    assert db.committed is True
+
+
+def test_sync_skips_feature_without_geometry(monkeypatch: Any) -> None:
+    """Фича без geometry → пропускается."""
+    feat: dict[str, Any] = {
+        "geometry": None,
+        "properties": {
+            "urban10_globalid": "uuid-xyz",
+            "full_name": "Производственные зоны",
+        },
+    }
+    db = _FakeDB()
+    monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: [feat])
+    monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
+
+    result = genplan_zones_sync.sync_functional_zones()
+
+    assert result == {"zones": 0}
+    assert db.executed == []
+
+
+def test_sync_globalid_dedup_two_identical_features(monkeypatch: Any) -> None:
+    """Две фичи с одинаковым globalid — оба UPSERT (ON CONFLICT дедуп на уровне БД)."""
+    feat = _make_feature(globalid="dup-uuid")
+    db = _FakeDB()
+    monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: [feat, feat])
+    monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
+
+    result = genplan_zones_sync.sync_functional_zones()
+
+    # Оба вызова UPSERT прошли (ON CONFLICT DO UPDATE разрулит на уровне БД).
+    assert result == {"zones": 2}
+    assert len(db.executed) == 2
+
+
+def test_sync_savepoint_per_row_error_does_not_abort_batch(monkeypatch: Any) -> None:
+    """SAVEPOINT per-row: ошибка в одной фиче не валит батч остальных."""
+    call_count = 0
+
+    class _FaultyDB(_FakeDB):
+        @contextmanager
+        def begin_nested(self) -> Any:
+            nonlocal call_count
+            call_count += 1
+            if call_count == 1:
+                raise Exception("upsert error row 1")
+            yield
+
+    features = [_make_feature(globalid=f"uuid-{i}") for i in range(3)]
+    db = _FaultyDB()
+    monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: features)
+    monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
+
+    result = genplan_zones_sync.sync_functional_zones()
+
+    # Строка 1 упала (savepoint), строки 2 и 3 записались.
+    assert result == {"zones": 2}
+    assert db.committed is True
+    assert db.closed is True
+
+
+def test_sync_empty_wfs_response(monkeypatch: Any) -> None:
+    """WFS вернул 0 фич → zones=0, commit без ошибок."""
+    db = _FakeDB()
+    monkeypatch.setattr(genplan_zones_sync, "fetch_functional_zones", lambda: [])
+    monkeypatch.setattr(genplan_zones_sync, "SessionLocal", lambda: db)
+
+    result = genplan_zones_sync.sync_functional_zones()
+
+    assert result == {"zones": 0}
+    assert db.committed is True
+    assert db.closed is True

data/sql/139_ekb_genplan_functional_zone.sql

@@ -0,0 +1,72 @@
+-- 139_ekb_genplan_functional_zone.sql
+-- #1058 / #1067 (Site Finder v2, «GG-форсайт») — функциональные зоны генплана ЕКБ-2045.
+--
+-- Источник: ГИСОГД-СО WFS (gisogd66.midural.ru), слой
+--   agate_ekbgo:agate_ekbgo_geo_urban_10_functional_zone_<hash>.
+-- Harvest-воркер: tasks/genplan_zones_sync.py (Celery beat «genplan-zones-sync-weekly»).
+-- Read-side: app/services/site_finder/functional_zone_lookup.py → parcel_functional_zones().
+-- Analyze-врезка: follow-up после merge #1131 (чтобы не конфликтить с ird-analyze PR).
+--
+-- Grain: одна строка = одна зона (Polygon/MultiPolygon). Дедуп-ключ — globalid (UUID из ГИСОГД).
+--
+-- Deploy: auto-applied deploy.yml через _schema_migrations (ровно один раз по NN). Idempotent:
+--   CREATE TABLE/INDEX IF NOT EXISTS + guarded DO-блок для UNIQUE (зеркаль м.132, м.125).
+--   НИКАКИХ строк здесь не вставляется — данные приходят из harvest-воркера.
+
+BEGIN;
+
+-- ── 1. Таблица ekb_genplan_functional_zone ────────────────────────────────────
+CREATE TABLE IF NOT EXISTS ekb_genplan_functional_zone (
+    id            BIGSERIAL PRIMARY KEY,
+    globalid      TEXT NOT NULL,              -- UUID из ГИСОГД (дедуп-ключ, UNIQUE ниже)
+    zone_type     TEXT,                        -- читаемый тип зоны (full_name / urban10_classid_v)
+    zone_class_code TEXT,                      -- числовой код класса (urban10_classid)
+    status        TEXT,                        -- «Существующий» / «Планируемый» (urban10_status_v)
+    area_ha       NUMERIC,                     -- площадь, га (urban10_area)
+    doc_ref       TEXT,                        -- ссылка на документ генплана (info_set_key_ref)
+    geom          geometry(Geometry, 4326) NOT NULL,  -- Polygon/MultiPolygon, EPSG:4326
+    raw_props     JSONB,                       -- полный properties фичи (forward-compat / debug)
+    fetched_at    timestamptz NOT NULL DEFAULT now()
+);
+
+COMMENT ON TABLE ekb_genplan_functional_zone IS
+    'Функциональные зоны генплана ЕКБ-2045 (#1058, GG-форсайт). '
+    'Источник: ГИСОГД-СО WFS (gisogd66.midural.ru). '
+    'UPSERT harvest-воркером tasks/genplan_zones_sync.py по globalid. '
+    'analyze читает overlap участка через ST_Intersects(geom). Пустая = 0 overlap.';
+
+COMMENT ON COLUMN ekb_genplan_functional_zone.globalid IS
+    'UUID объекта из ГИСОГД-СО (urban10_globalid); дедуп-ключ UPSERT';
+COMMENT ON COLUMN ekb_genplan_functional_zone.zone_type IS
+    'Читаемый тип зоны: full_name или urban10_classid_v («Жилые зоны» и т.д.)';
+COMMENT ON COLUMN ekb_genplan_functional_zone.zone_class_code IS
+    'Числовой код класса зоны (urban10_classid)';
+
+-- ── 2. UNIQUE по globalid (дедуп для UPSERT) ─────────────────────────────────
+-- Guard через pg_constraint — idempotent re-apply (зеркаль м.132 / м.125 / м.110).
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1 FROM pg_constraint WHERE conname = 'uq_ekb_genplan_functional_zone_globalid'
+    ) THEN
+        ALTER TABLE ekb_genplan_functional_zone
+            ADD CONSTRAINT uq_ekb_genplan_functional_zone_globalid
+            UNIQUE (globalid);
+    END IF;
+END
+$$;
+
+-- ── 3. Индексы ────────────────────────────────────────────────────────────────
+-- Read-path: ST_Intersects(parcel.geom, zone.geom) через GIST.
+CREATE INDEX IF NOT EXISTS idx_ekb_genplan_fz_geom
+    ON ekb_genplan_functional_zone USING GIST (geom);
+
+-- Фильтрация по типу зоны (выборка «все жилые» / «все производственные»).
+CREATE INDEX IF NOT EXISTS idx_ekb_genplan_fz_zone_type
+    ON ekb_genplan_functional_zone (zone_type);
+
+-- Фильтрация по статусу (существующие vs планируемые).
+CREATE INDEX IF NOT EXISTS idx_ekb_genplan_fz_status
+    ON ekb_genplan_functional_zone (status) WHERE status IS NOT NULL;
+
+COMMIT;