document-management/backend/migrations/versions/0002_add_restore_audit_action.py

"""增加文档恢复审计动作和回收站查询索引。

Revision ID: 0002_add_restore_audit_action
Revises: 0001_initial_schema
Create Date: 2026-07-26
"""

from __future__ import annotations

from collections.abc import Sequence

from alembic import op
import sqlalchemy as sa

revision: str = "0002_add_restore_audit_action"
down_revision: str | None = "0001_initial_schema"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

AUDIT_ACTIONS_V1 = (
    "LOGIN",
    "LOGOUT",
    "VIEW_DOCUMENT",
    "DOWNLOAD_DOCUMENT",
    "UPLOAD_DOCUMENT",
    "BATCH_IMPORT",
    "EDIT_DOCUMENT",
    "DELETE_DOCUMENT",
    "CREATE_CATEGORY",
    "EDIT_CATEGORY",
    "DELETE_CATEGORY",
    "CHANGE_PERMISSION",
    "BIND_ATTACHMENT",
    "UNBIND_ATTACHMENT",
)
RESTORE_ACTION = "RESTORE_DOCUMENT"
AUDIT_CONSTRAINT = "audit_action_type"
RECYCLE_INDEX = "ix_doc_document_recycle_deleted"


def _check_sql(actions: tuple[str, ...]) -> str:
    values = ", ".join(f"'{action}'" for action in actions)
    return f"action_type IN ({values})"


def _check_exists() -> bool:
    inspector = sa.inspect(op.get_bind())
    return any(
        item.get("name")
        in {AUDIT_CONSTRAINT, f"ck_sys_audit_log_{AUDIT_CONSTRAINT}"}
        for item in inspector.get_check_constraints("sys_audit_log")
    )


def _index_exists() -> bool:
    inspector = sa.inspect(op.get_bind())
    return any(
        item.get("name") == RECYCLE_INDEX
        for item in inspector.get_indexes("doc_document")
    )


def upgrade() -> None:
    if _check_exists():
        op.drop_constraint(
            AUDIT_CONSTRAINT,
            "sys_audit_log",
            type_="check",
        )
    op.create_check_constraint(
        AUDIT_CONSTRAINT,
        "sys_audit_log",
        _check_sql((*AUDIT_ACTIONS_V1, RESTORE_ACTION)),
    )
    if not _index_exists():
        op.create_index(
            RECYCLE_INDEX,
            "doc_document",
            ["is_deleted", "deleted_at", "id"],
            unique=False,
        )


def downgrade() -> None:
    restore_count = op.get_bind().scalar(
        sa.text(
            "SELECT COUNT(*) FROM sys_audit_log "
            "WHERE action_type = :action_type"
        ),
        {"action_type": RESTORE_ACTION},
    )
    if restore_count:
        raise RuntimeError(
            "存在RESTORE_DOCUMENT历史审计，拒绝无损降级；"
            "请保留0002或先按数据治理流程处理"
        )
    if _index_exists():
        op.drop_index(RECYCLE_INDEX, table_name="doc_document")
    if _check_exists():
        op.drop_constraint(
            AUDIT_CONSTRAINT,
            "sys_audit_log",
            type_="check",
        )
    op.create_check_constraint(
        AUDIT_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V1),
    )