"""借阅授权改造：doc_permission 瘦身为用户借阅表，doc_document 下线可见范围。 Revision ID: 0026_borrow_permission Revises: 0025_move_audit_actions Create Date: 2026-09-03 变更要点： 1. 借阅语义（契约1.19）：授权界面三种可见范围（全员/组织/自定义）全部下线， 只保留面向用户的限时借阅授权（仅查看/下载两个操作位，时限最长7天）； 2. doc_permission 瘦身：物理删除全部 ORG 主体行；删列 subject_type / can_edit / can_manage_permission / can_delete；唯一约束收敛为 (document_id, subject_id, active_marker)；索引 ix_doc_permission_subject_active 收敛为 (subject_id, is_deleted)；新增 expires_at 到期时刻（NOT NULL， 存量活跃 USER 行初始化为迁移时刻+7天）与审批单凭证四列 voucher_file_name / voucher_storage_path / voucher_file_size / voucher_uploaded_at； 3. doc_document：删除 visibility_type 列（连同引用它的 CHECK 约束）， shape 约束重建为 v3（附件分支去掉 visibility 条件）； 存量“全员可见/组织可见”文档回退角色权限矩阵控制； 4. sys_audit_log：action_type CHECK 白名单 +2 （UPLOAD_PERMISSION_VOUCHER / DELETE_PERMISSION_VOUCHER，38+2=40 项）； 5. 达梦兼容：探测统一走 SQLAlchemy inspector、删除以库中实际名执行 （规避 naming convention 渲染差异与达梦大写名）；存量初始化时刻由 Python 侧计算参数化传入（规避 DATE_ADD 方言差异）； 6. 升级幂等：列/约束/索引先探测；expires_at 初始化以 1970 哨兵为标记可重跑 （与 Flyway V11 并行执行结果一致）；降级仅重建结构（ORG 行已物理删除 不可恢复），且拒绝已存在凭证审计记录的库。 """ from __future__ import annotations from collections.abc import Sequence from datetime import datetime, timedelta, timezone import sqlalchemy as sa from alembic import op revision: str = "0026_borrow_permission" down_revision: str | None = "0025_move_audit_actions" branch_labels: str | Sequence[str] | None = None depends_on: str | Sequence[str] | None = None _GRACE = timedelta(days=7) _SENTINEL = "1970-01-01 00:00:00" _DROPPED_COLUMNS = ( "subject_type", "can_edit", "can_manage_permission", "can_delete", ) _VOUCHER_COLUMNS = ( ("voucher_file_name", sa.String(length=255)), ("voucher_storage_path", sa.String(length=512)), ("voucher_file_size", sa.BigInteger()), ("voucher_uploaded_at", sa.DateTime()), ) # 0025 的 38 项全集 + 凭证动作 2 项 = 40 项 AUDIT_ACTIONS_V9 = ( "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", "REORDER_DOCUMENT", "REORDER_ATTACHMENT", "REORDER_CATEGORY", "CREATE_ROLE", "EDIT_ROLE", "DELETE_ROLE", "CHANGE_ROLE_PERMISSION", "CHANGE_ROLE_USER", "DOCUMENT_PASSWORD_SET", "DOCUMENT_PASSWORD_REMOVE", "CREATE_IP_ENTRY", "EDIT_IP_ENTRY", "DELETE_IP_ENTRY", "ASSIGN_ORG_IP", "MOVE_DOCUMENT", "MOVE_ATTACHMENT", "UPLOAD_PERMISSION_VOUCHER", "DELETE_PERMISSION_VOUCHER", ) AUDIT_ACTIONS_V8 = AUDIT_ACTIONS_V9[:-2] ACTION_CONSTRAINT = "audit_action_type" _PERMISSION_TABLE = "doc_permission" _DOCUMENT_TABLE = "doc_document" # 唯一约束/索引的目标形态（升级后） _NEW_UNIQUE = "uq_doc_permission_active" _NEW_UNIQUE_COLUMNS = {"document_id", "subject_id", "active_marker"} _NEW_INDEX = "ix_doc_permission_subject_active" _NEW_INDEX_COLUMNS = ["subject_id", "is_deleted"] # shape v3：附件分支去掉 visibility_type 条件（密级放宽沿用 0020 v2 口径） _SHAPE_V3 = "ck_doc_document_type_shape_v3" _SHAPE_V3_DDL = ( "ALTER TABLE doc_document ADD CONSTRAINT " + _SHAPE_V3 + " CHECK (" "((document_type = 'MAIN') AND (parent_document_id IS NULL) " "AND (category_id IS NOT NULL) AND (attachment_type IS NULL)) " "OR ((document_type = 'SUB_PLAN') AND (parent_document_id IS NOT NULL) " "AND (root_document_id IS NOT NULL) AND (category_id IS NOT NULL) " "AND (attachment_type IS NULL)) " "OR ((document_type = 'ATTACHMENT') AND (parent_document_id IS NULL) " "AND (root_document_id IS NULL) AND (category_id IS NULL) " "AND (attachment_type IS NOT NULL)))" ) def _columns_of(table: str) -> set[str]: inspector = sa.inspect(op.get_bind()) return {column["name"] for column in inspector.get_columns(table)} def _unique_using(table: str, column: str) -> list[str]: """返回引用指定列的唯一约束实际名（兼容 convention 渲染与达梦大写名）。""" inspector = sa.inspect(op.get_bind()) return [ item["name"] for item in inspector.get_unique_constraints(table) if column in {col.lower() for col in item.get("column_names", [])} ] def _checks_referencing(table: str, column: str) -> list[str]: """返回 sqltext 引用指定列的 CHECK 约束实际名。""" inspector = sa.inspect(op.get_bind()) return [ item["name"] for item in inspector.get_check_constraints(table) if column in (item.get("sqltext") or "").lower() ] def _indexes_using(table: str, column: str) -> list[str]: """返回引用指定列的索引实际名。""" inspector = sa.inspect(op.get_bind()) return [ item["name"] for item in inspector.get_indexes(table) if column in {col.lower() for col in item.get("column_names", [])} ] def _run(statement: str) -> None: op.get_bind().execute(sa.text(statement)) def _drop_unique(table: str, name: str) -> None: # MySQL 的唯一约束即唯一索引，须走 DROP INDEX；达梦为 DROP CONSTRAINT。 if op.get_bind().dialect.name == "mysql": _run(f"ALTER TABLE {table} DROP INDEX {name}") else: _run(f"ALTER TABLE {table} DROP CONSTRAINT {name}") def _drop_index(table: str, name: str) -> None: if op.get_bind().dialect.name == "mysql": _run(f"ALTER TABLE {table} DROP INDEX {name}") else: _run(f"DROP INDEX {name}") def _has_unique_columns(table: str, columns: set[str]) -> bool: inspector = sa.inspect(op.get_bind()) for item in inspector.get_unique_constraints(table): existing = {col.lower() for col in item.get("column_names", [])} if existing == columns: return True return False def _has_index_columns(table: str, columns: list[str]) -> bool: inspector = sa.inspect(op.get_bind()) wanted = [col.lower() for col in columns] for item in inspector.get_indexes(table): existing = [col.lower() for col in item.get("column_names", [])] if existing == wanted: return True return False def _check_names(table: str) -> set[str]: return { item["name"].lower() for item in sa.inspect(op.get_bind()).get_check_constraints(table) } def _check_sql(values: tuple[str, ...]) -> str: # 安全性：values 为本文件内硬编码的枚举常量，非用户输入，无注入风险 inline = ", ".join(f"'{v}'" for v in values) return f"action_type IN ({inline})" def _drop_action_check() -> None: # op.drop_constraint 按 metadata.naming_convention 前缀化约束名， # 须传原始约束名渲染最终名；达梦返回的约束名为大写，统一小写比较。 existing = { item["name"].lower() for item in sa.inspect(op.get_bind()).get_check_constraints("sys_audit_log") } if f"ck_sys_audit_log_{ACTION_CONSTRAINT}" in existing: op.drop_constraint(ACTION_CONSTRAINT, "sys_audit_log", type_="check") def _upgrade_permission_table() -> None: existing = _columns_of(_PERMISSION_TABLE) # 1) ORG 主体行物理删除（组织授权语义整体下线，软删行一并清除） if "subject_type" in existing: _run(f"DELETE FROM {_PERMISSION_TABLE} WHERE subject_type = 'ORG'") # 2) 拆除引用 subject_type 的约束与索引（以库中实际名执行） for name in _unique_using(_PERMISSION_TABLE, "subject_type"): _drop_unique(_PERMISSION_TABLE, name) for name in _checks_referencing(_PERMISSION_TABLE, "subject_type"): _run(f"ALTER TABLE {_PERMISSION_TABLE} DROP CONSTRAINT {name}") for name in _indexes_using(_PERMISSION_TABLE, "subject_type"): _drop_index(_PERMISSION_TABLE, name) # 3) 删列 + 加列（幂等探测） for name in _DROPPED_COLUMNS: if name in existing: op.drop_column(_PERMISSION_TABLE, name) if "expires_at" not in existing: op.add_column( _PERMISSION_TABLE, sa.Column( "expires_at", sa.DateTime(), nullable=False, # 1970 哨兵=fail-safe：漏赋值路径插入的行立即过期拒绝访问 server_default=sa.text(f"'{_SENTINEL}'"), ), ) for name, type_ in _VOUCHER_COLUMNS: if name not in existing: op.add_column(_PERMISSION_TABLE, sa.Column(name, type_, nullable=True)) # 4) 存量活跃 USER 行初始化为迁移时刻+7天（UTC，与业务时间戳口径一致； # 参数化规避 DATE_ADD 方言差异） deadline = ( datetime.now(timezone.utc).replace(tzinfo=None) + _GRACE ).replace(microsecond=0) op.get_bind().execute( sa.text( f"UPDATE {_PERMISSION_TABLE} SET expires_at = :deadline " f"WHERE is_deleted = 0 AND expires_at = :sentinel" ), {"deadline": deadline, "sentinel": _SENTINEL}, ) # 5) 重建唯一约束与索引（收敛主体维度） if not _has_unique_columns(_PERMISSION_TABLE, _NEW_UNIQUE_COLUMNS): _run( f"ALTER TABLE {_PERMISSION_TABLE} ADD CONSTRAINT {_NEW_UNIQUE} " "UNIQUE (document_id, subject_id, active_marker)" ) if not _has_index_columns(_PERMISSION_TABLE, _NEW_INDEX_COLUMNS): _run( f"CREATE INDEX {_NEW_INDEX} ON {_PERMISSION_TABLE} " "(subject_id, is_deleted)" ) def _upgrade_document_table() -> None: existing = _columns_of(_DOCUMENT_TABLE) # 1) 拆除引用 visibility_type 的 CHECK 与索引（命中枚举约束与两代 shape） for name in _checks_referencing(_DOCUMENT_TABLE, "visibility_type"): _run(f"ALTER TABLE {_DOCUMENT_TABLE} DROP CONSTRAINT {name}") for name in _indexes_using(_DOCUMENT_TABLE, "visibility_type"): _drop_index(_DOCUMENT_TABLE, name) # 2) 删列 + 重建 shape v3（附件分支不再约束可见范围） if "visibility_type" in existing: op.drop_column(_DOCUMENT_TABLE, "visibility_type") if _SHAPE_V3.lower() not in _check_names(_DOCUMENT_TABLE): _run(_SHAPE_V3_DDL) def upgrade() -> None: _upgrade_permission_table() _upgrade_document_table() _drop_action_check() op.create_check_constraint( ACTION_CONSTRAINT, "sys_audit_log", _check_sql(AUDIT_ACTIONS_V9), ) def downgrade() -> None: # ORG 主体行已物理删除、expires_at 数据在删列时丢失，降级仅重建结构。 voucher_rows = op.get_bind().scalar( sa.text( "SELECT COUNT(*) FROM sys_audit_log WHERE action_type IN " "('UPLOAD_PERMISSION_VOUCHER', 'DELETE_PERMISSION_VOUCHER')" ) ) if voucher_rows: raise RuntimeError( "0026已存在审批单凭证审计记录，拒绝有损降级；" "请先备份并按数据治理流程处理" ) _drop_action_check() op.create_check_constraint( ACTION_CONSTRAINT, "sys_audit_log", _check_sql(AUDIT_ACTIONS_V8), ) # doc_document：重建 visibility_type（存量统一回退全员可见） for name in _checks_referencing(_DOCUMENT_TABLE, "document_type"): if name.lower() == _SHAPE_V3.lower(): _run(f"ALTER TABLE {_DOCUMENT_TABLE} DROP CONSTRAINT {name}") document_columns = _columns_of(_DOCUMENT_TABLE) if "visibility_type" not in document_columns: op.add_column( _DOCUMENT_TABLE, sa.Column( "visibility_type", sa.String(length=20), nullable=False, server_default=sa.text("'ALL_AUTHENTICATED'"), ), ) existing_checks = _check_names(_DOCUMENT_TABLE) if "ck_doc_document_document_visibility_type" not in existing_checks: _run( "ALTER TABLE doc_document ADD CONSTRAINT " "ck_doc_document_document_visibility_type CHECK (" "visibility_type IN ('ALL_AUTHENTICATED', 'ORGANIZATION', 'CUSTOM'))" ) if "ck_doc_document_type_shape_v2" not in existing_checks: _run( "ALTER TABLE doc_document ADD CONSTRAINT ck_doc_document_type_shape_v2 " "CHECK (" "((document_type = 'MAIN') AND (parent_document_id IS NULL) " "AND (category_id IS NOT NULL) AND (attachment_type IS NULL)) " "OR ((document_type = 'SUB_PLAN') AND (parent_document_id IS NOT NULL) " "AND (root_document_id IS NOT NULL) AND (category_id IS NOT NULL) " "AND (attachment_type IS NULL)) " "OR ((document_type = 'ATTACHMENT') AND (parent_document_id IS NULL) " "AND (root_document_id IS NULL) AND (category_id IS NULL) " "AND (attachment_type IS NOT NULL) " "AND (visibility_type = 'ALL_AUTHENTICATED')))" ) # doc_permission：拆除新约束/索引，恢复旧结构（subject_type 统一回填 USER） uniques = sa.inspect(op.get_bind()).get_unique_constraints(_PERMISSION_TABLE) for item in uniques: columns = {col.lower() for col in item.get("column_names", [])} if columns == _NEW_UNIQUE_COLUMNS: _drop_unique(_PERMISSION_TABLE, item["name"]) break if _has_index_columns(_PERMISSION_TABLE, _NEW_INDEX_COLUMNS): _drop_index(_PERMISSION_TABLE, _NEW_INDEX) permission_columns = _columns_of(_PERMISSION_TABLE) for name, type_ in _VOUCHER_COLUMNS: if name in permission_columns: op.drop_column(_PERMISSION_TABLE, name) if "expires_at" in permission_columns: op.drop_column(_PERMISSION_TABLE, "expires_at") if "subject_type" not in permission_columns: op.add_column( _PERMISSION_TABLE, sa.Column( "subject_type", sa.String(length=20), nullable=False, server_default=sa.text("'USER'"), ), ) for name in ("can_edit", "can_manage_permission", "can_delete"): if name not in permission_columns: op.add_column( _PERMISSION_TABLE, sa.Column( name, sa.Boolean(), nullable=False, server_default=sa.text("0"), ), ) if not _has_unique_columns( _PERMISSION_TABLE, {"document_id", "subject_type", "subject_id", "active_marker"}, ): _run( "ALTER TABLE doc_permission ADD CONSTRAINT uq_doc_permission_active " "UNIQUE (document_id, subject_type, subject_id, active_marker)" ) if not _has_index_columns( _PERMISSION_TABLE, ["subject_type", "subject_id", "is_deleted"] ): _run( "CREATE INDEX ix_doc_permission_subject_active ON doc_permission " "(subject_type, subject_id, is_deleted)" ) if "ck_doc_permission_permission_subject_type" not in ( item["name"].lower() for item in sa.inspect(op.get_bind()).get_check_constraints(_PERMISSION_TABLE) ): _run( "ALTER TABLE doc_permission ADD CONSTRAINT " "ck_doc_permission_permission_subject_type CHECK " "(subject_type IN ('ORG', 'USER'))" )