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

"""B8回收站查询和单条文档恢复。"""

from __future__ import annotations

import hashlib
import logging
import os
import stat
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, BinaryIO, Mapping

from flask import current_app
from sqlalchemy import and_, func, or_, select
from sqlalchemy.exc import IntegrityError

from dms.common.enums import (
    AuditAction,
    AuditResult,
    AuditTarget,
    DocumentType,
    EnabledStatus,
    SecurityLevel,
)
from dms.common.errors import (
    ConflictError,
    DocumentNotDeletedError,
    FileIntegrityMismatchError,
    FileNotFoundError as DmsFileNotFoundError,
    FilePathInvalidError,
    InvalidArgumentError,
    ResourceNotFoundError,
    RestoreCategoryInvalidError,
    RestoreFileChangedError,
    RestoreParentInvalidError,
    RestoreRelationConflictError,
    UnsupportedFileTypeError,
)
from dms.common.pagination import PageRequest, page_result
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,
    AttachmentBinding,
    AuditLog,
    Category,
    Document,
    Permission,
    User,
)
from dms.security.auth_context import get_auth_context
from dms.services.attachment_query_service import attachment_detail
from dms.services.audit_service import business_audit
from dms.services.authorization_service import evaluate_plan_access
from dms.services.document_query_service import _iso, document_detail
from dms.storage.paths import UnsafeStoragePathError, resolve_storage_path
from dms.storage.uploads import ALLOWED_EXTENSIONS, _validate

logger = logging.getLogger(__name__)

QUERY_FIELDS = {
    "keyword",
    "documentType",
    "categoryId",
    "deletedFrom",
    "deletedTo",
    "page",
    "pageSize",
    "sortField",
    "sortDirection",
}
SORT_FIELDS = {
    "deletedAt": Document.deleted_at,
    "documentName": Document.document_name,
    "documentType": Document.document_type,
    "updatedAt": Document.updated_at,
}
SUPPORTED_TYPES = {
    DocumentType.MAIN.value,
    DocumentType.SUB_PLAN.value,
    DocumentType.ATTACHMENT.value,
}
RESTORE_FIELDS = {"rowVersion"}


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


def _strict_single_params(params: Any) -> None:
    unknown = set(params.keys()) - QUERY_FIELDS
    if unknown:
        raise InvalidArgumentError(
            "存在未知查询参数",
            details={"parameters": sorted(unknown)},
        )
    for name in params.keys():
        if len(params.getlist(name)) != 1:
            raise InvalidArgumentError(
                f"{name}不允许重复传入",
                details={"parameter": name},
            )


def _positive_integer(value: str | None, name: str, default: int) -> int:
    if value is None:
        return default
    if not value.isdecimal() or int(value) < 1:
        raise InvalidArgumentError(f"{name}必须是正整数")
    return int(value)


def _utc_z(value: str | None, name: str) -> datetime | None:
    if value is None or value == "":
        return None
    if not value.endswith("Z"):
        raise InvalidArgumentError(f"{name}必须是UTC Z时间")
    try:
        parsed = datetime.fromisoformat(value[:-1] + "+00:00")
    except ValueError as exc:
        raise InvalidArgumentError(f"{name}必须是有效UTC Z时间") from exc
    if parsed.utcoffset() != timezone.utc.utcoffset(parsed):
        raise InvalidArgumentError(f"{name}必须是UTC Z时间")
    return parsed.astimezone(timezone.utc).replace(tzinfo=None)


def _escape_like(value: str) -> str:
    return value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def _deleted_by(documents: list[Document]) -> dict[int, dict[str, str | None]]:
    """单次窗口查询获取当前页每个文档最近一次可靠删除审计快照。"""
    if not documents:
        return {}
    keys = [
        (
            AuditTarget.ATTACHMENT.value
            if item.document_type == DocumentType.ATTACHMENT.value
            else AuditTarget.DOCUMENT.value,
            item.id,
        )
        for item in documents
    ]
    predicates = [
        and_(AuditLog.target_type == target_type, AuditLog.target_id == target_id)
        for target_type, target_id in keys
    ]
    ranked = (
        select(
            AuditLog.target_type.label("target_type"),
            AuditLog.target_id.label("target_id"),
            AuditLog.user_id.label("user_id"),
            AuditLog.username.label("username"),
            AuditLog.real_name.label("real_name"),
            AuditLog.organization_name.label("organization_name"),
            func.row_number()
            .over(
                partition_by=(AuditLog.target_type, AuditLog.target_id),
                order_by=(AuditLog.created_at.desc(), AuditLog.id.desc()),
            )
            .label("row_number"),
        )
        .where(
            AuditLog.action_type == AuditAction.DELETE_DOCUMENT.value,
            AuditLog.operation_result == AuditResult.SUCCESS.value,
            or_(*predicates),
        )
        .subquery()
    )
    rows = db.session.execute(
        select(ranked).where(ranked.c.row_number == 1)
    ).mappings()
    result: dict[int, dict[str, str | None]] = {}
    for row in rows:
        result[int(row["target_id"])] = {
            "userId": serialize_id(row["user_id"]),
            "username": row["username"],
            "realName": row["real_name"],
            "organizationName": row["organization_name"],
        }
    return result


def _recycle_summary(
    document: Document,
    deleted_by: dict[int, dict[str, str | None]],
) -> dict[str, object]:
    return {
        "id": serialize_id(document.id),
        "documentName": document.document_name,
        "documentType": document.document_type,
        "originalFileName": document.original_file_name,
        "categoryId": serialize_id(document.category_id),
        "categoryName": document.category_name,
        "parentDocumentId": serialize_id(document.parent_document_id),
        "securityLevel": document.security_level,
        "documentStatus": document.document_status,
        "deletedAt": _iso(document.deleted_at),
        "updatedBy": serialize_id(document.updated_by),
        "updatedByName": document.updated_by_name,
        "deletedBy": deleted_by.get(document.id),
        "rowVersion": document.row_version,
    }


def list_recycle_bin_documents(params: Any) -> dict[str, object]:
    _strict_single_params(params)
    page_number = _positive_integer(params.get("page"), "page", 1)
    page_size = _positive_integer(params.get("pageSize"), "pageSize", 20)
    if page_size > 100:
        raise InvalidArgumentError("pageSize不能超过100")
    page = PageRequest(page=page_number, page_size=page_size)

    statement = select(Document).where(
        Document.is_deleted.is_(True),
        Document.deleted_at.is_not(None),
        Document.document_type.in_(sorted(SUPPORTED_TYPES)),
    )
    document_type = params.get("documentType")
    if document_type:
        if document_type not in SUPPORTED_TYPES:
            raise InvalidArgumentError("documentType不是有效枚举值")
        statement = statement.where(Document.document_type == document_type)
    category_id = params.get("categoryId")
    if category_id:
        if not category_id.isdecimal() or int(category_id) <= 0:
            raise InvalidArgumentError("categoryId必须是正整数形式的字符串ID")
        statement = statement.where(Document.category_id == int(category_id))
    keyword = params.get("keyword")
    if keyword and keyword.strip():
        pattern = f"%{_escape_like(keyword.strip())}%"
        statement = statement.where(
            or_(
                Document.document_name.like(pattern, escape="\\"),
                Document.summary.like(pattern, escape="\\"),
                Document.search_text.like(pattern, escape="\\"),
                Document.original_file_name.like(pattern, escape="\\"),
            )
        )
    deleted_from = _utc_z(params.get("deletedFrom"), "deletedFrom")
    deleted_to = _utc_z(params.get("deletedTo"), "deletedTo")
    if deleted_from and deleted_to and deleted_from >= deleted_to:
        raise InvalidArgumentError("deletedFrom必须早于deletedTo")
    if deleted_from:
        statement = statement.where(Document.deleted_at >= deleted_from)
    if deleted_to:
        statement = statement.where(Document.deleted_at < deleted_to)

    sort_field = params.get("sortField", "deletedAt")
    direction = params.get("sortDirection", "desc").lower()
    if sort_field not in SORT_FIELDS:
        raise InvalidArgumentError("sortField不是允许的排序字段")
    if direction not in {"asc", "desc"}:
        raise InvalidArgumentError("sortDirection必须是asc或desc")
    sort_column = SORT_FIELDS[sort_field]
    order = sort_column.asc() if direction == "asc" else sort_column.desc()
    id_order = Document.id.asc() if direction == "asc" else Document.id.desc()

    total = db.session.scalar(
        select(func.count()).select_from(statement.order_by(None).subquery())
    ) or 0
    documents = db.session.scalars(
        statement.order_by(order, id_order).offset(page.offset).limit(page.page_size)
    ).all()
    deleted_by = _deleted_by(documents)
    return page_result(
        [_recycle_summary(item, deleted_by) for item in documents],
        page=page.page,
        page_size=page.page_size,
        total=total,
    )


@dataclass(slots=True)
class VerifiedFile:
    path: Path
    stream: BinaryIO
    state: tuple[int, int, int, int]
    document_state: tuple[str, str, int, str]

    def close(self) -> None:
        self.stream.close()


def _file_state(value: os.stat_result) -> tuple[int, int, int, int]:
    return (value.st_dev, value.st_ino, value.st_size, value.st_mtime_ns)


def _verify_file(document: Document) -> VerifiedFile:
    try:
        path = resolve_storage_path(
            document.file_relative_path,
            current_app.config["DMS_STORAGE_ROOT"],
        )
    except UnsafeStoragePathError as exc:
        logger.error("恢复文件路径校验失败：document_id=%s", document.id)
        raise FilePathInvalidError() from exc
    try:
        stream = path.open("rb")
    except FileNotFoundError as exc:
        raise DmsFileNotFoundError() from exc
    except OSError as exc:
        logger.error("恢复文件无法安全打开：document_id=%s", document.id)
        raise FilePathInvalidError() from exc

    try:
        handle_stat = os.fstat(stream.fileno())
        if not stat.S_ISREG(handle_stat.st_mode):
            raise DmsFileNotFoundError()
        extension = document.file_extension.lower().lstrip(".")
        if extension not in ALLOWED_EXTENSIONS:
            raise UnsupportedFileTypeError()
        if handle_stat.st_size != document.file_size:
            raise FileIntegrityMismatchError(details={"reason": "SIZE_MISMATCH"})
        digest = hashlib.sha256()
        while chunk := stream.read(1024 * 1024):
            digest.update(chunk)
        stream.seek(0)
        if digest.hexdigest() != document.file_hash:
            raise FileIntegrityMismatchError(details={"reason": "HASH_MISMATCH"})
        if path.suffix.lower() != f".{extension}":
            raise FileIntegrityMismatchError(details={"reason": "TYPE_MISMATCH"})
        try:
            _validate(path, extension)
        except UnsupportedFileTypeError as exc:
            raise FileIntegrityMismatchError(
                details={"reason": "TYPE_MISMATCH"}
            ) from exc
        path_stat = path.stat()
        if _file_state(path_stat) != _file_state(handle_stat):
            raise RestoreFileChangedError()
        return VerifiedFile(
            path=path,
            stream=stream,
            state=_file_state(handle_stat),
            document_state=(
                document.file_relative_path,
                document.file_extension,
                document.file_size,
                document.file_hash,
            ),
        )
    except Exception:
        stream.close()
        raise


def _recheck_file(verified: VerifiedFile, document: Document) -> None:
    if verified.document_state != (
        document.file_relative_path,
        document.file_extension,
        document.file_size,
        document.file_hash,
    ):
        raise RestoreFileChangedError()
    try:
        handle_state = _file_state(os.fstat(verified.stream.fileno()))
        path_state = _file_state(verified.path.stat())
    except OSError as exc:
        raise RestoreFileChangedError() from exc
    if handle_state != verified.state or path_state != verified.state:
        raise RestoreFileChangedError()


def _restore_payload(payload: Any) -> int:
    if not isinstance(payload, dict) or set(payload) != RESTORE_FIELDS:
        raise InvalidArgumentError("请求字段必须且只能包含rowVersion")
    value = payload["rowVersion"]
    if type(value) is not int or value < 0:
        raise InvalidArgumentError("rowVersion必须是非负整数")
    return value


def _category_reason(category: Category | None) -> str | None:
    if category is None:
        return "NOT_FOUND"
    if category.is_deleted:
        return "DELETED"
    if category.status != EnabledStatus.ENABLED.value:
        return "DISABLED"
    return None


def _lock_category(session, document: Document) -> Category:
    category = session.scalar(
        select(Category)
        .where(Category.id == document.category_id)
        .with_for_update()
    )
    reason = _category_reason(category)
    if reason:
        raise RestoreCategoryInvalidError(
            details={
                "categoryId": serialize_id(document.category_id),
                "reason": reason,
            }
        )
    return category


def _active_permission_conflicts(
    session,
    document_id: int,
    permissions: list[Permission],
) -> int:
    if not permissions:
        return 0
    keys = {item.subject_id for item in permissions}
    active = session.scalars(
        select(Permission).where(
            Permission.document_id == document_id,
            Permission.is_deleted.is_(False),
        )
    ).all()
    return sum(item.subject_id in keys for item in active)


def _restore_main_relations(
    session,
    document: Document,
    now: datetime,
) -> tuple[int, int, int, int]:
    context = get_auth_context()
    deleted_at = document.deleted_at
    deleted_by = document.updated_by
    permissions = session.scalars(
        select(Permission)
        .where(
            Permission.document_id == document.id,
            Permission.is_deleted.is_(True),
            Permission.deleted_at == deleted_at,
            Permission.updated_by == deleted_by,
        )
        .order_by(Permission.id.asc())
        .with_for_update()
    ).all()

    user_ids = sorted({item.subject_id for item in permissions})
    users = (
        session.scalars(
            select(User)
            .where(User.id.in_(user_ids))
            .order_by(User.id.asc())
            .with_for_update()
        ).all()
        if user_ids
        else []
    )
    valid_users = {
        item.id: item
        for item in users
        if not item.is_deleted and item.status == EnabledStatus.ENABLED.value
    }
    restorable_permissions: list[Permission] = []
    for permission in permissions:
        if permission.subject_id in valid_users:
            restorable_permissions.append(permission)
    skipped_permission_count = len(permissions) - len(restorable_permissions)

    binding_candidates = session.scalars(
        select(AttachmentBinding).where(
            AttachmentBinding.main_document_id == document.id,
            AttachmentBinding.is_deleted.is_(True),
            AttachmentBinding.deleted_at == deleted_at,
            AttachmentBinding.updated_by == deleted_by,
        )
    ).all()
    attachment_ids = sorted(
        {item.attachment_document_id for item in binding_candidates}
    )
    attachments = (
        session.scalars(
            select(Document)
            .where(Document.id.in_(attachment_ids))
            .order_by(Document.id.asc())
            .options(*LOCK_DEFER_LOB)
            .with_for_update()
        ).all()
        if attachment_ids
        else []
    )
    valid_attachment_ids = {
        item.id
        for item in attachments
        if not item.is_deleted and item.document_type == DocumentType.ATTACHMENT.value
    }
    bindings = session.scalars(
        select(AttachmentBinding)
        .where(
            AttachmentBinding.main_document_id == document.id,
            AttachmentBinding.is_deleted.is_(True),
            AttachmentBinding.deleted_at == deleted_at,
            AttachmentBinding.updated_by == deleted_by,
        )
        .order_by(AttachmentBinding.id.asc())
        .with_for_update()
    ).all()
    restorable_bindings = [
        item for item in bindings if item.attachment_document_id in valid_attachment_ids
    ]
    skipped_binding_count = len(bindings) - len(restorable_bindings)

    permission_conflicts = _active_permission_conflicts(
        session, document.id, restorable_permissions
    )
    binding_keys = {item.attachment_document_id for item in restorable_bindings}
    binding_conflicts = (
        session.scalar(
            select(func.count(AttachmentBinding.id)).where(
                AttachmentBinding.main_document_id == document.id,
                AttachmentBinding.attachment_document_id.in_(binding_keys),
                AttachmentBinding.is_deleted.is_(False),
            )
        )
        if binding_keys
        else 0
    ) or 0
    if permission_conflicts or binding_conflicts:
        raise RestoreRelationConflictError(
            details={
                "permissionConflictCount": permission_conflicts,
                "bindingConflictCount": binding_conflicts,
            }
        )

    # 借阅语义（0026）：授权仅面向用户；还原后已过期条目由读路径惰性判定
    # 自行失效，无需还原期校验可见范围。
    for permission in restorable_permissions:
        permission.subject_name = valid_users[permission.subject_id].real_name
        permission.is_deleted = False
        permission.deleted_at = None
        permission.updated_at = now
        permission.updated_by = context.user_id
        permission.row_version += 1
    for binding in restorable_bindings:
        binding.is_deleted = False
        binding.deleted_at = None
        binding.updated_at = now
        binding.updated_by = context.user_id
        binding.row_version += 1
    return (
        len(restorable_permissions),
        skipped_permission_count,
        len(restorable_bindings),
        skipped_binding_count,
    )


def _recount_category(session, category: Category, now: datetime) -> None:
    context = get_auth_context()
    count = session.scalar(
        select(func.count(Document.id)).where(
            Document.category_id == category.id,
            Document.document_type.in_(
                [DocumentType.MAIN.value, DocumentType.SUB_PLAN.value]
            ),
            Document.is_deleted.is_(False),
        )
    ) or 0
    if category.document_count != count:
        category.document_count = count
        category.row_version += 1
        category.updated_by = context.user_id
        category.updated_at = now


def _restore_operation(
    document_id: int,
    expected_version: int,
    verified: VerifiedFile,
) -> tuple[Document, tuple[int, int, int, int]]:
    context = get_auth_context()
    with transaction() as session:
        identity = session.execute(
            select(
                Document.document_type,
                Document.parent_document_id,
                Document.root_document_id,
            ).where(Document.id == document_id)
        ).one_or_none()
        if identity is None:
            raise ResourceNotFoundError("文档不存在")
        document_type, parent_id, root_id = identity
        if document_type not in SUPPORTED_TYPES:
            raise ResourceNotFoundError("文档不存在或类型不支持")

        parent: Document | None = None
        if document_type == DocumentType.SUB_PLAN.value:
            lock_parent_id = root_id or parent_id
            parent = session.scalar(
                select(Document)
                .where(Document.id == lock_parent_id)
                .options(*LOCK_DEFER_LOB)
                .with_for_update()
            )
        document = session.scalar(
            select(Document)
            .where(Document.id == document_id)
            .options(*LOCK_DEFER_LOB)
            .with_for_update()
        )
        if document is None:
            raise ResourceNotFoundError("文档不存在")
        if not document.is_deleted or document.deleted_at is None:
            raise DocumentNotDeletedError(
                details={"documentId": serialize_id(document.id)}
            )
        if document.row_version != expected_version:
            raise ConflictError(
                "数据已被其他用户修改，请刷新后重试",
                details={"currentRowVersion": document.row_version},
            )

        category: Category | None = None
        if document.document_type == DocumentType.MAIN.value:
            category = _lock_category(session, document)
            active_child_count = session.scalar(
                select(func.count(Category.id)).where(
                    Category.parent_id == category.id,
                    Category.is_deleted.is_(False),
                    Category.status == EnabledStatus.ENABLED.value,
                )
            ) or 0
            if active_child_count:
                raise RestoreCategoryInvalidError(
                    details={
                        "categoryId": serialize_id(category.id),
                        "reason": "NOT_LEAF",
                        "childCategoryCount": int(active_child_count),
                    }
                )

        if document.document_type == DocumentType.SUB_PLAN.value:
            if (
                document.parent_document_id is None
                or document.root_document_id is None
                or document.parent_document_id != document.root_document_id
            ):
                raise RestoreParentInvalidError(
                    details={
                        "parentDocumentId": serialize_id(document.parent_document_id),
                        "reason": "ROOT_MISMATCH",
                    }
                )
            reason = None
            if parent is None:
                reason = "NOT_FOUND"
            elif parent.is_deleted:
                reason = "DELETED"
            elif parent.document_type != DocumentType.MAIN.value:
                reason = "WRONG_TYPE"
            if reason:
                raise RestoreParentInvalidError(
                    details={
                        "parentDocumentId": serialize_id(document.parent_document_id),
                        "reason": reason,
                    }
                )
            category = session.scalar(
                select(Category)
                .where(Category.id == parent.category_id)
                .with_for_update()
            )
            category_reason = _category_reason(category)
            if category_reason:
                raise RestoreCategoryInvalidError(
                    details={
                        "categoryId": serialize_id(parent.category_id),
                        "reason": category_reason,
                    }
                )
            document.category_id = parent.category_id
            document.category_name = parent.category_name
            document.category_path = parent.category_path

        counts = (0, 0, 0, 0)
        now = _now()
        deleted_at_before = document.deleted_at
        row_version_before = document.row_version
        if document.document_type == DocumentType.MAIN.value:
            counts = _restore_main_relations(session, document, now)
        elif document.document_type == DocumentType.ATTACHMENT.value:
            shape_valid = (
                document.security_level == SecurityLevel.PUBLIC.value
                and document.category_id is None
                and document.parent_document_id is None
                and document.root_document_id is None
            )
            if not shape_valid:
                raise RestoreRelationConflictError(
                    details={
                        "permissionConflictCount": 0,
                        "bindingConflictCount": 0,
                    }
                )
            active_bindings = session.scalars(
                select(AttachmentBinding)
                .where(
                    AttachmentBinding.attachment_document_id == document.id,
                    AttachmentBinding.is_deleted.is_(False),
                )
                .order_by(AttachmentBinding.id.asc())
                .with_for_update()
            ).all()
            if active_bindings:
                raise RestoreRelationConflictError(
                    details={
                        "permissionConflictCount": 0,
                        "bindingConflictCount": len(active_bindings),
                    }
                )

        document.is_deleted = False
        document.deleted_at = None
        document.updated_at = now
        document.updated_by = context.user_id
        document.updated_by_name = context.real_name
        document.row_version += 1
        session.flush()

        if document.document_type == DocumentType.MAIN.value:
            document.child_count = session.scalar(
                select(func.count(Document.id)).where(
                    Document.document_type == DocumentType.SUB_PLAN.value,
                    Document.parent_document_id == document.id,
                    Document.root_document_id == document.id,
                    Document.is_deleted.is_(False),
                )
            ) or 0
            document.attachment_count = session.scalar(
                select(func.count(AttachmentBinding.id)).where(
                    AttachmentBinding.main_document_id == document.id,
                    AttachmentBinding.is_deleted.is_(False),
                )
            ) or 0
        elif document.document_type == DocumentType.SUB_PLAN.value:
            assert parent is not None
            parent.child_count = session.scalar(
                select(func.count(Document.id)).where(
                    Document.document_type == DocumentType.SUB_PLAN.value,
                    Document.parent_document_id == parent.id,
                    Document.root_document_id == parent.id,
                    Document.is_deleted.is_(False),
                )
            ) or 0
            parent.row_version += 1
            parent.updated_at = now
            parent.updated_by = context.user_id
            parent.updated_by_name = context.real_name
        if category is not None:
            _recount_category(session, category, now)

        _recheck_file(verified, document)
        session.add(
            business_audit(
                action=AuditAction.RESTORE_DOCUMENT,
                target=(
                    AuditTarget.ATTACHMENT
                    if document.document_type == DocumentType.ATTACHMENT.value
                    else AuditTarget.DOCUMENT
                ),
                target_id=document.id,
                target_name=document.document_name,
                detail={
                    "documentType": document.document_type,
                    "rowVersionBefore": row_version_before,
                    "rowVersionAfter": document.row_version,
                    "deletedAtBefore": _iso(deleted_at_before),
                    "restoredPermissionCount": counts[0],
                    "skippedPermissionCount": counts[1],
                    "restoredBindingCount": counts[2],
                    "skippedBindingCount": counts[3],
                    "parentDocumentId": serialize_id(document.parent_document_id),
                    "categoryId": serialize_id(document.category_id),
                },
            )
        )
        return document, counts


def _pure_detail(document: Document) -> dict[str, object]:
    context = get_auth_context()
    if document.document_type == DocumentType.ATTACHMENT.value:
        result = attachment_detail(document, context)
    else:
        result = document_detail(
            document,
            context,
            evaluate_plan_access(document, context),
        )
    result.pop("fileHash", None)
    return result


def restore_document(document_id: int, payload: Any) -> dict[str, object]:
    expected_version = _restore_payload(payload)
    document = db.session.get(Document, document_id)
    if document is None:
        raise ResourceNotFoundError("文档不存在")
    if document.document_type not in SUPPORTED_TYPES:
        raise ResourceNotFoundError("文档不存在或类型不支持")
    if not document.is_deleted or document.deleted_at is None:
        raise DocumentNotDeletedError(
            details={"documentId": serialize_id(document.id)}
        )
    if document.row_version != expected_version:
        raise ConflictError(
            "数据已被其他用户修改，请刷新后重试",
            details={"currentRowVersion": document.row_version},
        )
    verified = _verify_file(document)
    db.session.rollback()
    try:
        try:
            restored, counts = execute_with_deadlock_retry(
                lambda: _restore_operation(document_id, expected_version, verified)
            )
        except IntegrityError as exc:
            db.session.rollback()
            raise RestoreRelationConflictError(
                details={
                    "permissionConflictCount": 0,
                    "bindingConflictCount": 0,
                }
            ) from exc
        result = {
            "document": _pure_detail(restored),
            "restoredPermissionCount": counts[0],
            "skippedPermissionCount": counts[1],
            "restoredBindingCount": counts[2],
            "skippedBindingCount": counts[3],
        }
        return result
    finally:
        verified.close()


__all__ = ["list_recycle_bin_documents", "restore_document"]