document-management/backend/migrations/versions/0029_system_management.py

"""系统管理功能（契约1.23）：角色管理位加列、采样表、审计白名单扩展。

Revision ID: 0029_system_management
Revises: 0028_backup_audit_actions
Create Date: 2026-09-07

变更要点：
1. sys_role 新增 manage_system / manage_audit_log 布尔管理位（默认 0，
   与 manage_category_edit 等同列族；内置 admin 账号恒有，不受开关约束）；
2. 新建 sys_metric_sample 资源采样表（CPU/内存 30 秒采样，保留 7 天物理清理）；
3. 重建 sys_audit_log 的 audit_action_type CHECK 白名单（0028 的 39 项 + 3 = 42 项：
   PREVIEW_CACHE_CLEAR / FORCE_OFFLINE / BACKUP_TEST_TARGET）；
4. 新增审计目标类型 SYSTEM，重建 audit_target_type CHECK（原 11 项 + 1 = 12 项）；
5. 达梦兼容：约束名大写探测小写比较后 drop（同 0028 做法）；
6. 升级幂等：加列先探测、建表先探测、CHECK drop-if-exists 后重建，可重复跑。
"""

from __future__ import annotations

from collections.abc import Sequence

import sqlalchemy as sa
from alembic import op

revision: str = "0029_system_management"
down_revision: str | None = "0028_backup_audit_actions"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None

# 0028 的 V9 全集 + 本期新增动作
AUDIT_ACTIONS_V10 = (
    "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",
    "BACKUP_CREATE",
    "PREVIEW_CACHE_CLEAR",
    "FORCE_OFFLINE",
    "BACKUP_TEST_TARGET",
)
AUDIT_ACTIONS_V9 = AUDIT_ACTIONS_V10[:-3]

# 现行 target_type 白名单（0028 起 11 项）+ 本期新增 SYSTEM
AUDIT_TARGETS_V3 = (
    "AUTH",
    "CATEGORY",
    "DOCUMENT",
    "ATTACHMENT",
    "PERMISSION",
    "ATTACHMENT_BINDING",
    "ORGANIZATION",
    "USER",
    "ROLE",
    "IP_ENTRY",
    "BACKUP",
    "SYSTEM",
)
AUDIT_TARGETS_V2 = AUDIT_TARGETS_V3[:-1]

ACTION_CONSTRAINT = "audit_action_type"
TARGET_CONSTRAINT = "audit_target_type"
ROLE_MANAGE_COLUMNS = ("manage_system", "manage_audit_log")


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:
    inline = ", ".join(f"'{v}'" for v in values)
    return f"(target_type IS NULL OR target_type IN ({inline}))"


def _existing_check_names() -> set[str]:
    # 达梦返回的约束名为大写，统一小写比较（同 0028）。
    return {
        item["name"].lower()
        for item in sa.inspect(op.get_bind()).get_check_constraints("sys_audit_log")
    }


def _drop_checks(names: set[str]) -> None:
    for raw_name in (ACTION_CONSTRAINT, TARGET_CONSTRAINT):
        if f"ck_sys_audit_log_{raw_name}" in names:
            op.drop_constraint(raw_name, "sys_audit_log", type_="check")


def _existing_role_columns() -> set[str]:
    inspector = sa.inspect(op.get_bind())
    return {
        column["name"].lower()
        for column in inspector.get_columns("sys_role")
    }


def _table_exists(table_name: str) -> bool:
    inspector = sa.inspect(op.get_bind())
    return table_name in {t.lower() for t in inspector.get_table_names()}


def _has_system_audit_records() -> bool:
    result = op.get_bind().execute(
        sa.text(
            "SELECT COUNT(*) FROM sys_audit_log WHERE action_type IN "
            "('PREVIEW_CACHE_CLEAR', 'FORCE_OFFLINE', 'BACKUP_TEST_TARGET')"
        )
    )
    return bool(result.scalar())


def upgrade() -> None:
    # 1. sys_role 管理位加列（先探测后执行，幂等）
    existing_columns = _existing_role_columns()
    for column in ROLE_MANAGE_COLUMNS:
        if column not in existing_columns:
            op.add_column(
                "sys_role",
                sa.Column(
                    column,
                    sa.Boolean(),
                    nullable=False,
                    server_default=sa.text("0"),
                ),
            )

    # 2. 资源采样表（先探测，幂等）
    if not _table_exists("sys_metric_sample"):
        op.create_table(
            "sys_metric_sample",
            sa.Column("id", sa.BigInteger(), primary_key=True, autoincrement=True),
            sa.Column(
                "sampled_at",
                sa.dialects.mysql.DATETIME(fsp=3),
                nullable=False,
                server_default=sa.text("CURRENT_TIMESTAMP(3)"),
            ),
            sa.Column("cpu_percent", sa.Float(), nullable=True),
            sa.Column("mem_used_bytes", sa.BigInteger(), nullable=False),
            sa.Column("mem_total_bytes", sa.BigInteger(), nullable=False),
        )
        op.create_index(
            "ix_sys_metric_sample_time",
            "sys_metric_sample",
            ["sampled_at"],
        )

    # 3. 审计白名单重建（drop-if-exists + 重建）
    existing = _existing_check_names()
    _drop_checks(existing)
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V10),
    )
    op.create_check_constraint(
        TARGET_CONSTRAINT,
        "sys_audit_log",
        _target_check_sql(AUDIT_TARGETS_V3),
    )


def downgrade() -> None:
    if _has_system_audit_records():
        raise RuntimeError("已存在系统管理审计记录，拒绝有损降级")

    op.drop_index("ix_sys_metric_sample_time", table_name="sys_metric_sample")
    op.drop_table("sys_metric_sample")

    existing_columns = _existing_role_columns()
    for column in ROLE_MANAGE_COLUMNS:
        if column in existing_columns:
            op.drop_column("sys_role", column)

    existing = _existing_check_names()
    _drop_checks(existing)
    op.create_check_constraint(
        ACTION_CONSTRAINT,
        "sys_audit_log",
        _check_sql(AUDIT_ACTIONS_V9),
    )
    op.create_check_constraint(
        TARGET_CONSTRAINT,
        "sys_audit_log",
        _target_check_sql(AUDIT_TARGETS_V2),
    )