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

"""B6主案ACL查询、动态继承及权限全集差异保存。"""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

from dms.common.enums import (
    AllowedAction,
    AuditAction,
    AuditTarget,
    DocumentType,
    EnabledStatus,
    SubjectType,
    VisibilityType,
)
from dms.common.errors import (
    AttachmentHasNoAclError,
    ConflictError,
    DuplicatePermissionSubjectError,
    InvalidArgumentError,
    ResourceNotFoundError,
    SubPlanPermissionInheritedError,
)
from dms.common.response import serialize_id
from dms.database.transaction import execute_with_deadlock_retry, transaction
from dms.extensions import db
from dms.models import Document, Organization, Permission, User
from dms.security.auth_context import get_auth_context
from dms.services.audit_service import business_audit
from dms.services.document_password_service import (
    DocumentPasswordAction,
    get_request_password,
    require_document_password,
)
from dms.services.authorization_service import (
    require_plan_action,
    require_security_clearance,
)

PERMISSION_FIELDS = {
    "subjectType",
    "subjectId",
    "canView",
    "canDownload",
    "canEdit",
    "canManagePermission",
    "canDelete",
}
SAVE_FIELDS = {"visibilityType", "documentRowVersion", "entries"}
ACTION_FIELDS = (
    "canView",
    "canDownload",
    "canEdit",
    "canManagePermission",
    "canDelete",
)


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


def _string_id(value: Any, name: str) -> int:
    if not isinstance(value, str) or not value.isdecimal() or int(value) <= 0:
        raise InvalidArgumentError(f"{name}必须是正整数形式的字符串ID")
    return int(value)


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


def _document(document_id: int, *, lock: bool = False) -> Document:
    statement = select(Document).where(
        Document.id == document_id,
        Document.is_deleted.is_(False),
    )
    if lock:
        statement = statement.with_for_update()
    document = db.session.scalar(statement)
    if document is None:
        raise ResourceNotFoundError("文档不存在")
    return document


def _permission_source(document: Document, *, lock: bool = False) -> Document:
    if document.document_type == DocumentType.ATTACHMENT.value:
        raise AttachmentHasNoAclError()
    require_security_clearance(document, get_auth_context())
    if document.document_type == DocumentType.MAIN.value:
        # 读取与写入同口径：仅授予授权位（或管理员）可见授权清单
        require_plan_action(
            document, get_auth_context(), AllowedAction.CONFIG_PERMISSION
        )
        return document
    statement = select(Document).where(
        Document.id == document.root_document_id,
        Document.document_type == DocumentType.MAIN.value,
        Document.is_deleted.is_(False),
    )
    if lock:
        statement = statement.with_for_update()
    source = db.session.scalar(statement)
    if source is None:
        raise ResourceNotFoundError("子方案所属主案不存在")
    require_document_password(
        source,
        DocumentPasswordAction.CONFIG_PERMISSION.value,
        get_request_password(),
        get_auth_context(),
    )
    return source


def _entry(permission: Permission) -> dict[str, object]:
    return {
        "id": serialize_id(permission.id),
        "subjectType": permission.subject_type,
        "subjectId": serialize_id(permission.subject_id),
        "subjectName": permission.subject_name,
        "canView": bool(permission.can_view),
        "canDownload": bool(permission.can_download),
        "canEdit": bool(permission.can_edit),
        "canManagePermission": bool(permission.can_manage_permission),
        "canDelete": bool(permission.can_delete),
    }


def _active_permissions(source_id: int, *, lock: bool = False) -> list[Permission]:
    statement = (
        select(Permission)
        .where(
            Permission.document_id == source_id,
            Permission.is_deleted.is_(False),
        )
        .order_by(
            Permission.subject_type.asc(),
            Permission.subject_id.asc(),
            Permission.id.asc(),
        )
    )
    if lock:
        statement = statement.with_for_update()
    return list(db.session.scalars(statement).all())


def _detail(
    document: Document,
    source: Document,
    permissions: list[Permission],
) -> dict[str, object]:
    return {
        "documentId": serialize_id(document.id),
        "sourceDocumentId": serialize_id(source.id),
        "inherited": document.document_type == DocumentType.SUB_PLAN.value,
        "visibilityType": source.visibility_type,
        "documentRowVersion": source.row_version,
        "entries": [_entry(permission) for permission in permissions],
    }


def get_permissions(document_id: int) -> dict[str, object]:
    document = _document(document_id)
    source = _permission_source(document)
    return _detail(document, source, _active_permissions(source.id))


def _payload(payload: Any) -> tuple[str, int, list[dict[str, Any]]]:
    if not isinstance(payload, dict) or set(payload) != SAVE_FIELDS:
        raise InvalidArgumentError(
            "请求字段必须且只能包含visibilityType、documentRowVersion和entries"
        )
    try:
        visibility = VisibilityType(payload["visibilityType"]).value
    except (TypeError, ValueError) as exc:
        raise InvalidArgumentError("visibilityType不是有效枚举值") from exc
    expected_version = _version(payload["documentRowVersion"])
    raw_entries = payload["entries"]
    if not isinstance(raw_entries, list):
        raise InvalidArgumentError("entries必须是数组")

    entries: list[dict[str, Any]] = []
    seen: set[tuple[str, int]] = set()
    for index, raw_entry in enumerate(raw_entries):
        if not isinstance(raw_entry, dict) or set(raw_entry) != PERMISSION_FIELDS:
            raise InvalidArgumentError(
                f"entries[{index}]字段必须完整且不能包含未知字段"
            )
        try:
            subject_type = SubjectType(raw_entry["subjectType"]).value
        except (TypeError, ValueError) as exc:
            raise InvalidArgumentError(
                f"entries[{index}].subjectType不是有效枚举值"
            ) from exc
        subject_id = _string_id(
            raw_entry["subjectId"], f"entries[{index}].subjectId"
        )
        for field in ACTION_FIELDS:
            if type(raw_entry[field]) is not bool:
                raise InvalidArgumentError(
                    f"entries[{index}].{field}必须是boolean"
                )
        key = (subject_type, subject_id)
        if key in seen:
            raise DuplicatePermissionSubjectError(
                details={
                    "subjectType": subject_type,
                    "subjectId": serialize_id(subject_id),
                }
            )
        seen.add(key)
        entries.append(
            {
                "subjectType": subject_type,
                "subjectId": subject_id,
                **{field: raw_entry[field] for field in ACTION_FIELDS},
            }
        )

    if visibility == VisibilityType.ALL_AUTHENTICATED.value and entries:
        raise InvalidArgumentError("ALL_AUTHENTICATED的entries必须为空")
    if visibility == VisibilityType.ORGANIZATION.value:
        if not entries:
            raise InvalidArgumentError("ORGANIZATION至少需要一个ORG权限主体")
        if any(
            entry["subjectType"] != SubjectType.ORG.value
            for entry in entries
        ):
            raise InvalidArgumentError("ORGANIZATION只允许ORG权限主体")
    return visibility, expected_version, entries


def _locked_subject_names(
    entries: list[dict[str, Any]],
) -> dict[tuple[str, int], str]:
    organization_ids = sorted(
        entry["subjectId"]
        for entry in entries
        if entry["subjectType"] == SubjectType.ORG.value
    )
    user_ids = sorted(
        entry["subjectId"]
        for entry in entries
        if entry["subjectType"] == SubjectType.USER.value
    )
    names: dict[tuple[str, int], str] = {}
    if organization_ids:
        organizations = db.session.scalars(
            select(Organization)
            .where(Organization.id.in_(organization_ids))
            .order_by(Organization.id.asc())
            .with_for_update()
        ).all()
        valid = {
            item.id: item
            for item in organizations
            if not item.is_deleted and item.status == EnabledStatus.ENABLED.value
        }
        if set(valid) != set(organization_ids):
            raise ResourceNotFoundError("权限组织不存在、已删除或已禁用")
        names.update(
            {
                (SubjectType.ORG.value, item.id): item.org_name
                for item in valid.values()
            }
        )
    if user_ids:
        users = db.session.scalars(
            select(User)
            .where(User.id.in_(user_ids))
            .order_by(User.id.asc())
            .with_for_update()
        ).all()
        valid = {
            item.id: item
            for item in users
            if not item.is_deleted and item.status == EnabledStatus.ENABLED.value
        }
        if set(valid) != set(user_ids):
            raise ResourceNotFoundError("权限用户不存在、已删除或已禁用")
        names.update(
            {
                (SubjectType.USER.value, item.id): item.real_name
                for item in valid.values()
            }
        )
    return names


def _current_version(document_id: int) -> int | None:
    return db.session.scalar(
        select(Document.row_version).where(
            Document.id == document_id,
            Document.is_deleted.is_(False),
        )
    )


def save_permissions(document_id: int, payload: Any) -> dict[str, object]:
    visibility, expected_version, entries = _payload(payload)
    context = get_auth_context()

    def operation() -> dict[str, object]:
        with transaction() as session:
            document = _document(document_id, lock=True)
            if document.document_type == DocumentType.ATTACHMENT.value:
                raise AttachmentHasNoAclError()
            require_security_clearance(document, context)
            if document.document_type == DocumentType.SUB_PLAN.value:
                raise SubPlanPermissionInheritedError()
            require_plan_action(
                document, context, AllowedAction.CONFIG_PERMISSION
            )
            # 设密主案的授权操作仍受密码锁保护（与前端弹窗口径一致）
            require_document_password(
                document,
                DocumentPasswordAction.CONFIG_PERMISSION.value,
                get_request_password(),
                context,
            )
            if document.document_type != DocumentType.MAIN.value:
                raise ResourceNotFoundError("主案不存在或类型不匹配")
            if document.row_version != expected_version:
                raise ConflictError(
                    "数据已被其他用户修改，请刷新后重试",
                    details={"currentRowVersion": document.row_version},
                )

            current = _active_permissions(document.id, lock=True)
            subject_names = _locked_subject_names(entries)
            current_by_key = {
                (permission.subject_type, permission.subject_id): permission
                for permission in current
            }
            requested_by_key = {
                (entry["subjectType"], entry["subjectId"]): entry
                for entry in entries
            }
            added: list[dict[str, str]] = []
            updated: list[dict[str, str]] = []
            deleted: list[dict[str, str]] = []
            now = _now()

            for key, permission in current_by_key.items():
                requested = requested_by_key.get(key)
                if requested is None:
                    permission.is_deleted = True
                    permission.deleted_at = now
                    permission.updated_at = now
                    permission.updated_by = context.user_id
                    permission.row_version += 1
                    deleted.append(
                        {
                            "subjectType": key[0],
                            "subjectId": serialize_id(key[1]),
                        }
                    )
                    continue
                new_name = subject_names[key]
                changed = permission.subject_name != new_name or any(
                    bool(getattr(permission, _snake_action(field)))
                    != requested[field]
                    for field in ACTION_FIELDS
                )
                if changed:
                    permission.subject_name = new_name
                    for field in ACTION_FIELDS:
                        setattr(
                            permission,
                            _snake_action(field),
                            requested[field],
                        )
                    permission.updated_at = now
                    permission.updated_by = context.user_id
                    permission.row_version += 1
                    updated.append(
                        {
                            "subjectType": key[0],
                            "subjectId": serialize_id(key[1]),
                        }
                    )

            for key, requested in requested_by_key.items():
                if key in current_by_key:
                    continue
                permission = Permission(
                    document_id=document.id,
                    subject_type=key[0],
                    subject_id=key[1],
                    subject_name=subject_names[key],
                    created_by=context.user_id,
                    updated_by=context.user_id,
                    **{
                        _snake_action(field): requested[field]
                        for field in ACTION_FIELDS
                    },
                )
                session.add(permission)
                added.append(
                    {
                        "subjectType": key[0],
                        "subjectId": serialize_id(key[1]),
                    }
                )

            previous_visibility = document.visibility_type
            document.visibility_type = visibility
            document.updated_by = context.user_id
            document.updated_by_name = context.real_name
            document.updated_at = now
            document.row_version += 1
            session.flush()
            session.add(
                business_audit(
                    action=AuditAction.CHANGE_PERMISSION,
                    target=AuditTarget.PERMISSION,
                    target_id=document.id,
                    target_name=document.document_name,
                    detail={
                        "mainPlanId": serialize_id(document.id),
                        "visibilityTypeBefore": previous_visibility,
                        "visibilityTypeAfter": visibility,
                        "added": added,
                        "updated": updated,
                        "deleted": deleted,
                        "documentRowVersion": document.row_version,
                    },
                )
            )
            return _detail(
                document,
                document,
                _active_permissions(document.id),
            )

    try:
        return execute_with_deadlock_retry(operation)
    except IntegrityError as exc:
        db.session.rollback()
        raise ConflictError(
            "权限集合已被并发修改，请刷新后重试",
            details={"currentRowVersion": _current_version(document_id)},
        ) from exc


def _snake_action(field: str) -> str:
    return {
        "canView": "can_view",
        "canDownload": "can_download",
        "canEdit": "can_edit",
        "canManagePermission": "can_manage_permission",
        "canDelete": "can_delete",
    }[field]


__all__ = ["get_permissions", "save_permissions"]