fix(ptica): ключ gisogd_permits — id документа на портале, а не (группа, номер) (#2986)

Закрывает #2986.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

backend/app/services/scrapers/gisogd66.py

@@ -60,8 +60,11 @@ _SECTION = "razdel13"
 # гоняем его после того, как регион уже покрыл основную массу (см. circuit breaker ниже).
 SCHEMAS: tuple[str, ...] = ("agate_sverdregion", "agate_ekbgo")
 
-# Группа источника (doc group key) → наш doc_group-код в БД (CHECK IN ('RS','RV')).
+# Группа источника (doc group key) → наш doc_group-код в БД (CHECK IN ('RS','RV','IZ')).
-GROUP_CODE: dict[str, str] = {"DocRS": "RS", "DocRV": "RV"}
+#
+# DocIZ («Изменение в Разрешение на строительство») есть на портале с самого начала,
+# но в этом словаре его не было — 548 документов не грузились вовсе (#2986).
+GROUP_CODE: dict[str, str] = {"DocRS": "RS", "DocRV": "RV", "DocIZ": "IZ"}
 
 _HTTP_TIMEOUT = 20.0
 _MAX_CONNECTIONS = 5
@@ -259,11 +262,18 @@ def _existing_reg_dates(db: Session, schema: str) -> dict[str, date | None]:
 
 
 def _upsert_permit(db: Session, rec: dict[str, Any]) -> str:
-    """UPSERT одной записи в gisogd_permits по (doc_group, doc_num). Per-row SAVEPOINT.
+    """UPSERT одной записи в gisogd_permits по source_key. Per-row SAVEPOINT.
 
-    Конфликт-резолв: при коллизии бизнес-ключа обновляем ТОЛЬКО если у новой записи
+    Ключ — идентификатор документа на портале (#2986). Раньше ключом был
-    date_reg НЕ старше сохранённой (EXCLUDED.date_reg >= existing) — предпочитаем
+    (doc_group, doc_num), но docNum у ГИСОГД НЕ уникален: разрешение и изменения к
-    более позднюю регистрацию (или запись без даты не затирает датированную).
+    нему носят один номер, и UPSERT оставлял только одно из них. Замер 20.08.2026:
+    так схлопывалось 2243 документа, а межсхемных дублей — ради которых ключ и
+    вводился — всего 7, и у них key ОБЩИЙ, то есть новый ключ их тоже склеивает.
+
+    Конфликт-резолв: при коллизии обновляем ТОЛЬКО если у новой записи date_reg НЕ
+    старше сохранённой (EXCLUDED.date_reg >= existing) — предпочитаем более позднюю
+    регистрацию (или запись без даты не затирает датированную). Осмыслен ровно для
+    тех 7 межсхемных совпадений.
 
     Returns: 'inserted' | 'updated' | 'skipped_unchanged'.
     """
@@ -285,13 +295,14 @@ def _upsert_permit(db: Session, rec: dict[str, Any]) -> str:
                          ST_GeomFromGeoJSON(CAST(:geojson AS text)), 4326)) END,
                 NOW(), NOW()
             )
-            ON CONFLICT (doc_group, doc_num) DO UPDATE
+            ON CONFLICT (source_key) DO UPDATE
-               SET doc_name              = EXCLUDED.doc_name,
+               SET doc_group             = EXCLUDED.doc_group,
+                   doc_num               = EXCLUDED.doc_num,
+                   doc_name              = EXCLUDED.doc_name,
                    date_doc              = EXCLUDED.date_doc,
                    date_reg              = EXCLUDED.date_reg,
                    approved_organization = EXCLUDED.approved_organization,
                    source_schema         = EXCLUDED.source_schema,
-                   source_key            = EXCLUDED.source_key,
                    cad_nums              = EXCLUDED.cad_nums,
                    geom                  = EXCLUDED.geom,
                    updated_at            = NOW()

backend/app/services/site_finder/permits_nearby.py

@@ -50,7 +50,13 @@ _PERMITS_NEARBY_SQL = text("""
             ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography
         ) AS distance_m
       FROM gisogd_permits
-     WHERE geom IS NOT NULL
+     -- Только РНС/РВЭ: агрегат обещает total_count = rs_count + rv_count, а с
+     -- #2986 в таблице появилась третья группа 'IZ' (изменения в разрешение).
+     -- Она попадала бы в total и не попадала ни в один из счётчиков — молчаливое
+     -- расхождение. Показывать ли изменения отдельной строкой в §6 — вопрос
+     -- продуктовый (см. #2986); до его решения выборка сужена явно, а не молча.
+     WHERE doc_group IN ('RS', 'RV')
+       AND geom IS NOT NULL
        AND ST_DWithin(
                geom::geography,
                ST_Centroid(ST_GeomFromText(:wkt, 4326))::geography,

backend/tests/services/site_finder/test_permits_nearby.py

@@ -189,3 +189,42 @@ def test_db_error_propagates_not_swallowed() -> None:
     db.execute.side_effect = RuntimeError("db down")
     with pytest.raises(RuntimeError, match="db down"):
         get_permits_nearby(db, _WKT)
+
+
+def test_sql_excludes_the_amendments_group() -> None:
+    """Выборка сужена до РНС/РВЭ явно — иначе total_count перестанет сходиться (#2986).
+
+    С #2986 в gisogd_permits появилась третья группа 'IZ' (изменения в разрешение).
+    Агрегат обещает `total_count = rs_count + rv_count`, а строки 'IZ' попадали бы в
+    total и не попадали ни в один счётчик. Показывать ли изменения в §6 — вопрос
+    продуктовый; до его решения сужение должно стоять в запросе явно, а не
+    держаться на том, что таких строк «пока нет».
+    """
+    sql = str(_PERMITS_NEARBY_SQL)
+    assert (
+        "doc_group IN ('RS', 'RV')" in sql
+    ), f"запрос не сужен по группе — строки 'IZ' сломают total_count:\n{sql}"
+
+
+def test_total_equals_rs_plus_rv_even_if_iz_leaks_in() -> None:
+    """Контроль инварианта на данных: даже если строка 'IZ' просочилась, счёт сходится.
+
+    Проверяем не текст запроса, а поведение агрегата: сумма частей равна целому.
+    Тест краснеет, если кто-то снимет сужение и не добавит счётчик для 'IZ'.
+    """
+    rows = [
+        {
+            "doc_group": g,
+            "doc_name": f"Документ {g}",
+            "doc_num": f"№{i}",
+            "date_doc": dt.date(2026, 2, 26),
+            "date_reg": dt.date(2026, 2, 26),
+            "approved_organization": "Администрация",
+            "distance_m": 10.0 + i,
+        }
+        for i, g in enumerate(("RS", "RV", "RS"))
+    ]
+    db = MagicMock()
+    db.execute.return_value.mappings.return_value.all.return_value = rows
+    res = get_permits_nearby(db, _WKT)
+    assert res["total_count"] == res["rs_count"] + res["rv_count"], res

backend/tests/skip_allowlist.txt

@@ -154,3 +154,14 @@ tests/sql/test_2464_act_date_backfill.py::test_other_documents_with_same_date_ar
 tests/sql/test_2464_act_date_backfill.py::test_correct_rows_do_not_move
 tests/sql/test_2464_act_date_backfill.py::test_migration_is_idempotent
 tests/sql/test_2464_act_date_backfill.py::test_without_the_migration_rows_stay_wrong
+
+# ── #2986: бизнес-ключ gisogd_permits по source_key (миграция 191) ────────────
+# Нужен живой Postgres: тесты создают ВРЕМЕННУЮ копию таблицы в прод-форме (со СТАРЫМ
+# ключом), прогоняют тело миграции и проверяют, что разрешение и изменения к нему
+# живут раздельно, а межсхемный дубль по-прежнему склеивается. Два гейта на лоадер
+# (GROUP_CODE и цель ON CONFLICT) базы НЕ требуют и в этот список НЕ входят.
+# В CI ИДУТ (postgres-сервис, #2745); записи нужны для машины без БД.
+tests/sql/test_2986_permits_source_key.py::test_old_key_collapses_permit_and_its_amendment
+tests/sql/test_2986_permits_source_key.py::test_after_migration_both_documents_survive
+tests/sql/test_2986_permits_source_key.py::test_cross_schema_duplicate_still_merges
+tests/sql/test_2986_permits_source_key.py::test_migration_allows_the_izmeneniya_group

backend/tests/sql/test_2986_permits_source_key.py

@@ -0,0 +1,260 @@
+"""Бизнес-ключ gisogd_permits — key документа на портале, а не (группа, номер) (#2986).
+
+`UNIQUE (doc_group, doc_num)` вводился, чтобы склеивать ОДИН документ, пришедший из
+двух схем портала. Замер 20.08.2026 показал, что эта задача почти отсутствует, а
+побочный эффект огромен — `docNum` у ГИСОГД не уникален: разрешение и изменения к
+нему носят один номер.
+
+    группа   документов   различных key   различных docNum   схлопывается
+    DocRS         6098          6096            4305             1793
+    DocRV         5419          5415            4969              450
+    DocIZ          548           547             393              155 (+не грузилась)
+
+    общих docNum между схемами (DocRS): 2      ← ради этого и вводился ключ
+    общих key    между схемами (DocRS): 2      ← те же два, key их тоже склеит
+
+Пример 66-06-06-2026: портал отдаёт два документа (key …719586 — разрешение,
+key …752293 — изменения к нему), а на проде осталось только изменение. Так вытеснено
+598 из 4320 строк РНС (13.8 %).
+
+Тесты герметичные: временная копия таблицы, боевые данные не читаются и не меняются.
+"""
+
+from __future__ import annotations
+
+import os
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+
+import re
+from pathlib import Path
+
+import pytest
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+_MIGRATION = (
+    Path(__file__).resolve().parents[3]
+    / "data"
+    / "sql"
+    / "192_gisogd_permits_key_by_source_key.sql"
+)
+
+
+def _dsn() -> str:
+    raw = os.environ.get("TEST_DATABASE_URL") or os.environ.get(
+        "DATABASE_URL", "postgresql+psycopg://gendesign@localhost:15432/gendesign"
+    )
+    return (
+        raw
+        if raw.startswith("postgresql+")
+        else raw.replace("postgresql://", "postgresql+psycopg://")
+    )
+
+
+def _db_reachable() -> tuple[bool, str]:
+    try:
+        eng = create_engine(_dsn(), connect_args={"connect_timeout": 3})
+        with eng.connect() as c:
+            c.execute(text("SELECT 1"))
+        return True, ""
+    except Exception as exc:
+        return False, str(exc)
+
+
+_DB_OK, _DB_ERR = _db_reachable()
+
+# Прод-форма таблицы ДО миграции 191 (важные для ключа поля).
+_TEMP_OLD = """
+CREATE TEMP TABLE gisogd_permits (
+    id            bigserial PRIMARY KEY,
+    doc_group     text NOT NULL,
+    doc_num       text NOT NULL,
+    doc_name      text,
+    date_reg      date,
+    source_schema text NOT NULL,
+    source_key    text NOT NULL,
+    CONSTRAINT chk_gisogd_permits_doc_group CHECK (doc_group IN ('RS', 'RV')),
+    CONSTRAINT gisogd_permits_doc_group_doc_num_key UNIQUE (doc_group, doc_num)
+) ON COMMIT DROP
+"""
+
+# Два РАЗНЫХ документа с одним номером — дословно с портала.
+_РАЗРЕШЕНИЕ = {
+    "doc_group": "RS",
+    "doc_num": "66-06-06-2026",
+    "doc_name": "Разрешение на строительство № 66-06-06-2026 от 26.02.2026",
+    "date_reg": "2026-02-26",
+    "source_schema": "agate_sverdregion",
+    "source_key": "1000130002719586",
+}
+_ИЗМЕНЕНИЕ = {
+    "doc_group": "RS",
+    "doc_num": "66-06-06-2026",
+    "doc_name": "Изменения в разрешение на строительство №66-06-06-2026 от 26.02.2026",
+    "date_reg": "2026-08-04",
+    "source_schema": "agate_sverdregion",
+    "source_key": "1000130002752293",
+}
+
+_INSERT_COLS = (
+    "(doc_group, doc_num, doc_name, date_reg, source_schema, source_key)"
+    " VALUES (:doc_group,:doc_num,:doc_name,CAST(:date_reg AS date),"
+    ":source_schema,:source_key)"
+)
+# Старый UPSERT — цитата прежнего кода, нужна как контроль механизма.
+_OLD_UPSERT = (
+    f"INSERT INTO gisogd_permits {_INSERT_COLS}"
+    " ON CONFLICT (doc_group, doc_num) DO UPDATE SET doc_name = EXCLUDED.doc_name,"
+    " date_reg = EXCLUDED.date_reg, source_key = EXCLUDED.source_key"
+    " WHERE EXCLUDED.date_reg > gisogd_permits.date_reg"
+)
+_NEW_UPSERT = (
+    f"INSERT INTO gisogd_permits {_INSERT_COLS}"
+    " ON CONFLICT (source_key) DO UPDATE SET doc_name = EXCLUDED.doc_name,"
+    " date_reg = EXCLUDED.date_reg"
+)
+
+
+def _migration_chunks() -> list[str]:
+    raw = _MIGRATION.read_text(encoding="utf-8")
+    без_комментов = "\n".join(ln for ln in raw.splitlines() if not ln.lstrip().startswith("--"))
+    return [
+        ch.strip()
+        for ch in re.split(r";\s*(?:\n|$)", без_комментов)
+        if ch.strip() and ch.strip().upper() not in {"BEGIN", "COMMIT"}
+    ]
+
+
+def _session():
+    return sessionmaker(bind=create_engine(_dsn()))()
+
+
+# ── Гейт на лоадер: работает БЕЗ базы, двусторонний ───────────────────────────
+
+
+def test_loader_knows_the_amendments_group() -> None:
+    """DocIZ обязан быть в GROUP_CODE — иначе 548 документов не грузятся вовсе.
+
+    На origin/main словарь содержит только DocRS/DocRV — конкретное неверное
+    значение, а не отсутствующий символ.
+    """
+    from app.services.scrapers.gisogd66 import GROUP_CODE
+
+    assert (
+        GROUP_CODE.get("DocIZ") == "IZ"
+    ), f"группа изменений не грузится; GROUP_CODE = {GROUP_CODE}"
+
+
+def test_loader_upserts_by_source_key() -> None:
+    """UPSERT обязан конфликтовать по source_key, а не по (doc_group, doc_num).
+
+    На origin/main в тексте запроса стоит старый ключ — он и схлопывает
+    разрешение с изменениями к нему.
+    """
+    import inspect
+
+    from app.services.scrapers.gisogd66 import _upsert_permit
+
+    src = inspect.getsource(_upsert_permit)
+    assert "ON CONFLICT (source_key)" in src, "UPSERT конфликтует не по source_key"
+    assert (
+        "ON CONFLICT (doc_group, doc_num)" not in src
+    ), "старый ключ всё ещё в запросе — разрешение и изменения схлопнутся"
+
+
+# ── Механизм и миграция: нужен живой Postgres ────────────────────────────────
+
+_pg = pytest.mark.skipif(not _DB_OK, reason=f"Postgres недоступен: {_DB_ERR}")
+
+
+@_pg
+def test_old_key_collapses_permit_and_its_amendment() -> None:
+    """Демонстрация механизма: со СТАРЫМ ключом остаётся одна строка из двух.
+
+    Не двусторонний тест, а показ причины: и старая схема, и старый UPSERT здесь
+    процитированы явно, поэтому он одинаково зелёный на обеих сторонах. Нужен,
+    чтобы утверждение «ключ схлопывает разные документы» было проверяемым, а не
+    декларативным.
+    """
+    s = _session()
+    try:
+        s.execute(text(_TEMP_OLD))
+        s.execute(text(_OLD_UPSERT), _РАЗРЕШЕНИЕ)
+        s.execute(text(_OLD_UPSERT), _ИЗМЕНЕНИЕ)
+        rows = s.execute(text("SELECT doc_name, source_key FROM gisogd_permits")).all()
+        assert len(rows) == 1, f"ожидали схлопывание, получили {len(rows)} строк"
+        assert (
+            "Изменения" in rows[0][0]
+        ), f"вытеснено не то: осталось {rows[0][0]!r} — на проде остаётся именно изменение"
+    finally:
+        s.rollback()
+        s.close()
+
+
+@_pg
+def test_after_migration_both_documents_survive() -> None:
+    """Головной: после миграции разрешение и изменения к нему живут раздельно."""
+    s = _session()
+    try:
+        s.execute(text(_TEMP_OLD))
+        for chunk in _migration_chunks():
+            s.execute(text(chunk))
+        s.execute(text(_NEW_UPSERT), _РАЗРЕШЕНИЕ)
+        s.execute(text(_NEW_UPSERT), _ИЗМЕНЕНИЕ)
+        rows = s.execute(text("SELECT doc_name FROM gisogd_permits ORDER BY source_key")).all()
+        assert len(rows) == 2, f"документы всё ещё схлопываются: {rows}"
+        имена = " | ".join(r[0] for r in rows)
+        assert "Разрешение на строительство" in имена and "Изменения" in имена, имена
+    finally:
+        s.rollback()
+        s.close()
+
+
+@_pg
+def test_cross_schema_duplicate_still_merges() -> None:
+    """Контроль: настоящий межсхемный дубль (ОБЩИЙ key) по-прежнему склеивается.
+
+    Ради этого случая старый ключ и вводился — новый обязан его сохранить.
+    """
+    s = _session()
+    try:
+        s.execute(text(_TEMP_OLD))
+        for chunk in _migration_chunks():
+            s.execute(text(chunk))
+        s.execute(text(_NEW_UPSERT), _РАЗРЕШЕНИЕ)
+        из_другой_схемы = dict(_РАЗРЕШЕНИЕ, source_schema="agate_ekbgo")
+        s.execute(text(_NEW_UPSERT), из_другой_схемы)
+        n = s.execute(text("SELECT count(*) FROM gisogd_permits")).scalar()
+        assert n == 1, f"межсхемный дубль размножился: {n} строк"
+    finally:
+        s.rollback()
+        s.close()
+
+
+@_pg
+def test_migration_allows_the_izmeneniya_group() -> None:
+    """Контроль CHECK: после миграции doc_group='IZ' принимается, мусор — нет."""
+    s = _session()
+    try:
+        s.execute(text(_TEMP_OLD))
+        for chunk in _migration_chunks():
+            s.execute(text(chunk))
+        s.execute(
+            text(f"INSERT INTO gisogd_permits {_INSERT_COLS}"),
+            dict(_ИЗМЕНЕНИЕ, doc_group="IZ", source_key="izm-1"),
+        )
+        assert s.execute(text("SELECT count(*) FROM gisogd_permits")).scalar() == 1
+        s.execute(text("SAVEPOINT p"))
+        try:
+            s.execute(
+                text(f"INSERT INTO gisogd_permits {_INSERT_COLS}"),
+                dict(_ИЗМЕНЕНИЕ, doc_group="XX", source_key="mus-1"),
+            )
+            raise AssertionError("CHECK пропустил мусорную группу 'XX'")
+        except Exception as exc:
+            assert "chk_gisogd_permits_doc_group" in str(exc), str(exc)[:120]
+        s.execute(text("ROLLBACK TO SAVEPOINT p"))
+    finally:
+        s.rollback()
+        s.close()

data/sql/192_gisogd_permits_key_by_source_key.sql

@@ -0,0 +1,57 @@
+-- 192: бизнес-ключ gisogd_permits — идентификатор документа на портале (#2986).
+--
+-- Было: UNIQUE (doc_group, doc_num). Ключ выбирался, чтобы склеивать ОДИН документ,
+-- пришедший из двух схем портала. Замер 20.08.2026 показал, что задача, ради которой
+-- он введён, почти отсутствует, а побочный эффект огромен:
+--
+--   общих docNum между схемами (DocRS):   2
+--   общих key    между схемами (DocRS):   2      ← те же два
+--
+--   группа   документов   различных key   различных docNum   схлопывается
+--   DocRS         6098          6096            4305             1793
+--   DocRV         5419          5415            4969              450
+--   DocIZ          548           547             393              155
+--
+-- Причина: docNum у ГИСОГД не уникален. Разрешение и изменения к нему носят ОДИН
+-- номер. Пример 66-06-06-2026: портал отдаёт два документа (key 1000130002719586 —
+-- само разрешение, key 1000130002752293 — изменения к нему), а UPSERT с
+-- предпочтением позднего date_reg оставлял только изменение. Самого разрешения в
+-- данных не оставалось. На проде так вытеснено 598 из 4320 строк РНС (13.8%).
+--
+-- Стало: UNIQUE (source_key) — идентификатор документа на портале. Он разделяет
+-- разрешение и изменения (разные key) и по-прежнему склеивает настоящие межсхемные
+-- дубли (у них key ОБЩИЙ — те самые 7 записей по всем трём группам).
+--
+-- Дедуп перед сменой ключа не нужен: на проде 20.08.2026 source_key уже уникален —
+-- 9182 различных на 9182 строки, NOT NULL. Констрейнт встаёт как есть.
+--
+-- CHECK расширен значением 'IZ': группа DocIZ (изменения) есть на портале с самого
+-- начала, но в GROUP_CODE лоадера её не было, и 548 документов не грузились вовсе.
+
+BEGIN;
+
+SET LOCAL lock_timeout = '5s';
+
+ALTER TABLE gisogd_permits
+    DROP CONSTRAINT IF EXISTS gisogd_permits_doc_group_doc_num_key;
+
+ALTER TABLE gisogd_permits
+    DROP CONSTRAINT IF EXISTS chk_gisogd_permits_doc_group;
+
+ALTER TABLE gisogd_permits
+    ADD CONSTRAINT chk_gisogd_permits_doc_group
+    CHECK (doc_group IN ('RS', 'RV', 'IZ'));
+
+ALTER TABLE gisogd_permits
+    ADD CONSTRAINT uq_gisogd_permits_source_key UNIQUE (source_key);
+
+-- (doc_group, doc_num) перестаёт быть уникальным, но остаётся рабочим фильтром
+-- («все документы по номеру разрешения») — оставляем как обычный индекс.
+CREATE INDEX IF NOT EXISTS idx_gisogd_permits_group_num
+    ON gisogd_permits (doc_group, doc_num);
+
+COMMENT ON CONSTRAINT uq_gisogd_permits_source_key ON gisogd_permits IS
+    'Бизнес-ключ = key документа на портале ГИСОГД-СО. docNum НЕ уникален: '
+    'разрешение и изменения к нему носят один номер (#2986).';
+
+COMMIT;