"""Бизнес-ключ 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()