feat(sf): gknspecial_zone ИРД-harvest → ird_overlays (#1114)

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

backend/app/services/scrapers/ekb_geoportal_client.py

@@ -299,6 +299,21 @@ class EKBGeoportalClient:
         card = self._urban_card(terzone_key, keys.get("GknParcel"))
         return parse_zone_regulation(card)
 
+    def features_by_cql(self, layer: str, cql_filter: str) -> list[EKBFeature]:
+        """Фичи слоя по произвольному CQL-фильтру (напр. ``full_name LIKE 'X%'``).
+
+        Тонкая обёртка над ``_resolve_type_name`` + ``_get_feature``. Используется
+        для harvest ЗОУИТ-слоя по префиксу full_name (#1063/#1064/#1065/#1066).
+
+        Args:
+            layer: ключ WFS_LAYERS ('zouit', ...) или сырой typeName.
+            cql_filter: CQL-выражение (WFS 2.0 OGC CQL). Кириллица в одиночных
+                LIKE-выражениях работает. Составные AND с кириллицей могут давать
+                HTTP-ошибку геопортала — фильтруй client-side.
+        """
+        type_name = self._resolve_type_name(layer)
+        return self._get_feature(type_name, cql_filter)
+
 
 __all__ = [
     "EKB_GEOPORTAL_BASE",

backend/app/services/site_finder/ird_overlay_lookup.py

@@ -29,10 +29,11 @@ IRD_EMPTY_RESULT: dict[str, Any] = {"ird_overlaps": [], "ird_by_kind": {}}
 
 # ИРД-overlay по участку: ST_Intersects через GIST idx_ird_overlays_geom. geom уже EPSG:4326
 # (harvest стор ST_Transform 3857→4326). Участок передаётся как WKT в 4326. CAST psycopg v3.
+# content добавлен для геопортальных ЗОУИТ-слоёв (gknspecial_harvest #1063/#1064/#1065/#1066).
 _IRD_OVERLAP_SQL = text(
     """
     SELECT layer_kind, reg_numb_border, category, subcategory,
-           zone_index, zone_type_name
+           zone_index, zone_type_name, content
     FROM ird_overlays
     WHERE ST_Intersects(geom, ST_GeomFromText(CAST(:parcel_wkt AS text), 4326))
     ORDER BY layer_kind, reg_numb_border
@@ -49,7 +50,9 @@ def parcel_ird_overlaps(db: Session, parcel_wkt: str | None) -> dict[str, Any]:
 
     Returns:
         {'ird_overlaps': [{layer_kind, reg_numb_border, category, subcategory,
-        zone_index, zone_type_name}], 'ird_by_kind': {layer_kind: count}}.
+        zone_index, zone_type_name, content}], 'ird_by_kind': {layer_kind: count}}.
+        content — JSONB-dict с ограничениями/реквизитами зоны (геопортальные слои:
+        gknspecial_harvest #1063/#1064/#1065/#1066); None для НСПД-слоёв до резолва D9.
         Пусто при отсутствии участка/таблицы/пересечений (analyze не падает).
     """
     if not parcel_wkt:
@@ -73,6 +76,7 @@ def parcel_ird_overlaps(db: Session, parcel_wkt: str | None) -> dict[str, Any]:
                 "subcategory": r["subcategory"],
                 "zone_index": r["zone_index"],
                 "zone_type_name": r["zone_type_name"],
+                "content": r["content"],
             }
         )
         by_kind[kind] = by_kind.get(kind, 0) + 1

backend/app/workers/beat_schedule.py

@@ -332,6 +332,16 @@ def build_beat_schedule() -> dict:
         "options": {"queue": "celery"},
     }
 
+    # ЗОУИТ-harvest ЕКБ-геопортала (#1063/#1064/#1065/#1066, #1067 GG-форсайт): WFS-фетч
+    # gknspecial_zone по full_name-префиксам → ird_overlays (layer_kind=servitude/water_*/
+    # airfield/okn_protection_zone/zrz). Понедельник 05:30 МСК — после ird-harvest (05:00),
+    # до supply-layers (06:00). Геометрия зон стабильна, еженедельно достаточно.
+    schedule["gknspecial-harvest-weekly"] = {
+        "task": "tasks.gknspecial_harvest.harvest_gknspecial_zones",
+        "schedule": _parse_cron("30 5 * * mon"),
+        "options": {"queue": "celery"},
+    }
+
     # Opportunity-ЗУ harvest (#1086): grid-walk НСПД opportunity-слоёв по кварталам ЕКБ →
     # ird_overlays (layer_kind='opportunity_*'). 1-е число месяца 05:30 МСК — реже ИРД (weekly),
     # opportunity-земля меняется медленнее ограничений. Тот же движок, что ird-harvest.

backend/app/workers/celery_app.py

@@ -64,6 +64,7 @@ celery_app = Celery(
         "app.workers.tasks.supply_layers_refresh",
         "app.workers.tasks.forecast",
         "app.workers.tasks.ird_harvest",
+        "app.workers.tasks.gknspecial_harvest",
         "app.workers.tasks.opportunity_harvest",
         "app.workers.tasks.planning_harvest",
         "app.workers.tasks.zone_regulation_refresh",

backend/app/workers/tasks/gknspecial_harvest.py

@@ -0,0 +1,287 @@
+"""Celery task: harvest ЗОУИТ ЕКБ-геопортала (gknspecial_zone) → ``ird_overlays`` (м.132).
+
+#1067 GG-форсайт: закрывает ИРД-пункты:
+  #1063 — сервитут (Зона публичного сервитута)
+  #1064 — водоохранные зоны (Водоохранная зона / Прибрежная защитная полоса /
+           Зона санитарной охраны / Санитарно-защитная зона)
+  #1065 — приаэродромная территория (Приаэродромная территория — ЕГРН)
+  #1066 — ОКН-зоны/ЗРЗ (Зона охраны объекта культурного / Зона регулирования застройки)
+
+Источник: ЕКБ-геопортал, WFS-слой ``portal:portal_geo_gknspecial_zone`` (ключ 'zouit').
+Геометрия уже в EPSG:4326 — ST_Transform НЕ нужен.
+
+Sentinel source_layer_id'ы (вне диапазона НСПД): 990001–990008 + 990099 (unmatched).
+
+Conventions (зеркало ird_harvest.py):
+  • SessionLocal() + try/finally close.
+  • SAVEPOINT per-row (with db.begin_nested():) — одна битая фича не валит батч.
+  • CAST(:x AS type) — НИКОГДА :x::type (psycopg v3).
+  • ON CONFLICT (source_layer_id, geom_data_id) DO UPDATE.
+  • content_* пишем мы же (геопортал как источник); content НЕ трогается НСПД-harvest'ом.
+  • Кириллица в combined-CQL WFS ломает запрос → фильтр zone_state_v client-side.
+
+Расписание: beat_schedule.py → gknspecial-harvest-weekly (пн 05:30 МСК).
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import re
+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.ekb_geoportal_client import EKBGeoportalClient
+from app.workers.celery_app import celery_app
+
+logger = logging.getLogger(__name__)
+
+# ── Mapping: full_name-префикс → (layer_kind, sentinel source_layer_id) ─────
+# Sentinels вне диапазона НСПД (990001+), чтобы не конфликтовать с реальными layer_id.
+_PREFIX_MAP: list[tuple[str, str, int]] = [
+    ("Зона публичного сервитута", "servitude", 990001),  # #1063
+    ("Водоохранная зона", "water_protection", 990002),  # #1064
+    ("Прибрежная защитная полоса", "coastal_strip", 990003),  # #1064
+    ("Зона санитарной охраны", "water_sanitary", 990004),  # #1064
+    ("Санитарно-защитная зона", "sanitary_protection", 990005),  # #1064
+    ("Зона охраны объекта культурного", "okn_protection_zone", 990006),  # #1066
+    ("Зона регулирования застройки", "zrz", 990007),  # #1066
+    ("Приаэродромная территория", "airfield", 990008),  # #1065
+]
+
+# Sentinel для фич, чей full_name не попал ни под один префикс (defensive).
+_UNMATCHED_LAYER_KIND = "gknspecial_unmatched"
+_UNMATCHED_SOURCE_LAYER_ID = 990099
+
+# Regex для извлечения reg_numb_border из full_name (формат «66:41-6.4125»).
+_REG_NUMB_RE = re.compile(r"\d{2}:\d{2}-\d+\.\d+")
+
+# UPSERT: геометрия уже EPSG:4326 → ST_SetSRID(ST_GeomFromGeoJSON, 4326), без ST_Transform.
+# content_* поля обновляются в DO UPDATE (геопортал — их источник).
+_UPSERT_SQL = text(
+    """
+    INSERT INTO ird_overlays (
+        source_layer_id, layer_kind, geom_data_id, reg_numb_border,
+        category, subcategory, geom, raw_props,
+        zone_type_name, content, content_source, content_fetched_at, fetched_at
+    ) VALUES (
+        CAST(:source_layer_id AS integer),
+        CAST(:layer_kind AS text),
+        CAST(:geom_data_id AS bigint),
+        CAST(:reg_numb_border AS text),
+        CAST(:category AS integer),
+        CAST(:subcategory AS integer),
+        ST_SetSRID(ST_GeomFromGeoJSON(CAST(:geojson AS text)), 4326),
+        CAST(:raw_props AS jsonb),
+        CAST(:zone_type_name AS text),
+        CAST(:content AS jsonb),
+        CAST(:content_source AS text),
+        CAST(:content_fetched_at AS timestamptz),
+        CAST(:fetched_at AS timestamptz)
+    )
+    ON CONFLICT (source_layer_id, geom_data_id) DO UPDATE SET
+        layer_kind          = EXCLUDED.layer_kind,
+        reg_numb_border     = EXCLUDED.reg_numb_border,
+        geom                = EXCLUDED.geom,
+        raw_props           = EXCLUDED.raw_props,
+        zone_type_name      = EXCLUDED.zone_type_name,
+        content             = EXCLUDED.content,
+        content_source      = EXCLUDED.content_source,
+        content_fetched_at  = EXCLUDED.content_fetched_at,
+        fetched_at          = EXCLUDED.fetched_at
+    """
+)
+
+
+def _classify_full_name(full_name: str) -> tuple[str, int] | None:
+    """Сопоставить full_name с записью _PREFIX_MAP → (layer_kind, source_layer_id) или None."""
+    for prefix, layer_kind, source_layer_id in _PREFIX_MAP:
+        if full_name.startswith(prefix):
+            return layer_kind, source_layer_id
+    return None
+
+
+def _extract_reg_numb(full_name: str) -> str | None:
+    """Извлечь reg_numb_border из full_name по паттерну \\d{2}:\\d{2}-\\d+\\.\\d+."""
+    m = _REG_NUMB_RE.search(full_name)
+    return m.group(0) if m else None
+
+
+def _extract_geom_data_id(feature_id: str | None) -> int | None:
+    """Trailing int из WFS feature_id (формат 'portal_geo_gknspecial_zone.NNNN') или None."""
+    if not feature_id:
+        return None
+    parts = feature_id.split(".")
+    if len(parts) < 2:
+        return None
+    try:
+        return int(parts[-1])
+    except (ValueError, TypeError):
+        return None
+
+
+def _build_content(props: dict[str, Any]) -> str:
+    """Собрать content-jsonb из свойств фичи (только присутствующие ключи, ensure_ascii=False)."""
+    mapping: dict[str, str] = {
+        "restrictions": "gkn_specialz_contentrestrictions",
+        "protected_object": "gkn_specialz_protectedobject",
+        "authority_decision": "authority_decision",
+        "document": "gkn_document",
+        "registration_date": "registration_date",
+        "area": "area",
+        "zone_state": "zone_state_v",
+    }
+    content: dict[str, Any] = {}
+    for out_key, prop_key in mapping.items():
+        val = props.get(prop_key)
+        if val is not None:
+            content[out_key] = val
+    return json.dumps(content, ensure_ascii=False)
+
+
+def _upsert_feature(
+    db: Session,
+    *,
+    source_layer_id: int,
+    layer_kind: str,
+    geom_data_id: int,
+    props: dict[str, Any],
+    geom: dict[str, Any],
+    fetched_at: str,
+) -> bool:
+    """UPSERT одной фичи в ird_overlays под SAVEPOINT. True если записана."""
+    full_name: str = str(props.get("full_name") or "")
+    reg_numb = _extract_reg_numb(full_name)
+    content_json = _build_content(props)
+    raw_props_json = json.dumps(props, ensure_ascii=False)
+    try:
+        with db.begin_nested():
+            db.execute(
+                _UPSERT_SQL,
+                {
+                    "source_layer_id": source_layer_id,
+                    "layer_kind": layer_kind,
+                    "geom_data_id": geom_data_id,
+                    "reg_numb_border": reg_numb,
+                    "category": None,
+                    "subcategory": None,
+                    "geojson": json.dumps(geom),
+                    "raw_props": raw_props_json,
+                    "zone_type_name": full_name or None,
+                    "content": content_json,
+                    "content_source": "ekb_geoportal",
+                    "content_fetched_at": fetched_at,
+                    "fetched_at": fetched_at,
+                },
+            )
+        return True
+    except Exception as exc:
+        logger.warning(
+            "gknspecial_harvest: upsert failed layer=%s geom_data_id=%s: %s",
+            layer_kind,
+            geom_data_id,
+            exc,
+        )
+        return False
+
+
+@celery_app.task(name="tasks.gknspecial_harvest.harvest_gknspecial_zones")
+def harvest_gknspecial_zones() -> dict[str, int]:
+    """Harvest ЗОУИТ-слоя ЕКБ-геопортала (gknspecial_zone) → ird_overlays.
+
+    Для каждого full_name-префикса запрашивает WFS с CQL ``full_name LIKE 'X%'``,
+    фильтрует архивные записи client-side (zone_state_v != 'Действующий'), UPSERT-ит
+    в ird_overlays под SAVEPOINT per-row.
+
+    Фичи, чей full_name не попал ни под один известный префикс, пишутся под
+    layer_kind='gknspecial_unmatched' (source_layer_id=990099) — defensive, чтобы
+    новые типы зон не терялись молча.
+
+    Returns:
+        {'types': N_prefixes_fetched, 'features': M_upserted}
+    """
+    fetched_at = datetime.now(UTC).isoformat()
+    client = EKBGeoportalClient()
+    db = SessionLocal()
+    n_types = 0
+    n_features = 0
+    try:
+        for prefix, expected_layer_kind, _ in _PREFIX_MAP:
+            cql = f"full_name LIKE '{prefix}%'"
+            try:
+                feats = client.features_by_cql("zouit", cql)
+            except Exception as exc:
+                logger.warning(
+                    "gknspecial_harvest: WFS failed prefix=%r layer_kind=%s: %s",
+                    prefix,
+                    expected_layer_kind,
+                    exc,
+                )
+                continue
+            n_types += 1
+            n_written = 0
+            for feat in feats:
+                props = feat.properties or {}
+                # Фильтр активности client-side: пропускаем архивные зоны.
+                # Если поля нет вообще — не скипаем (неизвестное = считаем активным).
+                zone_state = props.get("zone_state_v")
+                if zone_state is not None and zone_state != "Действующий":
+                    continue
+                geom = feat.geometry
+                if not geom:
+                    logger.debug(
+                        "gknspecial_harvest: нет геометрии layer=%s feature_id=%s, skip",
+                        expected_layer_kind,
+                        feat.feature_id,
+                    )
+                    continue
+                geom_data_id = _extract_geom_data_id(feat.feature_id)
+                if geom_data_id is None:
+                    logger.debug(
+                        "gknspecial_harvest: нет geom_data_id feature_id=%s, skip",
+                        feat.feature_id,
+                    )
+                    continue
+                # Роутинг: классифицируем по full_name (единый источник истины).
+                # Defensive: WFS мог вернуть фичу с неожиданным full_name — пишем в unmatched.
+                full_name: str = str(props.get("full_name") or "")
+                cls = _classify_full_name(full_name)
+                if cls is None:
+                    logger.warning(
+                        "gknspecial_harvest: unmatched full_name=%r geom_data_id=%s, "
+                        "пишем в unmatched",
+                        full_name,
+                        geom_data_id,
+                    )
+                    actual_layer_kind = _UNMATCHED_LAYER_KIND
+                    actual_source_layer_id = _UNMATCHED_SOURCE_LAYER_ID
+                else:
+                    actual_layer_kind, actual_source_layer_id = cls
+                if _upsert_feature(
+                    db,
+                    source_layer_id=actual_source_layer_id,
+                    layer_kind=actual_layer_kind,
+                    geom_data_id=geom_data_id,
+                    props=props,
+                    geom=geom,
+                    fetched_at=fetched_at,
+                ):
+                    n_written += 1
+                    n_features += 1
+            logger.info(
+                "gknspecial_harvest: prefix=%r layer_kind=%s fetched=%d written=%d",
+                prefix,
+                expected_layer_kind,
+                len(feats),
+                n_written,
+            )
+
+        db.commit()
+        logger.info("gknspecial_harvest: готово, types=%d features=%d", n_types, n_features)
+        return {"types": n_types, "features": n_features}
+    finally:
+        db.close()

backend/tests/services/test_ird_overlay_lookup.py

@@ -45,6 +45,7 @@ def test_groups_overlaps_by_kind() -> None:
             "subcategory": 20,
             "zone_index": None,
             "zone_type_name": None,
+            "content": None,
         },
         {
             "layer_kind": "zouit_okn",
@@ -53,6 +54,7 @@ def test_groups_overlaps_by_kind() -> None:
             "subcategory": 20,
             "zone_index": None,
             "zone_type_name": None,
+            "content": None,
         },
         {
             "layer_kind": "territorial_zones",
@@ -61,6 +63,7 @@ def test_groups_overlaps_by_kind() -> None:
             "subcategory": 1,
             "zone_index": "Ц-1",
             "zone_type_name": None,
+            "content": None,
         },
     ]
     res = parcel_ird_overlaps(_FakeDB(rows), _WKT)
@@ -71,6 +74,24 @@ def test_groups_overlaps_by_kind() -> None:
     assert tz["zone_index"] == "Ц-1"
 
 
+def test_content_passthrough() -> None:
+    """content из gknspecial_harvest пробрасывается в каждый overlap-dict."""
+    content_val = {"restrictions": "запрещена жилая застройка", "zone_state": "Действующий"}
+    rows = [
+        {
+            "layer_kind": "water_protection",
+            "reg_numb_border": "66:41-6.4125",
+            "category": None,
+            "subcategory": None,
+            "zone_index": None,
+            "zone_type_name": "Водоохранная зона р. Исеть",
+            "content": content_val,
+        },
+    ]
+    res = parcel_ird_overlaps(_FakeDB(rows), _WKT)
+    assert res["ird_overlaps"][0]["content"] == content_val
+
+
 def test_empty_when_no_wkt() -> None:
     res = parcel_ird_overlaps(_FakeDB([{"layer_kind": "x"}]), None)
     assert res == {"ird_overlaps": [], "ird_by_kind": {}}

backend/tests/workers/tasks/test_gknspecial_harvest.py

@@ -0,0 +1,304 @@
+"""Тесты harvest_gknspecial_zones (#1063/#1064/#1065/#1066, #1067 GG-форсайт).
+
+Чистые тесты без живого геопортала / БД:
+  • classify: full_name → корректный (layer_kind, source_layer_id) по каждому префиксу.
+  • reg_numb_border regex: извлечение из full_name, мусор → None.
+  • geom_data_id: 'portal_geo_gknspecial_zone.12345' → 12345; без точки/нечисло → None.
+  • активность: zone_state_v != 'Действующий' → skip; поле отсутствует → не skip.
+  • upsert idempotency: mock-клиент + mock-SessionLocal, два прогона → строки не дублируются.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+from app.workers.tasks.gknspecial_harvest import (
+    _UNMATCHED_LAYER_KIND,
+    _UNMATCHED_SOURCE_LAYER_ID,
+    _classify_full_name,
+    _extract_geom_data_id,
+    _extract_reg_numb,
+    harvest_gknspecial_zones,
+)
+
+# ── classify ─────────────────────────────────────────────────────────────────
+
+
+def test_classify_servitude() -> None:
+    cls = _classify_full_name("Зона публичного сервитута 66:41-6.100")
+    assert cls == ("servitude", 990001)
+
+
+def test_classify_water_protection() -> None:
+    assert _classify_full_name("Водоохранная зона р. Исеть") == ("water_protection", 990002)
+
+
+def test_classify_coastal_strip() -> None:
+    assert _classify_full_name("Прибрежная защитная полоса р.Исеть") == ("coastal_strip", 990003)
+
+
+def test_classify_water_sanitary() -> None:
+    assert _classify_full_name("Зона санитарной охраны ВЗУ") == ("water_sanitary", 990004)
+
+
+def test_classify_sanitary_protection() -> None:
+    assert _classify_full_name("Санитарно-защитная зона завода") == ("sanitary_protection", 990005)
+
+
+def test_classify_okn_protection_zone() -> None:
+    assert _classify_full_name("Зона охраны объекта культурного наследия") == (
+        "okn_protection_zone",
+        990006,
+    )
+
+
+def test_classify_zrz() -> None:
+    assert _classify_full_name("Зона регулирования застройки") == ("zrz", 990007)
+
+
+def test_classify_airfield() -> None:
+    assert _classify_full_name("Приаэродромная территория") == ("airfield", 990008)
+
+
+def test_classify_airfield_with_suffix() -> None:
+    assert _classify_full_name("Приаэродромная территория аэропорта Кольцово") == (
+        "airfield",
+        990008,
+    )
+
+
+def test_classify_unmatched_returns_none() -> None:
+    assert _classify_full_name("Неизвестная зона нового типа") is None
+
+
+def test_classify_empty_returns_none() -> None:
+    assert _classify_full_name("") is None
+
+
+# ── reg_numb_border regex ─────────────────────────────────────────────────────
+
+
+def test_reg_numb_extracts_from_full_name() -> None:
+    assert _extract_reg_numb("Водоохранная зона 66:41-6.4125 р. Исеть") == "66:41-6.4125"
+
+
+def test_reg_numb_extracts_with_multiple_digits() -> None:
+    assert _extract_reg_numb("Зона охраны 66:01-12.300") == "66:01-12.300"
+
+
+def test_reg_numb_returns_none_for_garbage() -> None:
+    assert _extract_reg_numb("Нет номера границы здесь") is None
+
+
+def test_reg_numb_returns_none_for_empty() -> None:
+    assert _extract_reg_numb("") is None
+
+
+# ── geom_data_id ─────────────────────────────────────────────────────────────
+
+
+def test_geom_data_id_normal() -> None:
+    assert _extract_geom_data_id("portal_geo_gknspecial_zone.12345") == 12345
+
+
+def test_geom_data_id_large_number() -> None:
+    assert _extract_geom_data_id("portal_geo_gknspecial_zone.9999999") == 9999999
+
+
+def test_geom_data_id_no_dot() -> None:
+    assert _extract_geom_data_id("portal_geo_gknspecial_zone") is None
+
+
+def test_geom_data_id_non_numeric() -> None:
+    assert _extract_geom_data_id("portal_geo_gknspecial_zone.abc") is None
+
+
+def test_geom_data_id_none_input() -> None:
+    assert _extract_geom_data_id(None) is None
+
+
+def test_geom_data_id_empty_string() -> None:
+    assert _extract_geom_data_id("") is None
+
+
+# ── zone_state_v фильтр активности ───────────────────────────────────────────
+
+
+def _make_feature(
+    feature_id: str = "portal_geo_gknspecial_zone.1",
+    full_name: str = "Водоохранная зона",
+    zone_state_v: str | None = "Действующий",
+    geom: dict[str, Any] | None = None,
+) -> MagicMock:
+    """Фейковый EKBFeature для тестов."""
+    from app.services.scrapers.ekb_geoportal_client import EKBFeature
+
+    props: dict[str, Any] = {"full_name": full_name}
+    if zone_state_v is not None:
+        props["zone_state_v"] = zone_state_v
+    if geom is None:
+        geom = {"type": "Polygon", "coordinates": [[[60.5, 56.8], [60.6, 56.8], [60.5, 56.8]]]}
+    return EKBFeature(feature_id=feature_id, geometry=geom, properties=props)
+
+
+def _run_task_with_feats(feats_per_prefix: list[list[Any]]) -> dict[str, int]:
+    """Запустить harvest_gknspecial_zones с мок-клиентом и мок-БД.
+
+    feats_per_prefix: список из 8 элементов (по числу _PREFIX_MAP), каждый — список EKBFeature.
+    """
+    mock_db = MagicMock()
+    mock_db.begin_nested.return_value.__enter__ = MagicMock(return_value=None)
+    mock_db.begin_nested.return_value.__exit__ = MagicMock(return_value=False)
+
+    # Счётчик вызовов features_by_cql → разные фичи для разных префиксов
+    call_idx: list[int] = [0]
+
+    def fake_features_by_cql(layer: str, cql_filter: str) -> list[Any]:
+        idx = call_idx[0]
+        call_idx[0] += 1
+        if idx < len(feats_per_prefix):
+            return feats_per_prefix[idx]
+        return []
+
+    with (
+        patch("app.workers.tasks.gknspecial_harvest.EKBGeoportalClient") as mock_client_cls,
+        patch("app.workers.tasks.gknspecial_harvest.SessionLocal", return_value=mock_db),
+    ):
+        mock_client = MagicMock()
+        mock_client.features_by_cql.side_effect = fake_features_by_cql
+        mock_client_cls.return_value = mock_client
+        return harvest_gknspecial_zones()
+
+
+def test_skip_archived_zone_state() -> None:
+    """zone_state_v='Архивный' → фича не пишется."""
+    feat = _make_feature(zone_state_v="Архивный")
+    # 8 префиксов: первый с архивной фичей, остальные пустые
+    feats_per_prefix = [[feat]] + [[] for _ in range(7)]
+    result = _run_task_with_feats(feats_per_prefix)
+    assert result["features"] == 0
+
+
+def test_include_feature_without_zone_state() -> None:
+    """Нет поля zone_state_v → фича не скипается (неизвестное = активное)."""
+    feat = _make_feature(zone_state_v=None)
+    feats_per_prefix = [[feat]] + [[] for _ in range(7)]
+    result = _run_task_with_feats(feats_per_prefix)
+    assert result["features"] == 1
+
+
+def test_include_active_zone_state() -> None:
+    """zone_state_v='Действующий' → фича пишется."""
+    feat = _make_feature(zone_state_v="Действующий")
+    feats_per_prefix = [[feat]] + [[] for _ in range(7)]
+    result = _run_task_with_feats(feats_per_prefix)
+    assert result["features"] == 1
+
+
+# ── upsert idempotency ───────────────────────────────────────────────────────
+
+
+def _run_single_feat_task(feat: Any, execute_calls: list[Any]) -> None:
+    """Вспомогательная — один прогон harvest_gknspecial_zones с одной фичей в первом префиксе."""
+    feats_per_prefix = [[feat]] + [[] for _ in range(7)]
+    call_idx: list[int] = [0]
+
+    def fake_by_cql(layer: str, cql_filter: str) -> list[Any]:
+        idx = call_idx[0]
+        call_idx[0] += 1
+        return feats_per_prefix[idx] if idx < len(feats_per_prefix) else []
+
+    mock_db = MagicMock()
+    mock_db.begin_nested.return_value.__enter__ = MagicMock(return_value=None)
+    mock_db.begin_nested.return_value.__exit__ = MagicMock(return_value=False)
+    mock_db.execute.side_effect = lambda *a, **kw: execute_calls.append(a)
+
+    with (
+        patch("app.workers.tasks.gknspecial_harvest.EKBGeoportalClient") as mock_client_cls,
+        patch("app.workers.tasks.gknspecial_harvest.SessionLocal", return_value=mock_db),
+    ):
+        mock_client = MagicMock()
+        mock_client.features_by_cql.side_effect = fake_by_cql
+        mock_client_cls.return_value = mock_client
+        harvest_gknspecial_zones()
+
+
+def test_upsert_idempotency_no_duplicates() -> None:
+    """Два прогона с одной и той же фичей → db.execute дважды, но ON CONFLICT — не дублируем."""
+    feat = _make_feature(
+        feature_id="portal_geo_gknspecial_zone.777",
+        full_name="Водоохранная зона 66:41-6.100",
+        zone_state_v="Действующий",
+    )
+    execute_calls: list[Any] = []
+    _run_single_feat_task(feat, execute_calls)
+    _run_single_feat_task(feat, execute_calls)
+
+    # Оба прогона должны были вызвать db.execute по одному разу (за каждый run по одной фиче).
+    # ON CONFLICT в SQL — гарантирует что строка не дублируется на уровне БД.
+    # Здесь проверяем: execute вызывался ровно дважды (1 фича × 2 прогона).
+    assert len(execute_calls) == 2
+
+
+def test_returns_types_and_features_counts() -> None:
+    """Возврат {'types': N, 'features': M}."""
+    feat1 = _make_feature(feature_id="portal_geo_gknspecial_zone.1", full_name="Водоохранная зона")
+    feat2 = _make_feature(feature_id="portal_geo_gknspecial_zone.2", full_name="Водоохранная зона")
+    # 2 фичи в первом префиксе, остальные пустые
+    feats_per_prefix = [[feat1, feat2]] + [[] for _ in range(7)]
+    result = _run_task_with_feats(feats_per_prefix)
+    # types = 8 (все 8 префиксов запрошены успешно)
+    assert result["types"] == 8
+    assert result["features"] == 2
+
+
+def test_unmatched_feature_written_to_sentinel() -> None:
+    """Фича с full_name не из _PREFIX_MAP → пишется в unmatched (source_layer_id=990099)."""
+    feat = _make_feature(
+        feature_id="portal_geo_gknspecial_zone.999",
+        full_name="Неизвестная зона нового типа",
+        zone_state_v=None,
+    )
+    feats_per_prefix = [[feat]] + [[] for _ in range(7)]
+
+    written_params: list[dict[str, Any]] = []
+
+    def make_db() -> MagicMock:
+        mock_db = MagicMock()
+        mock_db.begin_nested.return_value.__enter__ = MagicMock(return_value=None)
+        mock_db.begin_nested.return_value.__exit__ = MagicMock(return_value=False)
+
+        def capture_execute(sql: Any, params: dict[str, Any] | None = None) -> None:
+            if params:
+                written_params.append(params)
+
+        mock_db.execute.side_effect = capture_execute
+        return mock_db
+
+    call_idx: list[int] = [0]
+
+    def fake_by_cql(layer: str, cql_filter: str) -> list[Any]:
+        idx = call_idx[0]
+        call_idx[0] += 1
+        return feats_per_prefix[idx] if idx < len(feats_per_prefix) else []
+
+    with (
+        patch("app.workers.tasks.gknspecial_harvest.EKBGeoportalClient") as mock_client_cls,
+        patch(
+            "app.workers.tasks.gknspecial_harvest.SessionLocal",
+            return_value=make_db(),
+        ),
+    ):
+        mock_client = MagicMock()
+        mock_client.features_by_cql.side_effect = fake_by_cql
+        mock_client_cls.return_value = mock_client
+        result = harvest_gknspecial_zones()
+
+    assert result["features"] == 1
+    assert len(written_params) == 1
+    assert written_params[0]["source_layer_id"] == _UNMATCHED_SOURCE_LAYER_ID
+    assert written_params[0]["layer_kind"] == _UNMATCHED_LAYER_KIND