feat(migrations): откат v5.19 с отказом на несовместимых данных и журналом потерь

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01BJKMLHdxMwSUeYcdkwqX4W

api/alembic/versions/0007_dbml_v5_19.py

@@ -11,13 +11,15 @@
 - `header_source` — из `*_original_value` и `inn_confidence` (P3-1), `organization_name_original_value` =
   `organization_name`, `organization_name_modified` = false (P3-2); `form_variant` = NULL (набор §7 — полный);
 - `file_ordinal` — по `(created_at, id)` внутри файла с 0.
-Уникальные индексы создаются после заполнения. Откат (`downgrade`) — задача A2 плана 3.
+Уникальные индексы создаются после заполнения. Откат: сперва проверки (повтор/пустой код строки, две формы типа
+в файле, `crm_id`, правка наименования — отказ `RuntimeError`), затем журнал потерь и DDL.
 
 Revision ID: 0007_dbml_v5_19
 Revises: 0006_dbml_v5_18
 Create Date: 2026-10-18
 """
 
+import logging
 from collections.abc import Sequence
 
 from alembic import op
@@ -187,6 +189,113 @@ def upgrade() -> None:
     _create_document_issue()
 
 
+# (причина отказа, SQL подсчёта) — порядок = порядок в сообщении
+_REFUSALS: tuple[tuple[str, str], ...] = (
+    (
+        "повтор кода строки в форме",
+        f"SELECT count(*) FROM (SELECT 1 FROM {S}.financial_statement_line WHERE line_code IS NOT NULL "
+        "GROUP BY meta_id, line_code HAVING count(*) > 1) t",
+    ),
+    ("строки формы без кода", f"SELECT count(*) FROM {S}.financial_statement_line WHERE line_code IS NULL"),
+    (
+        "две формы одного типа в файле",
+        f"SELECT count(*) FROM (SELECT 1 FROM {S}.document WHERE document_type IN ('BALANCE', 'INCOME') "
+        "GROUP BY stored_file_id, document_type HAVING count(*) > 1) t",
+    ),
+    ("crm_id ГСК/ГСЗ задан", f"SELECT count(*) FROM {S}.related_group WHERE crm_id IS NOT NULL"),
+    (
+        "наименование организации правилось",
+        f"SELECT count(*) FROM {S}.financial_statement_meta WHERE organization_name_modified IS TRUE",
+    ),
+)
+
+# (таблица, колонка, условие «непусто») — колонки v5.19, значения которых откат теряет
+_LOST_COLUMNS: tuple[tuple[str, str, str], ...] = (
+    ("financial_statement_line", "row_index", "IS NOT NULL"),
+    ("financial_statement_line", "source_ref", "IS NOT NULL"),
+    ("financial_statement_line", "line_status", "<> 'ok'"),
+    ("financial_statement_line", "line_code_text", "IS NOT NULL"),
+    ("financial_statement_line", "parent_row_index", "IS NOT NULL"),
+    ("financial_statement_line", "needs_review", "IS TRUE"),
+    ("financial_statement_line", "is_confirmed", "IS TRUE"),
+    ("financial_statement_value", "source_text", "IS NOT NULL"),
+    ("financial_statement_value", "needs_review", "IS TRUE"),
+    ("financial_statement_value", "is_confirmed", "IS TRUE"),
+    ("financial_statement_value", "confirmed_by", "IS NOT NULL"),
+    ("financial_statement_value", "confirmed_at", "IS NOT NULL"),
+    ("financial_statement_column", "period_kind", "IS NOT NULL"),
+    ("financial_statement_column", "label_confidence", "IS NOT NULL"),
+    ("financial_statement_meta", "form_variant", "IS NOT NULL"),
+    ("financial_statement_meta", "needs_review_fields", "IS NOT NULL"),
+    ("financial_statement_meta", "header_source", "IS NOT NULL"),
+    ("financial_statement_meta", "organization_name_original_value", "IS NOT NULL"),
+    ("financial_statement_meta", "organization_name_modified", "IS TRUE"),
+    ("document", "file_ordinal", "<> 0"),
+    ("recognition_job", "statements_version", "IS NOT NULL"),
+    ("related_group", "crm_id", "IS NOT NULL"),
+)
+
+
+def _scalar(sql: str) -> int:
+    return int(op.get_bind().exec_driver_sql(sql).scalar_one())
+
+
+def _check_downgrade_allowed() -> None:
+    """Все проверки — до любого DDL: несовместимые данные v5.18 не вместить, откат отказывает целиком."""
+    problems = [f"{reason}: {count}" for reason, sql in _REFUSALS if (count := _scalar(sql))]
+    if problems:
+        raise RuntimeError("откат v5.19 невозможен: " + "; ".join(problems))
+
+
+def _log_lost_values() -> None:
+    lost = [
+        f"{table}.{column}: {count}"
+        for table, column, condition in _LOST_COLUMNS
+        if (count := _scalar(f"SELECT count(*) FROM {S}.{table} WHERE {column} {condition}"))
+    ]
+    if count := _scalar(f"SELECT count(*) FROM {S}.document_issue"):
+        lost.append(f"document_issue: {count}")
+    if lost:
+        logging.getLogger("alembic.runtime.migration").warning(
+            "откат v5.19: удалены непустые значения — %s", "; ".join(lost)
+        )
+
+
 def downgrade() -> None:
-    # откат (охранные проверки, потеря данных, журнал) — задача A2 плана 3
+    _check_downgrade_allowed()
-    raise NotImplementedError("downgrade 0007_dbml_v5_19 ещё не реализован (задача A2 плана 3)")
+    _log_lost_values()
+    op.execute(f"DROP TABLE {S}.document_issue")
+    for index in (
+        "uq_document_stored_file_id_file_ordinal",
+        "uq_financial_statement_line_meta_id_source_ref",
+        "ix_financial_statement_line_meta_id_line_code",
+        "uq_document_file_type_account",
+    ):
+        op.execute(f"DROP INDEX {S}.{index}")
+    op.execute(
+        "CREATE UNIQUE INDEX uq_financial_statement_line_meta_id_line_code "
+        f"ON {S}.financial_statement_line (meta_id, line_code)"
+    )
+    op.execute(
+        "CREATE UNIQUE INDEX uq_document_file_type_account "
+        f"ON {S}.document (stored_file_id, document_type, osv_account_code) NULLS NOT DISTINCT"
+    )
+    drops = {
+        "financial_statement_line": (
+            "row_index", "source_ref", "line_status", "line_code_text", "parent_row_index",
+            "needs_review", "is_confirmed",
+        ),
+        "financial_statement_value": ("source_text", "needs_review", "is_confirmed", "confirmed_by", "confirmed_at"),
+        "financial_statement_column": ("period_kind", "label_confidence"),
+        "financial_statement_meta": (
+            "form_variant", "needs_review_fields", "header_source", "organization_name_original_value",
+            "organization_name_modified",
+        ),
+        "document": ("file_ordinal",),
+        "recognition_job": ("statements_version",),
+        "related_group": ("crm_id",),
+    }  # fmt: skip
+    for table, columns in drops.items():
+        op.execute(f"ALTER TABLE {S}.{table} " + ", ".join(f"DROP COLUMN {column}" for column in columns))
+    for name in reversed(ENUMS_V5_19):
+        op.execute(f"DROP TYPE {S}.{name}")

api/tests/test_migrations.py

@@ -869,6 +869,162 @@ async def test_v5_19_constraints(scratch_db_url: str) -> None:
         await engine.dispose()
 
 
+_INSERT_2000 = (
+    "INSERT INTO dkb.financial_statement_line (id, meta_id, line_code) VALUES (gen_random_uuid(), :meta, '2000')"
+)
+_SEED_KEYS = (
+    "le",
+    "s3",
+    "file",
+    "job",
+    "income",
+    "balance",
+    "meta",
+    "c1",
+    "c2",
+    "l1600",
+    "l1100",
+    "l1250",
+    "grp",
+    "rg",
+)
+
+
+async def _count(conn: AsyncConnection, table: str) -> int:
+    return (await _scalars(conn, f"SELECT count(*) FROM dkb.{table}"))[0]
+
+
+async def _expect_downgrade_0007_refused(scratch_db_url: str, mutate_sql: str, reason: str) -> None:
+    """Данные v5.19, несовместимые с v5.18: откат отказывает целиком, ревизия и данные не меняются."""
+    engine = create_async_engine(scratch_db_url)
+    ids = {k: uuid.uuid4() for k in _SEED_KEYS}
+    try:
+        await alembic(scratch_db_url, "upgrade", V5_18)
+        async with engine.begin() as conn:
+            await _seed_v5_18_statements(conn, ids)
+        await alembic(scratch_db_url, "upgrade", "head")
+        async with engine.begin() as conn:
+            await conn.execute(text(mutate_sql.format(**{k: str(v) for k, v in ids.items()})))
+            lines_before = await _count(conn, "financial_statement_line")
+            documents_before = await _count(conn, "document")
+        with pytest.raises(AssertionError) as error:
+            await alembic(scratch_db_url, "downgrade", V5_18)
+        assert f"откат v5.19 невозможен: {reason}: 1" in str(error.value)
+        async with engine.connect() as conn:
+            assert await _scalars(conn, "SELECT version_num FROM dkb.alembic_version") == ["0007_dbml_v5_19"]
+            assert await _count(conn, "financial_statement_line") == lines_before  # данные не изменены
+            assert await _count(conn, "document") == documents_before
+            assert await _scalars(
+                conn, "SELECT count(*) FROM information_schema.tables WHERE table_name = 'document_issue'"
+            ) == [1]
+    finally:
+        await engine.dispose()
+
+
+@pytest.mark.integration
+async def test_downgrade_0007_refuses_duplicate_code(scratch_db_url: str) -> None:
+    await _expect_downgrade_0007_refused(
+        scratch_db_url,
+        "INSERT INTO dkb.financial_statement_line (id, meta_id, line_code) "
+        "VALUES (gen_random_uuid(), '{meta}', '1100')",
+        "повтор кода строки в форме",
+    )
+
+
+@pytest.mark.integration
+async def test_downgrade_0007_refuses_null_code(scratch_db_url: str) -> None:
+    await _expect_downgrade_0007_refused(
+        scratch_db_url,
+        "INSERT INTO dkb.financial_statement_line (id, meta_id, line_code, line_status) "
+        "VALUES (gen_random_uuid(), '{meta}', NULL, 'unreadable')",
+        "строки формы без кода",
+    )
+
+
+@pytest.mark.integration
+async def test_downgrade_0007_refuses_two_forms_of_type(scratch_db_url: str) -> None:
+    await _expect_downgrade_0007_refused(
+        scratch_db_url,
+        "INSERT INTO dkb.document (id, stored_file_id, legal_entity_id, document_type, status, file_ordinal) "
+        "VALUES (gen_random_uuid(), '{file}', '{le}', 'BALANCE', 'NEEDS_CHECK', 2)",
+        "две формы одного типа в файле",
+    )
+
+
+@pytest.mark.integration
+async def test_downgrade_0007_refuses_related_group_crm_id(scratch_db_url: str) -> None:
+    await _expect_downgrade_0007_refused(
+        scratch_db_url, "UPDATE dkb.related_group SET crm_id = 'CRM-1' WHERE id = '{rg}'", "crm_id ГСК/ГСЗ задан"
+    )
+
+
+@pytest.mark.integration
+async def test_downgrade_0007_refuses_modified_organization_name(scratch_db_url: str) -> None:
+    await _expect_downgrade_0007_refused(
+        scratch_db_url,
+        "UPDATE dkb.financial_statement_meta SET organization_name_modified = true WHERE id = '{meta}'",
+        "наименование организации правилось",
+    )
+
+
+@pytest.mark.integration
+async def test_downgrade_0007_drops_signals_and_logs_counts(scratch_db_url: str) -> None:
+    """Без несовместимых данных откат проходит, журнал называет потерянное; цикл замыкается."""
+    engine = create_async_engine(scratch_db_url)
+    ids = {k: uuid.uuid4() for k in _SEED_KEYS}
+    try:
+        await alembic(scratch_db_url, "upgrade", V5_18)
+        async with engine.begin() as conn:
+            await _seed_v5_18_statements(conn, ids)
+        await alembic(scratch_db_url, "upgrade", "head")
+        async with engine.begin() as conn:
+            await conn.execute(
+                text("UPDATE dkb.financial_statement_value SET is_confirmed = true WHERE amount IN (1, 2)")
+            )
+            await conn.execute(
+                text(
+                    "INSERT INTO dkb.document_issue (id, document_id, kind, code, severity, message, source, "
+                    "created_at) "
+                    "VALUES (gen_random_uuid(), :balance, 'warning', 'X', 'info', 'm', 'core', now())"
+                ),
+                ids,
+            )
+        output = await alembic(scratch_db_url, "downgrade", V5_18)
+        assert "откат v5.19: удалены непустые значения" in output
+        assert "financial_statement_value.is_confirmed: 2" in output
+        assert "document_issue: 1" in output
+        async with engine.connect() as conn:
+            assert await _scalars(conn, "SELECT version_num FROM dkb.alembic_version") == [V5_18]
+            assert await _scalars(
+                conn,
+                "SELECT count(*) FROM information_schema.columns WHERE table_schema = 'dkb' AND "
+                "((table_name = 'financial_statement_line' "
+                "AND column_name IN ('row_index', 'source_ref', 'is_confirmed')) "
+                "OR (table_name = 'document' AND column_name = 'file_ordinal') "
+                "OR (table_name = 'related_group' AND column_name = 'crm_id'))",
+            ) == [0]
+            assert await _scalars(
+                conn, "SELECT count(*) FROM information_schema.tables WHERE table_name = 'document_issue'"
+            ) == [0]
+            assert await _scalars(conn, "SELECT count(*) FROM pg_type WHERE typname = 'document_issue_kind'") == [0]
+        await engine.dispose()
+        async with engine.begin() as conn:  # уникальность (meta_id, line_code) вернулась
+            await conn.execute(
+                text(_INSERT_2000),
+                ids,
+            )
+        with pytest.raises(IntegrityError):
+            async with engine.begin() as conn:
+                await conn.execute(
+                    text(_INSERT_2000),
+                    ids,
+                )
+        await engine.dispose()
+        await alembic(scratch_db_url, "upgrade", "head")
+    finally:
+        await engine.dispose()
+
+
 @pytest.mark.integration
 async def test_upgrade_0005_adds_nullable_kpp_columns(scratch_db_url: str) -> None:
     """0005: `kpp varchar` nullable у `legal_entity` и `legal_entity_history`, существующие ЮЛ — с `kpp IS NULL`;