fix(ptica): land_reservation перестаёт копить дубли — 91% таблицы были копиями (#2464) (#2966)

backend/app/workers/tasks/izyatie_ocr_ingest.py

@@ -7,14 +7,18 @@ UPSERT-ит в land_reservation (м.136). Reservation_lookup / analyze-wiring (#
 
 Дедуп-ключ:
   ON CONFLICT (cad_num, act_number) — унаследован из reservation_ingest.py.
-  Если act_number IS NULL (не извлечён из сканов) → конфликт НЕ возникает при NULL-UPSERT
-  (NULL != NULL в SQL). Чтобы предотвратить дубли при act_number IS NULL, дедуплицируем
-  по (cad_num, doc_url) на уровне Python перед UPSERT: один URL = один батч,
-  повторный запуск с тем же URL обновит существующую строку через source+fetched_at
-  (где act_number IS NULL используем DO NOTHING вместо DO UPDATE — нет stable key).
-  Решение: для строк с act_number IS NULL добавляем в ON CONFLICT УНИКАЛЬНОСТЬ через
-  отдельный UPSERT с COALESCE-fallback: если запись с (cad_num, doc_url) уже есть —
-  UPDATE, иначе INSERT. Реализовано через двухшаговый UPSERT ниже.
+  Уникальность держит констрейнт uq_land_reservation_cad_act; с миграции 189 он
+  объявлен как UNIQUE NULLS NOT DISTINCT, поэтому записи без номера акта тоже
+  конфликтуют между собой и ON CONFLICT DO NOTHING реально их ловит.
+
+  До м.189 констрейнт был обычным UNIQUE, где NULL != NULL: у записей с
+  act_number IS NULL конфликт не наступал никогда, и каждый недельный прогон
+  вставлял копию. Замер прода 20.08.2026 до правки — 297 строк, все без номера
+  акта, 27 групп с дублями, до 11 копий, 270 лишних строк (91% таблицы).
+
+  Прежняя редакция этого docstring обещала python-дедуп по (cad_num, doc_url)
+  перед UPSERT и «двухшаговый UPSERT ниже». Ни того, ни другого в коде не было —
+  описание расходилось с реализацией и скрывало накопление дублей (#2464).
 
 Beat: еженедельно (пятница 07:00 МСК) — изъятия выходят редко.
 
@@ -45,10 +49,13 @@ logger = logging.getLogger(__name__)
 #   Stable key = (cad_num, act_number). Идемпотентно при повторном прогоне.
 #
 # Вариант B (act_number IS NULL): INSERT ... ON CONFLICT DO NOTHING.
-#   NULL != NULL → (cad_num, NULL) никогда не конфликтует по индексу.
-#   Python-дедуп per-batch предотвращает дубли в рамках одного прогона.
-#   Повторные прогоны добавят дубли если строки нет — acceptable (rare, data audit OK).
-#   Альтернатива (partial unique index на NULL) — задача database-expert, не здесь.
+#   Работает с миграции 189: uq_land_reservation_cad_act объявлен как
+#   UNIQUE NULLS NOT DISTINCT, поэтому (cad_num, NULL) конфликтует с такой же
+#   строкой и повторный прогон становится no-op.
+#   Прежний комментарий здесь оценивал накопление дублей как «rare, data audit OK»
+#   и откладывал уникальный индекс. Оценка не подтвердилась: на 20.08.2026 дубли
+#   составляли 91% таблицы (270 лишних строк из 297), максимум 11 копий одной
+#   записи. Отложенный вариант и реализован м.189 (#2464).
 
 _UPSERT_WITH_ACT_SQL = text(
     """

backend/tests/skip_allowlist.txt

@@ -115,3 +115,13 @@ tests/sql/test_2464_leads_stats_suffix_contract.py::test_window_suffixed_fields_
 tests/sql/test_2464_leads_stats_suffix_contract.py::test_total_suffixed_fields_are_all_time
 tests/sql/test_2464_leads_stats_suffix_contract.py::test_revenue_and_deals_are_named_by_their_scope
 tests/sql/test_2464_leads_stats_suffix_contract.py::test_window_months_present_on_empty_data
+# ── #2464: дедуп land_reservation (миграция 189) ──────────────────────────────
+# Нужен живой Postgres: тесты создают ВРЕМЕННУЮ копию таблицы, проверяют семантику
+# UNIQUE NULLS NOT DISTINCT и репетируют миграцию на засеянных дублях. В CI ИДУТ
+# (postgres-сервис, #2745); записи нужны для машины без БД и без туннеля.
+tests/sql/test_2464_land_reservation_dedup.py::test_nulls_not_distinct_deduplicates
+tests/sql/test_2464_land_reservation_dedup.py::test_plain_unique_does_not_deduplicate
+tests/sql/test_2464_land_reservation_dedup.py::test_records_with_act_number_still_deduplicate
+tests/sql/test_2464_land_reservation_dedup.py::test_different_parcels_are_not_collapsed
+tests/sql/test_2464_land_reservation_dedup.py::test_migration_dedup_statement_matches_the_key
+tests/sql/test_2464_land_reservation_dedup.py::test_migration_body_runs_on_a_prod_shaped_replica

backend/tests/sql/test_2464_land_reservation_dedup.py

@@ -0,0 +1,215 @@
+"""ON CONFLICT DO NOTHING в land_reservation обязан реально ловить дубли (#2464).
+
+`_UPSERT_NO_ACT_SQL` (workers/tasks/izyatie_ocr_ingest.py) заканчивается
+`ON CONFLICT DO NOTHING`, а единственный подходящий констрейнт был
+`UNIQUE (cad_num, act_number)` с обычной NULL-семантикой. В Postgres NULL != NULL,
+поэтому у записей БЕЗ номера акта конфликт не наступал никогда — каждый недельный
+прогон вставлял копию.
+
+Замер прода 20.08.2026 до правки: 297 строк, все с `act_number IS NULL`, 27 групп с
+дублями, до 11 копий, 270 лишних строк — 91 % таблицы.
+
+Миграция 189 дедуплицирует таблицу и пересоздаёт констрейнт как
+`UNIQUE NULLS NOT DISTINCT`. Здесь проверяется САМ МЕХАНИЗМ на временной копии:
+с новой семантикой повторная вставка — no-op, со старой — дубль. Второе
+утверждение обязательно: без него тест не отличить от «оно и так работало».
+
+Тест герметичный: таблицы временные, боевые данные не читаются и не меняются.
+"""
+
+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"
+    / "189_land_reservation_nulls_not_distinct.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()
+pytestmark = pytest.mark.skipif(not _DB_OK, reason=f"Postgres недоступен: {_DB_ERR}")
+
+_TABLE = """
+CREATE TEMP TABLE land_reservation (
+    id bigserial PRIMARY KEY,
+    cad_num text NOT NULL,
+    act_number text,
+    doc_url text,
+    reservation_kind text,
+    is_active boolean DEFAULT true,
+    fetched_at timestamptz DEFAULT now()
+) ON COMMIT DROP;
+"""
+
+_INSERT = """
+INSERT INTO land_reservation (cad_num, act_number, doc_url, reservation_kind)
+VALUES (:cad, :act, :url, 'изъятие')
+ON CONFLICT DO NOTHING
+"""
+
+
+@pytest.fixture
+def db():
+    engine = create_engine(_dsn())
+    session = sessionmaker(bind=engine)()
+    try:
+        session.execute(text(_TABLE))
+        n = session.execute(text("SELECT count(*) FROM land_reservation")).scalar()
+        assert n == 0, f"запрос попал НЕ во временную таблицу ({n} строк)"
+        yield session
+    finally:
+        session.rollback()
+        session.close()
+        engine.dispose()
+
+
+def _add_constraint(db, nulls_not_distinct: bool) -> None:
+    kind = "UNIQUE NULLS NOT DISTINCT" if nulls_not_distinct else "UNIQUE"
+    db.execute(
+        text(f"ALTER TABLE land_reservation ADD CONSTRAINT uq_t {kind} (cad_num, act_number)")
+    )
+
+
+def _insert_twice(db) -> int:
+    for _ in range(2):
+        db.execute(
+            text(_INSERT), {"cad": "66:41:0303004:22", "act": None, "url": "https://x/y.pdf"}
+        )
+    return int(db.execute(text("SELECT count(*) FROM land_reservation")).scalar())
+
+
+def test_nulls_not_distinct_deduplicates(db) -> None:
+    """С новой семантикой повторная вставка act-less записи — no-op."""
+    _add_constraint(db, nulls_not_distinct=True)
+    assert _insert_twice(db) == 1, "дубль всё равно вставился"
+
+
+def test_plain_unique_does_not_deduplicate(db) -> None:
+    """Фальсификация: со СТАРЫМ констрейнтом дубль обязан появиться.
+
+    Без этой проверки зелёный тест выше неотличим от «оно и так работало».
+    """
+    _add_constraint(db, nulls_not_distinct=False)
+    assert (
+        _insert_twice(db) == 2
+    ), "обычный UNIQUE неожиданно поймал дубль — значит тест выше ничего не доказывает"
+
+
+def test_records_with_act_number_still_deduplicate(db) -> None:
+    """Контроль: записи С номером акта дедуплицировались и раньше — не сломали."""
+    _add_constraint(db, nulls_not_distinct=True)
+    for _ in range(2):
+        db.execute(text(_INSERT), {"cad": "66:41:1", "act": "12-АК", "url": "https://x/1.pdf"})
+    assert int(db.execute(text("SELECT count(*) FROM land_reservation")).scalar()) == 1
+
+
+def test_different_parcels_are_not_collapsed(db) -> None:
+    """Контроль: разные участки без номера акта остаются разными строками.
+
+    Ловит «починку» через слишком широкий ключ.
+    """
+    _add_constraint(db, nulls_not_distinct=True)
+    for cad in ("66:41:1", "66:41:2", "66:41:3"):
+        db.execute(text(_INSERT), {"cad": cad, "act": None, "url": "https://x/z.pdf"})
+    assert int(db.execute(text("SELECT count(*) FROM land_reservation")).scalar()) == 3
+
+
+def test_migration_dedup_statement_matches_the_key(db) -> None:
+    """DELETE в миграции обязан чистить ровно по ключу констрейнта.
+
+    Расхождение ключа дедупа и ключа констрейнта означало бы, что после DELETE
+    констрейнт всё равно не создастся — миграция упала бы на проде.
+    """
+    sql = _MIGRATION.read_text()
+    assert "UNIQUE NULLS NOT DISTINCT (cad_num, act_number)" in sql
+    delete_stmt = re.search(r"DELETE FROM land_reservation.*?;", sql, re.S)
+    assert delete_stmt is not None, "в миграции нет DELETE — дедуп не выполняется"
+    body = delete_stmt.group(0)
+    assert "a.cad_num = b.cad_num" in body, "дедуп не по cad_num"
+    assert (
+        "a.act_number IS NULL" in body and "b.act_number IS NULL" in body
+    ), "дедуп затрагивает записи С номером акта — они и так были уникальны"
+    assert "a.id > b.id" in body, "не задан выживающий (минимальный id)"
+
+
+def test_migration_body_runs_on_a_prod_shaped_replica(db) -> None:
+    """Репетиция миграции: 11 копий → 1 строка, констрейнт создаётся.
+
+    Сильнее проверки регулярками: исполняются РЕАЛЬНЫЕ выражения из файла миграции.
+    Если DELETE чистит не по тому ключу, ADD CONSTRAINT здесь же и упадёт — как
+    упал бы на проде.
+    """
+    # Засев как на проде: одна группа, 11 точных копий, плюс соседний участок.
+    for _ in range(11):
+        db.execute(
+            text(
+                "INSERT INTO land_reservation (cad_num, act_number, doc_url, reservation_kind)"
+                " VALUES ('66:41:0303004:22', NULL, 'https://x/y.pdf', 'изъятие')"
+            )
+        )
+    db.execute(
+        text(
+            "INSERT INTO land_reservation (cad_num, act_number, doc_url, reservation_kind)"
+            " VALUES ('66:41:0206032:8499', NULL, 'https://x/z.pdf', 'изъятие')"
+        )
+    )
+    assert int(db.execute(text("SELECT count(*) FROM land_reservation")).scalar()) == 12
+
+    sql = _MIGRATION.read_text()
+    body = sql[sql.index("BEGIN;") + len("BEGIN;") : sql.rindex("COMMIT;")]
+    # Комментарии снимаем ДО разбиения на выражения — иначе точка с запятой внутри
+    # комментария разрежет SQL посередине. Обе ловушки этот тест уже ловил на себе:
+    # сперва пропуск куска, начинающегося с «--» (потерялся DELETE, репетиция упала
+    # на ADD CONSTRAINT), затем «;» в тексте комментария.
+    code = "\n".join(
+        ln for ln in body.splitlines() if ln.strip() and not ln.lstrip().startswith("--")
+    )
+    for chunk in code.split(";"):
+        stmt = chunk.strip()
+        if stmt:
+            db.execute(text(stmt))
+
+    rows = db.execute(
+        text("SELECT cad_num, count(*) FROM land_reservation GROUP BY 1 ORDER BY 1")
+    ).all()
+    assert [(r[0], r[1]) for r in rows] == [
+        ("66:41:0206032:8499", 1),
+        ("66:41:0303004:22", 1),
+    ], f"после миграции осталось не по одной строке: {rows}"
+
+    # И теперь повторная вставка действительно no-op.
+    db.execute(text(_INSERT), {"cad": "66:41:0303004:22", "act": None, "url": "https://x/y.pdf"})
+    assert int(db.execute(text("SELECT count(*) FROM land_reservation")).scalar()) == 2

data/sql/189_land_reservation_nulls_not_distinct.sql

@@ -0,0 +1,58 @@
+-- 189_land_reservation_nulls_not_distinct.sql
+-- #2464 — дедуп land_reservation и UNIQUE NULLS NOT DISTINCT на живой таблице.
+--
+-- БАГ. `_UPSERT_NO_ACT_SQL` (workers/tasks/izyatie_ocr_ingest.py) заканчивается
+-- `ON CONFLICT DO NOTHING`, а единственный подходящий констрейнт —
+-- `uq_land_reservation_cad_act UNIQUE (cad_num, act_number)` с обычной NULL-семантикой.
+-- В Postgres NULL != NULL, поэтому у записей БЕЗ номера акта конфликт не наступает
+-- никогда: `ON CONFLICT DO NOTHING` не срабатывает, и каждый недельный прогон
+-- вставляет копию. Docstring таски при этом обещает per-batch дедуп и двухшаговый
+-- upsert по (cad_num, doc_url) — ни того, ни другого в коде нет.
+--
+-- ЗАМЕР ПРОДА 2026-08-20 (до правки):
+--     строк всего                          297
+--     из них с act_number IS NULL          297   (то есть все)
+--     групп (cad_num, doc_url) с дублями    27
+--     максимум копий в группе               11
+--     лишних строк                         270   (91% таблицы)
+--
+-- Проверено, что ключ подходит: ни у одного cad_num нет более одного doc_url
+-- (max = 1), то есть NULLS NOT DISTINCT по (cad_num, act_number) НЕ схлопнет
+-- разные документы одного участка. Дубли внутри групп — точные копии: по одному
+-- различному значению reservation_kind и act_date на группу.
+--
+-- ЧТО УДАЛЯЕТСЯ. Строки-копии сверх первой (по возрастанию id) в каждой группе
+-- (cad_num, act_number) среди act_number IS NULL. Это порождение бага, а не
+-- пользовательские данные; таблица — кэш OCR-разбора PDF с сайта, пересобираемый
+-- прогоном таски. Первая строка группы (минимальный id) сохраняется целиком.
+--
+-- ПОЧЕМУ ОТДЕЛЬНОЙ МИГРАЦИЕЙ, а не правкой CREATE TABLE: та же причина, что в
+-- м.158 — исходный файл уже в _schema_migrations и на деплое пропускается.
+-- Прецеденты NULLS NOT DISTINCT в репо: м.110, м.125, м.140, м.158. Prod = PG16.4.
+-- Apply after: 188_regrant_quarter_price_index_fdw.sql
+
+BEGIN;
+
+-- #2752: блокирующий DDL обязан иметь lock_timeout — иначе ALTER TABLE встанет в
+-- очередь за чужой сессией и уведёт за собой запросы приложения. Таблица крошечная
+-- (297 строк), сам DDL мгновенный; пять секунд — про ОЖИДАНИЕ блокировки, не про
+-- работу. Не дождались — миграция падает, а не подвешивает прод.
+SET LOCAL lock_timeout = '5s';
+
+-- 1) Дедуп: оставляем строку с минимальным id в каждой группе.
+DELETE FROM land_reservation a
+ USING land_reservation b
+ WHERE a.act_number IS NULL
+   AND b.act_number IS NULL
+   AND a.cad_num = b.cad_num
+   AND a.id > b.id;
+
+-- 2) Пересоздаём констрейнт с NULL-семантикой, при которой ON CONFLICT матчит.
+ALTER TABLE land_reservation
+    DROP CONSTRAINT IF EXISTS uq_land_reservation_cad_act;
+
+ALTER TABLE land_reservation
+    ADD CONSTRAINT uq_land_reservation_cad_act
+    UNIQUE NULLS NOT DISTINCT (cad_num, act_number);
+
+COMMIT;