document-management/backend/migrations/versions/0026_borrow_permission.py

"""借阅授权改造：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'))"
        )