document-management/backend/dms/services/role_service.py

"""角色 CRUD、权限矩阵与用户关联维护。"""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any

from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError

from dms.common.enums import (
    AuditAction,
    AuditTarget,
    DeptScope,
    EnabledStatus,
    RoleCode,
    SecurityLevel,
)
from dms.common.errors import (
    ConflictError,
    InvalidArgumentError,
    ResourceNotFoundError,
    RoleInUseError,
)
from dms.common.pagination import PageRequest, page_result
from dms.common.response import serialize_id
from dms.database.transaction import transaction
from dms.extensions import db
from dms.models import Role, RolePermission, User
from dms.security.auth_context import get_auth_context
from dms.services.audit_service import business_audit


ROLE_CREATE_FIELDS = {"roleName", "sortNo", "status", "remark", "isDefault"}
ROLE_UPDATE_FIELDS = {
    "roleName",
    "sortNo",
    "status",
    "remark",
    "isDefault",
    "rowVersion",
}
ROLE_PERMISSION_FIELDS = {"matrix", "rowVersion"}
ROLE_USER_FIELDS = {"userIds", "rowVersion"}

# 矩阵 7 个操作布尔位与 AllowedAction 一一对应
PERMISSION_FIELDS: tuple[tuple[str, str], ...] = (
    ("canView", "can_view"),
    ("canDownload", "can_download"),
    ("canEdit", "can_edit"),
    ("canDelete", "can_delete"),
    ("canConfigPermission", "can_config_permission"),
    ("canBindAttachment", "can_bind_attachment"),
    ("canUnbindAttachment", "can_unbind_attachment"),
)
MATRIX_KEYS = {
    (level.value, scope.value)
    for level in SecurityLevel
    for scope in DeptScope
}


def _utc_now() -> datetime:
    return datetime.now(timezone.utc).replace(tzinfo=None)


def _required_text(
    value: Any,
    *,
    field: str,
    max_length: int,
) -> str:
    if not isinstance(value, str) or not value.strip():
        raise InvalidArgumentError(f"{field}不能为空")
    normalized = value.strip()
    if len(normalized) > max_length:
        raise InvalidArgumentError(f"{field}长度不能超过{max_length}")
    return normalized


def _optional_text(value: Any, *, field: str, max_length: int) -> str | None:
    if value is None:
        return None
    if not isinstance(value, str):
        raise InvalidArgumentError(f"{field}必须是字符串")
    normalized = value.strip()
    if len(normalized) > max_length:
        raise InvalidArgumentError(f"{field}长度不能超过{max_length}")
    return normalized or None


def _sort_no(value: Any) -> int:
    if type(value) is not int or value < 0 or value > 999999:
        raise InvalidArgumentError("sortNo必须是0~999999的整数")
    return value


def _status_value(value: Any) -> str:
    if not isinstance(value, str):
        raise InvalidArgumentError("status必须是字符串")
    try:
        return EnabledStatus(value).value
    except ValueError as exc:
        raise InvalidArgumentError("status必须是ENABLED或DISABLED") from exc


def _boolean(value: Any, *, field: str) -> bool:
    if type(value) is not bool:
        raise InvalidArgumentError(f"{field}必须是布尔值")
    return value


def _row_version(value: Any) -> int:
    if type(value) is not int or value < 0:
        raise InvalidArgumentError("rowVersion必须是非负整数")
    return value


def _id_list(value: Any, *, field: str) -> list[int]:
    if not isinstance(value, list) or any(
        not isinstance(item, str) or not item.isdecimal() or int(item) <= 0
        for item in value
    ):
        raise InvalidArgumentError(f"{field}必须是正整数形式的字符串ID数组")
    ids = [int(item) for item in value]
    if len(set(ids)) != len(ids):
        raise InvalidArgumentError(f"{field}不能包含重复ID")
    return ids


def _exact_payload(payload: Any, expected: set[str]) -> dict[str, Any]:
    if not isinstance(payload, dict) or set(payload) != expected:
        raise InvalidArgumentError(
            "请求体字段必须且只能包含：" + "、".join(sorted(expected))
        )
    return payload


def _role_row(session, role_id: int, *, for_update: bool = False) -> Role:
    stmt = select(Role).where(Role.id == role_id, Role.is_deleted.is_(False))
    if for_update:
        stmt = stmt.with_for_update()
    role = session.scalar(stmt)
    if role is None:
        raise ResourceNotFoundError("角色不存在")
    return role


def _member_counts(session, role_ids: list[int]) -> dict[int, int]:
    if not role_ids:
        return {}
    rows = session.execute(
        select(User.role_id, func.count(User.id))
        .where(User.role_id.in_(role_ids), User.is_deleted.is_(False))
        .group_by(User.role_id)
    ).all()
    return {role_id: count for role_id, count in rows}


def role_summary(role: Role, member_count: int) -> dict[str, object]:
    return {
        "id": serialize_id(role.id),
        "roleName": role.role_name,
        "sortNo": role.sort_no,
        "status": role.status,
        "remark": role.remark,
        "isDefault": role.is_default,
        "memberCount": member_count,
        "rowVersion": role.row_version,
    }


def list_roles(
    *,
    keyword: str | None,
    status: str | None,
    page_request: PageRequest,
) -> dict[str, object]:
    filters = [Role.is_deleted.is_(False)]
    if keyword:
        filters.append(Role.role_name.like(f"%{keyword.strip()}%"))
    if status:
        try:
            normalized_status = EnabledStatus(status).value
        except ValueError as exc:
            raise InvalidArgumentError("status必须是ENABLED或DISABLED") from exc
        filters.append(Role.status == normalized_status)

    total = db.session.scalar(select(func.count(Role.id)).where(*filters)) or 0
    roles = db.session.scalars(
        select(Role)
        .where(*filters)
        .order_by(Role.sort_no.asc(), Role.id.asc())
        .offset(page_request.offset)
        .limit(page_request.page_size)
    ).all()
    counts = _member_counts(db.session, [role.id for role in roles])
    return page_result(
        [role_summary(role, counts.get(role.id, 0)) for role in roles],
        page=page_request.page,
        page_size=page_request.page_size,
        total=total,
    )


def _clear_other_defaults(session, *, keep_id: int | None) -> None:
    """将除 keep_id 外的活跃默认角色取消默认标记（先清后设避免唯一约束冲突）。"""
    stmt = select(Role).where(
        Role.is_default.is_(True),
        Role.is_deleted.is_(False),
    )
    if keep_id is not None:
        stmt = stmt.where(Role.id != keep_id)
    actor_id = get_auth_context().user_id
    now = _utc_now()
    for existing in session.scalars(stmt).all():
        existing.is_default = False
        existing.updated_by = actor_id
        existing.updated_at = now
        existing.row_version += 1


def _ensure_role_name_free(session, role_name: str, *, exclude_id: int) -> None:
    duplicated = session.scalar(
        select(Role.id).where(
            Role.role_name == role_name,
            Role.is_deleted.is_(False),
            Role.id != exclude_id,
        )
    )
    if duplicated is not None:
        raise ConflictError("角色名已存在")


def _seed_permission_rows(session, role_id: int, actor_id: int | None) -> None:
    for level in SecurityLevel:
        for scope in DeptScope:
            session.add(
                RolePermission(
                    role_id=role_id,
                    security_level=level.value,
                    dept_scope=scope.value,
                    created_by=actor_id,
                    updated_by=actor_id,
                )
            )


def create_role(payload: Any) -> dict[str, object]:
    body = _exact_payload(payload, ROLE_CREATE_FIELDS)
    role_name = _required_text(body["roleName"], field="roleName", max_length=64)
    sort_no = _sort_no(body["sortNo"])
    status = _status_value(body["status"])
    remark = _optional_text(body["remark"], field="remark", max_length=255)
    is_default = _boolean(body["isDefault"], field="isDefault")
    actor_id = get_auth_context().user_id

    try:
        with transaction() as session:
            _ensure_role_name_free(session, role_name, exclude_id=0)
            if is_default:
                _clear_other_defaults(session, keep_id=None)
                session.flush()
            role = Role(
                role_name=role_name,
                sort_no=sort_no,
                status=status,
                remark=remark,
                is_default=is_default,
                created_by=actor_id,
                updated_by=actor_id,
            )
            session.add(role)
            session.flush()
            _seed_permission_rows(session, role.id, actor_id)
            session.add(
                business_audit(
                    action=AuditAction.CREATE_ROLE,
                    target=AuditTarget.ROLE,
                    target_id=role.id,
                    target_name=role.role_name,
                    detail={
                        "roleName": role.role_name,
                        "sortNo": role.sort_no,
                        "status": role.status,
                        "remark": role.remark,
                        "isDefault": role.is_default,
                    },
                )
            )
        return role_summary(role, 0)
    except IntegrityError as exc:
        raise ConflictError("角色名已存在或默认角色设置冲突") from exc


def update_role(role_id: int, payload: Any) -> dict[str, object]:
    body = _exact_payload(payload, ROLE_UPDATE_FIELDS)
    role_name = _required_text(body["roleName"], field="roleName", max_length=64)
    sort_no = _sort_no(body["sortNo"])
    status = _status_value(body["status"])
    remark = _optional_text(body["remark"], field="remark", max_length=255)
    is_default = _boolean(body["isDefault"], field="isDefault")
    expected_version = _row_version(body["rowVersion"])
    actor_id = get_auth_context().user_id
    now = _utc_now()

    with transaction() as session:
        role = _role_row(session, role_id, for_update=True)
        if role.row_version != expected_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": role.row_version},
            )
        _ensure_role_name_free(session, role_name, exclude_id=role.id)

        before = {
            "roleName": role.role_name,
            "sortNo": role.sort_no,
            "status": role.status,
            "remark": role.remark,
            "isDefault": role.is_default,
        }
        # 新设默认：先清旧默认，避免 default_marker 唯一约束冲突
        if is_default and not role.is_default:
            _clear_other_defaults(session, keep_id=role.id)
            session.flush()

        role.role_name = role_name
        role.sort_no = sort_no
        role.status = status
        role.remark = remark
        role.is_default = is_default
        role.updated_by = actor_id
        role.updated_at = now
        role.row_version += 1
        after = {
            "roleName": role.role_name,
            "sortNo": role.sort_no,
            "status": role.status,
            "remark": role.remark,
            "isDefault": role.is_default,
        }
        session.add(
            business_audit(
                action=AuditAction.EDIT_ROLE,
                target=AuditTarget.ROLE,
                target_id=role.id,
                target_name=role.role_name,
                detail={"before": before, "after": after},
            )
        )
    member_count = _member_counts(db.session, [role.id]).get(role.id, 0)
    return role_summary(role, member_count)


def delete_role(role_id: int, row_version: int) -> dict[str, object]:
    actor_id = get_auth_context().user_id
    now = _utc_now()

    with transaction() as session:
        role = _role_row(session, role_id, for_update=True)
        if role.row_version != row_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": role.row_version},
            )
        bound_users = (
            session.scalar(
                select(func.count(User.id)).where(
                    User.role_id == role_id,
                    User.is_deleted.is_(False),
                )
            )
            or 0
        )
        if bound_users:
            raise RoleInUseError(
                details={"memberCount": bound_users},
            )

        target_name = role.role_name
        role.is_deleted = True
        role.deleted_at = now
        role.updated_by = actor_id
        role.updated_at = now
        role.row_version += 1
        # 矩阵行随角色软删，保留审计追溯
        for row in session.scalars(
            select(RolePermission).where(
                RolePermission.role_id == role_id,
                RolePermission.is_deleted.is_(False),
            )
        ).all():
            row.is_deleted = True
            row.deleted_at = now
            row.updated_by = actor_id
            row.updated_at = now
        session.add(
            business_audit(
                action=AuditAction.DELETE_ROLE,
                target=AuditTarget.ROLE,
                target_id=role_id,
                target_name=target_name,
                detail={"roleName": target_name, "rowVersion": role.row_version},
            )
        )
    return {"id": serialize_id(role_id), "deleted": True}


def _permission_row_dto(row: RolePermission) -> dict[str, object]:
    dto: dict[str, object] = {
        "securityLevel": row.security_level,
        "deptScope": row.dept_scope,
    }
    for name, attr in PERMISSION_FIELDS:
        dto[name] = getattr(row, attr)
    return dto


def list_role_permissions(role_id: int) -> list[dict[str, object]]:
    _role_row(db.session, role_id)
    rows = db.session.scalars(
        select(RolePermission)
        .where(
            RolePermission.role_id == role_id,
            RolePermission.is_deleted.is_(False),
        )
        .order_by(RolePermission.id.asc())
    ).all()
    return [_permission_row_dto(row) for row in rows]


def _validate_matrix(value: Any) -> list[dict[str, Any]]:
    expected_row_fields = {"securityLevel", "deptScope"} | {
        name for name, _ in PERMISSION_FIELDS
    }
    if not isinstance(value, list) or len(value) != len(MATRIX_KEYS):
        raise InvalidArgumentError(f"权限矩阵必须为{len(MATRIX_KEYS)}行")
    rows: dict[tuple[str, str], dict[str, Any]] = {}
    for item in value:
        if not isinstance(item, dict) or set(item) != expected_row_fields:
            raise InvalidArgumentError(
                "矩阵行字段必须且只能包含：" + "、".join(sorted(expected_row_fields))
            )
        key = (item["securityLevel"], item["deptScope"])
        if key not in MATRIX_KEYS:
            raise InvalidArgumentError("矩阵行的密级或部门范围不是有效枚举值")
        if key in rows:
            raise InvalidArgumentError("矩阵行不能重复")
        for name, _ in PERMISSION_FIELDS:
            if type(item[name]) is not bool:
                raise InvalidArgumentError(f"{name}必须是布尔值")
        rows[key] = item
    return list(value)


def update_role_permissions(role_id: int, payload: Any) -> dict[str, object]:
    body = _exact_payload(payload, ROLE_PERMISSION_FIELDS)
    matrix = _validate_matrix(body["matrix"])
    expected_version = _row_version(body["rowVersion"])
    actor_id = get_auth_context().user_id
    now = _utc_now()

    with transaction() as session:
        role = _role_row(session, role_id, for_update=True)
        if role.row_version != expected_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": role.row_version},
            )
        existing_rows = session.scalars(
            select(RolePermission).where(
                RolePermission.role_id == role_id,
                RolePermission.is_deleted.is_(False),
            )
        ).all()
        by_key = {(row.security_level, row.dept_scope): row for row in existing_rows}
        if set(by_key) != MATRIX_KEYS:
            raise ConflictError("角色权限矩阵数据不完整，请联系管理员核查")

        changed_keys: list[dict[str, object]] = []
        for item in matrix:
            row = by_key[(item["securityLevel"], item["deptScope"])]
            before_flags = {
                name: getattr(row, attr) for name, attr in PERMISSION_FIELDS
            }
            after_flags = {name: item[name] for name, _ in PERMISSION_FIELDS}
            if before_flags != after_flags:
                row.updated_by = actor_id
                row.updated_at = now
                for name, attr in PERMISSION_FIELDS:
                    setattr(row, attr, item[name])
                changed_keys.append(
                    {
                        "securityLevel": row.security_level,
                        "deptScope": row.dept_scope,
                        "before": before_flags,
                        "after": after_flags,
                    }
                )

        role.updated_by = actor_id
        role.updated_at = now
        role.row_version += 1
        session.add(
            business_audit(
                action=AuditAction.CHANGE_ROLE_PERMISSION,
                target=AuditTarget.ROLE,
                target_id=role_id,
                target_name=role.role_name,
                detail={"scope": "matrix", "changedRows": changed_keys},
            )
        )
    return {
        "id": serialize_id(role_id),
        "matrix": list_role_permissions(role_id),
        "rowVersion": role.row_version,
    }


def list_role_users(role_id: int) -> dict[str, object]:
    _role_row(db.session, role_id)
    user_ids = db.session.scalars(
        select(User.id)
        .where(User.role_id == role_id, User.is_deleted.is_(False))
        .order_by(User.id.asc())
    ).all()
    return {"id": serialize_id(role_id), "userIds": [serialize_id(v) for v in user_ids]}


def update_role_users(role_id: int, payload: Any) -> dict[str, object]:
    body = _exact_payload(payload, ROLE_USER_FIELDS)
    user_ids = _id_list(body["userIds"], field="userIds")
    expected_version = _row_version(body["rowVersion"])
    actor_id = get_auth_context().user_id
    now = _utc_now()

    with transaction() as session:
        role = _role_row(session, role_id, for_update=True)
        if role.row_version != expected_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": role.row_version},
            )
        target_ids = set(user_ids)
        found: dict[int, User] = {}
        if target_ids:
            for user in session.scalars(
                select(User)
                .where(User.id.in_(target_ids), User.is_deleted.is_(False))
                .with_for_update()
            ).all():
                found[user.id] = user
            if set(found) != target_ids:
                raise InvalidArgumentError("userIds包含无效用户")
            for user in found.values():
                if user.role_code != RoleCode.USER.value:
                    raise InvalidArgumentError("仅普通用户可关联自定义角色")

        current_users = session.scalars(
            select(User)
            .where(User.role_id == role_id, User.is_deleted.is_(False))
            .with_for_update()
        ).all()
        added_ids: list[int] = []
        removed_ids: list[int] = []
        for user in current_users:
            if user.id not in target_ids:
                user.role_id = None
                user.auth_version += 1
                user.updated_by = actor_id
                user.updated_at = now
                user.row_version += 1
                removed_ids.append(user.id)
        for user_id in sorted(target_ids):
            user = found[user_id]
            if user.role_id != role_id:
                user.role_id = role_id
                user.auth_version += 1
                user.updated_by = actor_id
                user.updated_at = now
                user.row_version += 1
                added_ids.append(user_id)

        role.updated_by = actor_id
        role.updated_at = now
        role.row_version += 1
        session.add(
            business_audit(
                action=AuditAction.CHANGE_ROLE_USER,
                target=AuditTarget.ROLE,
                target_id=role_id,
                target_name=role.role_name,
                detail={
                    "addedUserIds": [serialize_id(v) for v in added_ids],
                    "removedUserIds": [serialize_id(v) for v in removed_ids],
                },
            )
        )
    return {
        "id": serialize_id(role_id),
        "userIds": [serialize_id(v) for v in sorted(target_ids)],
        "rowVersion": role.row_version,
    }