"""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, SubjectType, VisibilityType, ) from dms.common.errors import ( ConflictError, DocumentNotDeletedError, FileIntegrityMismatchError, FileNotFoundError as DmsFileNotFoundError, FilePathInvalidError, InvalidArgumentError, ResourceNotFoundError, RestoreCategoryInvalidError, RestoreFileChangedError, RestoreParentInvalidError, RestorePermissionInvalidError, 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 ( AttachmentBinding, AuditLog, Category, Document, Organization, 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_type, 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_type, 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() organization_ids = sorted( {item.subject_id for item in permissions if item.subject_type == SubjectType.ORG} ) user_ids = sorted( {item.subject_id for item in permissions if item.subject_type == SubjectType.USER} ) organizations = ( session.scalars( select(Organization) .where(Organization.id.in_(organization_ids)) .order_by(Organization.id.asc()) .with_for_update() ).all() if organization_ids else [] ) 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_organizations = { item.id: item for item in organizations if not item.is_deleted and item.status == EnabledStatus.ENABLED.value } 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_type == SubjectType.ORG.value: if permission.subject_id in valid_organizations: restorable_permissions.append(permission) elif 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()) .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, } ) active_permissions = session.scalars( select(Permission).where( Permission.document_id == document.id, Permission.is_deleted.is_(False), ) ).all() final_permissions = [*active_permissions, *restorable_permissions] if document.visibility_type == VisibilityType.ALL_AUTHENTICATED.value: if final_permissions: raise RestorePermissionInvalidError( details={ "invalidCount": len(final_permissions), "reason": "ALL_AUTHENTICATED_HAS_PERMISSIONS", } ) elif document.visibility_type == VisibilityType.ORGANIZATION.value: valid_org_view = any( item.subject_type == SubjectType.ORG.value and item.can_view for item in final_permissions ) if not valid_org_view: raise RestorePermissionInvalidError( details={ "invalidCount": skipped_permission_count, "reason": "NO_VALID_ORGANIZATION_PERMISSION", } ) for permission in restorable_permissions: if permission.subject_type == SubjectType.ORG.value: permission.subject_name = valid_organizations[permission.subject_id].org_name else: 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) .with_for_update() ) document = session.scalar( select(Document) .where(Document.id == document_id) .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.ATTACHMENT.value: category = _lock_category(session, document) 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, } ) 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.visibility_type == VisibilityType.ALL_AUTHENTICATED.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"]