"""机构与用户维护：复合唯一索引 + 审计CHECK扩展。 Revision ID: 0004_organization_user_maintenance Revises: 0003_q3_business_alignment_and_content_search Create Date: 2026-08-12 变更要点： 1. sys_organization / doc_category 引入 active_marker 虚拟生成列， 将 org_code / category_code 的唯一性约束改造为"仅活跃行唯一"， 支持软删除后重建同名条目，与既有 attachment_binding / permission 表保持一致。 2. sys_audit_log 的 action_type / target_type CHECK 约束扩展， 覆盖机构与用户维护相关的新枚举值。 """ from __future__ import annotations from collections.abc import Sequence from alembic import op import sqlalchemy as sa revision: str = "0004_organization_user_maintenance" down_revision: str | None = "0003_q3_business_alignment_and_content_search" branch_labels: str | Sequence[str] | None = None depends_on: str | Sequence[str] | None = None # 既有审计动作（与 0002 对齐）+ 本期新增动作 AUDIT_ACTIONS_V2 = ( "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_DOCUMENT", "CREATE_ORGANIZATION", "EDIT_ORGANIZATION", "DELETE_ORGANIZATION", "CREATE_USER", "EDIT_USER", "DELETE_USER", "RESET_USER_PASSWORD", ) AUDIT_ACTIONS_V1 = AUDIT_ACTIONS_V2[:15] AUDIT_TARGETS_V2 = ( "AUTH", "CATEGORY", "DOCUMENT", "ATTACHMENT", "PERMISSION", "ATTACHMENT_BINDING", "ORGANIZATION", "USER", ) AUDIT_TARGETS_V1 = AUDIT_TARGETS_V2[:6] ACTION_CONSTRAINT = "audit_action_type" TARGET_CONSTRAINT = "audit_target_type" def _check_sql(values: tuple[str, ...]) -> str: # 安全性：values 为本文件内硬编码的枚举常量，非用户输入，无注入风险 inline = ", ".join(f"'{v}'" for v in values) return f"action_type IN ({inline})" def _target_check_sql(values: tuple[str, ...]) -> str: # 安全性：values 为本文件内硬编码的枚举常量，非用户输入，无注入风险 inline = ", ".join(f"'{v}'" for v in values) return f"(target_type IS NULL OR target_type IN ({inline}))" def _existing_constraint_names(table: str) -> set[str]: inspector = sa.inspect(op.get_bind()) return {item["name"] for item in inspector.get_check_constraints(table)} def _existing_index_names(table: str) -> set[str]: inspector = sa.inspect(op.get_bind()) return {item["name"] for item in inspector.get_indexes(table)} def _existing_columns(table: str) -> set[str]: inspector = sa.inspect(op.get_bind()) return {item["name"] for item in inspector.get_columns(table)} def _drop_audit_checks(existing: set[str]) -> None: # op.drop_constraint 会按 metadata.naming_convention 再次前缀化约束名 # （"ck_%(table_name)s_%(constraint_name)s"），因此必须传入原始约束名， # 让 SQLAlchemy 渲染出最终名；数据库实际存的也是渲染后的最终名。 # 这里用最终名判断是否实际存在，避免被二次前缀化导致MySQL 3821。 for raw_name, final_name in ( (ACTION_CONSTRAINT, f"ck_sys_audit_log_{ACTION_CONSTRAINT}"), (TARGET_CONSTRAINT, f"ck_sys_audit_log_{TARGET_CONSTRAINT}"), ): if final_name in existing: op.drop_constraint(raw_name, "sys_audit_log", type_="check") # 复合唯一索引迁移逻辑 -------------------------------------------------- def _migrate_unique_to_active( table: str, code_column: str, old_unique_index: str, new_unique_index: str, ) -> None: """将单列唯一索引升级为 (code, active_marker) 复合唯一索引。""" columns = _existing_columns(table) indexes = _existing_index_names(table) if "active_marker" not in columns: op.execute( sa.text( f"ALTER TABLE `{table}` ADD COLUMN `active_marker` " f"INT GENERATED ALWAYS AS (IF(is_deleted = 0, 1, NULL)) VIRTUAL" ) ) if old_unique_index in indexes: op.drop_index(old_unique_index, table_name=table) if new_unique_index not in indexes: op.create_index( new_unique_index, table, [code_column, "active_marker"], unique=True, ) def _revert_unique_to_simple( table: str, code_column: str, old_unique_index: str, new_unique_index: str, ) -> None: indexes = _existing_index_names(table) if new_unique_index in indexes: op.drop_index(new_unique_index, table_name=table) if old_unique_index not in indexes: op.create_index( old_unique_index, table, [code_column], unique=True, ) columns = _existing_columns(table) if "active_marker" in columns: op.drop_column(table, "active_marker") # 升级 / 降级 ---------------------------------------------------------- def upgrade() -> None: # 1) 复合唯一索引（机构 + 分类同步改造） _migrate_unique_to_active( table="sys_organization", code_column="org_code", old_unique_index="uq_sys_organization_org_code", new_unique_index="uq_sys_organization_org_code_active", ) _migrate_unique_to_active( table="doc_category", code_column="category_code", old_unique_index="uq_doc_category_category_code", new_unique_index="uq_doc_category_category_code_active", ) # 2) 审计 CHECK 扩展 _drop_audit_checks(_existing_constraint_names("sys_audit_log")) op.create_check_constraint( ACTION_CONSTRAINT, "sys_audit_log", _check_sql(AUDIT_ACTIONS_V2), ) op.create_check_constraint( TARGET_CONSTRAINT, "sys_audit_log", _target_check_sql(AUDIT_TARGETS_V2), ) def downgrade() -> None: bind = op.get_bind() new_action_rows = bind.scalar( sa.text( "SELECT COUNT(*) FROM sys_audit_log " "WHERE action_type IN (" "'CREATE_ORGANIZATION','EDIT_ORGANIZATION','DELETE_ORGANIZATION'," "'CREATE_USER','EDIT_USER','DELETE_USER','RESET_USER_PASSWORD')" ) ) new_target_rows = bind.scalar( sa.text( "SELECT COUNT(*) FROM sys_audit_log " "WHERE target_type IN ('ORGANIZATION','USER')" ) ) if new_action_rows or new_target_rows: raise RuntimeError( "0004已产生机构/用户维护相关审计记录，拒绝有损降级；" "请先备份并按数据治理流程处理" ) _drop_audit_checks(_existing_constraint_names("sys_audit_log")) op.create_check_constraint( ACTION_CONSTRAINT, "sys_audit_log", _check_sql(AUDIT_ACTIONS_V1), ) op.create_check_constraint( TARGET_CONSTRAINT, "sys_audit_log", _target_check_sql(AUDIT_TARGETS_V1), ) _revert_unique_to_simple( table="sys_organization", code_column="org_code", old_unique_index="uq_sys_organization_org_code", new_unique_index="uq_sys_organization_org_code_active", ) _revert_unique_to_simple( table="doc_category", code_column="category_code", old_unique_index="uq_doc_category_category_code", new_unique_index="uq_doc_category_category_code_active", )