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

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

借阅语义（0026迁移，契约1.19）：授权仅面向用户（subject_id），限时生效
（expires_at，最长7天），操作位收敛为查看/下载两项；每条授权可选挂一份
PDF审批单凭证（voucher_*，凭证读写见 voucher_service）。
"""

from __future__ import annotations

from datetime import datetime, timedelta, timezone
from typing import Any

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

from dms.common.enums import (
    AllowedAction,
    AuditAction,
    AuditTarget,
    DocumentType,
    EnabledStatus,
)
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 LOCK_DEFER_LOB, Document, 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 = {"subjectId", "canView", "canDownload", "expiresAt"}
SAVE_FIELDS = {"documentRowVersion", "entries"}
ACTION_FIELDS = ("canView", "canDownload")

# 借阅时限：最长7天；60秒容差覆盖前端分钟粒度的拾取舍入
MAX_BORROW = timedelta(days=7)
_BORROW_TOLERANCE = timedelta(seconds=60)


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


def _iso(value: datetime) -> str:
    return (
        value.replace(tzinfo=timezone.utc)
        .isoformat(timespec="milliseconds")
        .replace("+00:00", "Z")
    )


def _parse_expires(value: Any, index: int, *, now: datetime) -> datetime:
    """解析UTC时间字符串（Z结尾）并执行借阅时限校验。"""
    if not isinstance(value, str) or not value.endswith("Z"):
        raise InvalidArgumentError(
            f"entries[{index}].expiresAt必须是UTC时间字符串（Z结尾）"
        )
    try:
        parsed = datetime.fromisoformat(f"{value[:-1]}+00:00")
    except ValueError as exc:
        raise InvalidArgumentError(
            f"entries[{index}].expiresAt不是有效的时间格式"
        ) from exc
    if parsed.utcoffset() != timezone.utc.utcoffset(parsed):
        raise InvalidArgumentError(
            f"entries[{index}].expiresAt必须是UTC时间"
        )
    expires = parsed.astimezone(timezone.utc).replace(tzinfo=None)
    if expires <= now:
        raise InvalidArgumentError(
            f"entries[{index}].expiresAt必须晚于当前时间"
        )
    if expires > now + MAX_BORROW + _BORROW_TOLERANCE:
        raise InvalidArgumentError(
            f"entries[{index}].expiresAt借阅时限最长为7天"
        )
    return expires


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.options(*LOCK_DEFER_LOB).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.options(*LOCK_DEFER_LOB).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),
        "subjectId": serialize_id(permission.subject_id),
        "subjectName": permission.subject_name,
        "canView": bool(permission.can_view),
        "canDownload": bool(permission.can_download),
        "expiresAt": _iso(permission.expires_at),
        "voucherFileName": permission.voucher_file_name,
        "voucherFileSize": permission.voucher_file_size,
        "voucherUploadedAt": (
            _iso(permission.voucher_uploaded_at)
            if permission.voucher_uploaded_at is not None
            else None
        ),
    }


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_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,
        "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[int, list[dict[str, Any]]]:
    if not isinstance(payload, dict) or set(payload) != SAVE_FIELDS:
        raise InvalidArgumentError(
            "请求字段必须且只能包含documentRowVersion和entries"
        )
    expected_version = _version(payload["documentRowVersion"])
    raw_entries = payload["entries"]
    if not isinstance(raw_entries, list):
        raise InvalidArgumentError("entries必须是数组")

    now = _now()
    entries: list[dict[str, Any]] = []
    seen: set[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}]字段必须完整且不能包含未知字段"
            )
        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"
                )
        expires_at = _parse_expires(
            raw_entry["expiresAt"], index, now=now
        )
        if subject_id in seen:
            raise DuplicatePermissionSubjectError(
                details={"subjectId": serialize_id(subject_id)}
            )
        seen.add(subject_id)
        entries.append(
            {
                "subjectId": subject_id,
                "canView": raw_entry["canView"],
                "canDownload": raw_entry["canDownload"],
                "expiresAt": expires_at,
            }
        )
    return expected_version, entries


def _locked_subject_names(
    entries: list[dict[str, Any]],
) -> dict[int, str]:
    user_ids = sorted(entry["subjectId"] for entry in entries)
    names: dict[int, str] = {}
    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({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 _audit_item(
    subject_id: int,
    subject_name: str,
    entry: dict[str, Any] | None,
) -> dict[str, object]:
    item: dict[str, object] = {
        "subjectId": serialize_id(subject_id),
        "subjectName": subject_name,
    }
    if entry is not None:
        item.update(
            {
                "canView": entry["canView"],
                "canDownload": entry["canDownload"],
                "expiresAt": _iso(entry["expiresAt"]),
            }
        )
    return item


def save_permissions(document_id: int, payload: Any) -> dict[str, object]:
    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_id: permission for permission in current
            }
            requested_by_key = {entry["subjectId"]: entry for entry in entries}
            added: list[dict[str, object]] = []
            updated: list[dict[str, object]] = []
            deleted: list[dict[str, object]] = []
            now = _now()

            for subject_id, permission in current_by_key.items():
                requested = requested_by_key.get(subject_id)
                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(
                        _audit_item(
                            subject_id, permission.subject_name, None
                        )
                    )
                    continue
                new_name = subject_names[subject_id]
                changed = (
                    permission.subject_name != new_name
                    or bool(permission.can_view) != requested["canView"]
                    or bool(permission.can_download) != requested["canDownload"]
                    or permission.expires_at != requested["expiresAt"]
                )
                if changed:
                    permission.subject_name = new_name
                    permission.can_view = requested["canView"]
                    permission.can_download = requested["canDownload"]
                    permission.expires_at = requested["expiresAt"]
                    permission.updated_at = now
                    permission.updated_by = context.user_id
                    permission.row_version += 1
                    updated.append(
                        _audit_item(subject_id, new_name, requested)
                    )

            for subject_id, requested in requested_by_key.items():
                if subject_id in current_by_key:
                    continue
                permission = Permission(
                    document_id=document.id,
                    subject_id=subject_id,
                    subject_name=subject_names[subject_id],
                    can_view=requested["canView"],
                    can_download=requested["canDownload"],
                    expires_at=requested["expiresAt"],
                    created_by=context.user_id,
                    updated_by=context.user_id,
                )
                session.add(permission)
                added.append(
                    _audit_item(
                        subject_id, subject_names[subject_id], requested
                    )
                )

            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),
                        "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


__all__ = ["get_permissions", "save_permissions"]