"""B5文档与共享附件写操作及文件事务补偿。""" from __future__ import annotations from datetime import datetime, timezone import logging import tempfile from typing import Any from flask import current_app from sqlalchemy import func, or_, select from werkzeug.datastructures import FileStorage from dms.common.enums import ( AttachmentType, AuditAction, AuditTarget, DocumentStatus, DocumentType, EnabledStatus, SecurityLevel, VisibilityType, ) from dms.common.errors import ( AttachmentInUseError, CategoryNotLeafError, ConflictError, DocumentNameConflictError, InvalidArgumentError, MainPlanHasChildrenError, ResourceNotFoundError, ) from dms.common.response import serialize_id from dms.database.transaction import transaction from dms.extensions import db from dms.models import AttachmentBinding, Category, Document, Permission 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 document_detail from dms.services.document_content_service import ( ContentExtraction, build_search_text, extract_document_content, ) from dms.services.preview_service import extract_doc_content_via_pdf from dms.storage.uploads import StagedUpload, stage_upload from dms.storage.paths import UnsafeStoragePathError, resolve_storage_path logger = logging.getLogger(__name__) MAIN_CREATE_FIELDS = { "documentName", "documentType", "summary", "categoryId", "securityLevel", "tags", } OVERWRITE_FIELDS = {"overwriteDocumentId", "rowVersion"} SUB_PLAN_CREATE_FIELDS = { "documentName", "documentType", "summary", "parentDocumentId", "securityLevel", "tags", } MAIN_EDIT_FIELDS = { "documentName", "summary", "categoryId", "securityLevel", "tags", "rowVersion", } SUB_PLAN_EDIT_FIELDS = { "documentName", "summary", "securityLevel", "tags", "rowVersion", } ATTACHMENT_CREATE_FIELDS = {"documentName", "attachmentType", "summary", "tags"} ATTACHMENT_EDIT_FIELDS = ATTACHMENT_CREATE_FIELDS | {"rowVersion"} def _now() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) def _exact(payload: Any, fields: set[str]) -> dict[str, Any]: if not isinstance(payload, dict) or set(payload) != fields: raise InvalidArgumentError( "metadata字段必须且只能包含:" + "、".join(sorted(fields)) ) return payload def _text(value: Any, name: str, limit: int, *, nullable: bool = False): if value is None and nullable: return None if not isinstance(value, str) or not value.strip(): raise InvalidArgumentError(f"{name}不能为空") result = value.strip() if len(result) > limit: raise InvalidArgumentError(f"{name}长度不能超过{limit}") return result def _tags(value: Any) -> list[str]: if not isinstance(value, list) or len(value) > 50: raise InvalidArgumentError("tags必须是最多50项的字符串数组") result: list[str] = [] for item in value: normalized = _text(item, "tags元素", 64) if normalized not in result: result.append(normalized) return result def _enum(value: Any, enum_type, name: str) -> str: if not isinstance(value, str): raise InvalidArgumentError(f"{name}必须是字符串枚举") try: return enum_type(value).value except ValueError as exc: raise InvalidArgumentError(f"{name}不是有效枚举值") from exc 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("rowVersion必须是非负整数") return value def _category(category_id: int) -> Category: category = db.session.scalar( select(Category).where( Category.id == category_id, Category.is_deleted.is_(False), Category.status == EnabledStatus.ENABLED.value, ) ) if category is None: raise ResourceNotFoundError("方案分类不存在或不可用") return category def _leaf_category(category_id: int) -> Category: category = _category(category_id) child_count = db.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 child_count: raise CategoryNotLeafError( details={ "categoryId": serialize_id(category.id), "childCategoryCount": int(child_count), } ) return category def _serialize(document: Document) -> dict[str, object]: context = get_auth_context() if document.document_type == DocumentType.ATTACHMENT.value: return attachment_detail(document, context) return document_detail(document, context, evaluate_plan_access(document, context)) def _create_record( upload: StagedUpload, payload: dict[str, Any], *, attachment: bool, batch: bool, ) -> dict[str, object]: if not isinstance(payload, dict): raise InvalidArgumentError("metadata必须是JSON对象") context = get_auth_context() actor_id = context.user_id document_type = DocumentType.ATTACHMENT.value category = None parent = None attachment_type = None if attachment: _exact(payload, ATTACHMENT_CREATE_FIELDS) attachment_type = _enum( payload["attachmentType"], AttachmentType, "attachmentType" ) security = SecurityLevel.PUBLIC.value visibility = VisibilityType.ALL_AUTHENTICATED.value status = DocumentStatus.PUBLISHED.value else: raw_type = payload.get("documentType") try: document_type = DocumentType(raw_type).value except (ValueError, TypeError) as exc: raise InvalidArgumentError("documentType只允许MAIN或SUB_PLAN") from exc if document_type not in { DocumentType.MAIN.value, DocumentType.SUB_PLAN.value, }: raise InvalidArgumentError("documentType只允许MAIN或SUB_PLAN") create_fields = ( SUB_PLAN_CREATE_FIELDS if document_type == DocumentType.SUB_PLAN.value else MAIN_CREATE_FIELDS ) if frozenset(payload) not in { frozenset(create_fields), frozenset(create_fields | OVERWRITE_FIELDS), }: raise InvalidArgumentError( "metadata字段必须为新增字段全集,或额外同时包含overwriteDocumentId和rowVersion" ) security = _enum(payload["securityLevel"], SecurityLevel, "securityLevel") visibility = VisibilityType.CUSTOM.value status = DocumentStatus.PUBLISHED.value if document_type == DocumentType.SUB_PLAN.value: parent_id = _string_id(payload["parentDocumentId"], "parentDocumentId") parent = db.session.scalar( select(Document).where( Document.id == parent_id, Document.document_type == DocumentType.MAIN.value, Document.is_deleted.is_(False), ) ) if parent is None: raise ResourceNotFoundError("父文档必须是有效主案") visibility = parent.visibility_type category = _category(parent.category_id) else: category = _leaf_category( _string_id(payload["categoryId"], "categoryId") ) name = _text(payload["documentName"], "documentName", 255) summary = _text(payload["summary"], "summary", 20000, nullable=True) tags = _tags(payload["tags"]) if upload.extension == "doc": temp_pdf: Path | None = None try: from dms.services.document_converter import create_converter converter = create_converter(current_app.config) if not converter.available: extraction = ContentExtraction( text=None, status="FAILED", extracted_at=_now(), ) else: pdf_bytes = converter.convert_to_pdf( upload.temporary_path, upload.file_hash ) with tempfile.NamedTemporaryFile( suffix=".pdf", delete=False ) as temp_file: temp_file.write(pdf_bytes) temp_pdf = Path(temp_file.name) extraction = extract_document_content(temp_pdf, "pdf") except Exception: logger.exception("DOC转换PDF提取正文失败") extraction = ContentExtraction( text=None, status="FAILED", extracted_at=_now(), ) finally: if temp_pdf is not None: temp_pdf.unlink(missing_ok=True) else: extraction = extract_document_content(upload.temporary_path, upload.extension) final_created = False committed = False old_file_path = None try: with transaction() as session: duplicate = None if not attachment: duplicate_conditions = [ Document.document_name == name, Document.document_type == document_type, Document.is_deleted.is_(False), ] if document_type == DocumentType.SUB_PLAN.value: duplicate_conditions.append(Document.parent_document_id == parent.id) duplicate = session.scalar( select(Document).where(*duplicate_conditions).with_for_update() ) overwrite_id = payload.get("overwriteDocumentId") if duplicate is not None and overwrite_id is None: raise DocumentNameConflictError( details={ "existingDocumentId": serialize_id(duplicate.id), "existingDocumentName": duplicate.document_name, "existingRowVersion": duplicate.row_version, "documentType": duplicate.document_type, "parentDocumentId": serialize_id(duplicate.parent_document_id), } ) if overwrite_id is not None: expected_id = _string_id(overwrite_id, "overwriteDocumentId") expected_version = _version(payload["rowVersion"]) if duplicate is None or duplicate.id != expected_id: raise ConflictError( "同名方案已发生变化,请重新确认", details={"currentRowVersion": duplicate.row_version if duplicate else None}, ) _check_version(duplicate, expected_version) document = duplicate old_category_id = document.category_id if document_type == DocumentType.MAIN.value and old_category_id != category.id: old_category = session.get(Category, old_category_id) if old_category is not None: old_category.document_count = max(0, old_category.document_count - 1) old_category.row_version += 1 category.document_count += 1 category.row_version += 1 try: old_file_path = resolve_storage_path( document.file_relative_path, current_app.config["DMS_STORAGE_ROOT"], ) except UnsafeStoragePathError: logger.error("覆盖方案时检测到异常旧文件路径:document_id=%s", document.id) old_file_path = None document.summary = summary document.security_level = security document.category_id = category.id if category else None document.category_name = category.category_name if category else None document.category_path = category.category_path if category else None document.tags = tags document.original_file_name = upload.original_file_name document.file_relative_path = upload.relative_path document.file_extension = upload.extension document.mime_type = upload.mime_type document.file_size = upload.file_size document.file_hash = upload.file_hash document.content_text = extraction.text document.content_extract_status = extraction.status document.content_extracted_at = extraction.extracted_at document.search_text = build_search_text(name, summary, tags, extraction.text) document.updated_by = actor_id document.updated_by_name = context.real_name document.updated_at = _now() document.row_version += 1 if document_type == DocumentType.MAIN.value and old_category_id != category.id: session.query(Document).filter( Document.parent_document_id == document.id, Document.document_type == DocumentType.SUB_PLAN.value, Document.is_deleted.is_(False), ).update( { Document.category_id: category.id, Document.category_name: category.category_name, Document.category_path: category.category_path, Document.updated_by: actor_id, Document.updated_by_name: context.real_name, Document.updated_at: _now(), Document.row_version: Document.row_version + 1, }, synchronize_session=False, ) session.add( business_audit( action=AuditAction.UPLOAD_DOCUMENT, target=AuditTarget.DOCUMENT, target_id=document.id, target_name=document.document_name, detail={ "overwrite": True, "originalFileName": document.original_file_name, "fileSize": document.file_size, "rowVersion": document.row_version, }, ) ) upload.promote() final_created = True else: document = Document( document_name=name, summary=summary, document_type=document_type, document_status=status, security_level=security, visibility_type=visibility, attachment_type=attachment_type, category_id=category.id if category else None, category_name=category.category_name if category else None, category_path=category.category_path if category else None, parent_document_id=parent.id if parent else None, root_document_id=parent.id if parent else None, tags=tags, original_file_name=upload.original_file_name, file_relative_path=upload.relative_path, file_extension=upload.extension, mime_type=upload.mime_type, file_size=upload.file_size, file_hash=upload.file_hash, content_text=extraction.text, content_extract_status=extraction.status, content_extracted_at=extraction.extracted_at, search_text=build_search_text( name, summary, tags, extraction.text ), created_by=actor_id, updated_by=actor_id, created_by_name=context.real_name, updated_by_name=context.real_name, ) session.add(document) session.flush() if category is not None and document_type == DocumentType.MAIN.value: category.document_count += 1 category.row_version += 1 category.updated_by = actor_id if parent is not None: parent.child_count += 1 parent.row_version += 1 parent.updated_by = actor_id parent.updated_by_name = context.real_name session.add( business_audit( action=( AuditAction.BATCH_IMPORT if batch else AuditAction.UPLOAD_DOCUMENT ), target=( AuditTarget.ATTACHMENT if attachment else AuditTarget.DOCUMENT ), target_id=document.id, target_name=document.document_name, detail={ "documentType": document.document_type, "originalFileName": document.original_file_name, "fileSize": document.file_size, }, ) ) upload.promote() final_created = True committed = True if old_file_path is not None and old_file_path != upload.final_path: try: old_file_path.unlink(missing_ok=True) except OSError: logger.exception("覆盖成功后清理旧文件失败:document_id=%s", document.id) except Exception: if final_created and not committed: upload.final_path.unlink(missing_ok=True) raise finally: upload.cleanup() return _serialize(document) def create_document(file: FileStorage, payload: dict[str, Any], *, batch=False): upload = stage_upload(file) return _create_record(upload, payload, attachment=False, batch=batch) def create_attachment(file: FileStorage, payload: dict[str, Any], *, batch=False): upload = stage_upload(file) return _create_record(upload, payload, attachment=True, batch=batch) def _active(document_id: int, types: set[str]) -> Document: document = db.session.scalar( select(Document).where( Document.id == document_id, Document.document_type.in_(types), Document.is_deleted.is_(False), ) ) if document is None: raise ResourceNotFoundError("文档不存在或类型不匹配") return document def _check_version(document: Document, expected: int) -> None: if document.row_version != expected: raise ConflictError( "数据已被其他用户修改,请刷新后重试", details={"currentRowVersion": document.row_version}, ) def update_document(document_id: int, payload: dict[str, Any]): context = get_auth_context() with transaction() as session: document = session.scalar( select(Document) .where( Document.id == document_id, Document.document_type.in_( [DocumentType.MAIN.value, DocumentType.SUB_PLAN.value] ), Document.is_deleted.is_(False), ) .with_for_update() ) if document is None: raise ResourceNotFoundError("方案文档不存在或类型不匹配") _exact( payload, MAIN_EDIT_FIELDS if document.document_type == DocumentType.MAIN.value else SUB_PLAN_EDIT_FIELDS, ) _check_version(document, _version(payload["rowVersion"])) if document.document_type == DocumentType.MAIN.value: category = _leaf_category( _string_id(payload["categoryId"], "categoryId") ) else: parent = session.scalar( select(Document).where( Document.id == document.parent_document_id, Document.document_type == DocumentType.MAIN.value, Document.is_deleted.is_(False), ) ) if parent is None: raise ResourceNotFoundError("父文档必须是有效主案") category = _category(parent.category_id) before = { "documentName": document.document_name, "categoryId": serialize_id(document.category_id), "securityLevel": document.security_level, "visibilityType": document.visibility_type, "status": document.document_status, } category_changed = document.category_id != category.id if document.document_type == DocumentType.MAIN.value and category_changed: old = session.get(Category, document.category_id) if old is not None: old.document_count = max(0, old.document_count - 1) old.row_version += 1 category.document_count += 1 category.row_version += 1 document.document_name = _text( payload["documentName"], "documentName", 255 ) document.summary = _text( payload["summary"], "summary", 20000, nullable=True ) document.tags = _tags(payload["tags"]) document.category_id = category.id document.category_name = category.category_name document.category_path = category.category_path document.security_level = _enum( payload["securityLevel"], SecurityLevel, "securityLevel" ) if document.document_type == DocumentType.SUB_PLAN.value: document.visibility_type = parent.visibility_type document.search_text = build_search_text( document.document_name, document.summary, document.tags, document.content_text, ) document.updated_by = context.user_id document.updated_by_name = context.real_name document.updated_at = _now() document.row_version += 1 if document.document_type == DocumentType.MAIN.value and category_changed: session.query(Document).filter( Document.parent_document_id == document.id, Document.document_type == DocumentType.SUB_PLAN.value, Document.is_deleted.is_(False), ).update( { Document.category_id: category.id, Document.category_name: category.category_name, Document.category_path: category.category_path, Document.updated_by: context.user_id, Document.updated_by_name: context.real_name, Document.updated_at: _now(), Document.row_version: Document.row_version + 1, }, synchronize_session=False, ) session.add( business_audit( action=AuditAction.EDIT_DOCUMENT, target=AuditTarget.DOCUMENT, target_id=document.id, target_name=document.document_name, detail={"before": before, "rowVersion": document.row_version}, ) ) return _serialize(document) def update_attachment(document_id: int, payload: dict[str, Any]): _exact(payload, ATTACHMENT_EDIT_FIELDS) context = get_auth_context() with transaction() as session: document = session.scalar( select(Document) .where( Document.id == document_id, Document.document_type == DocumentType.ATTACHMENT.value, Document.is_deleted.is_(False), ) .with_for_update() ) if document is None: raise ResourceNotFoundError("共享附件不存在或类型不匹配") _check_version(document, _version(payload["rowVersion"])) before = { "documentName": document.document_name, "attachmentType": document.attachment_type, } document.document_name = _text( payload["documentName"], "documentName", 255 ) document.attachment_type = _enum( payload["attachmentType"], AttachmentType, "attachmentType" ) document.summary = _text( payload["summary"], "summary", 20000, nullable=True ) document.tags = _tags(payload["tags"]) document.search_text = build_search_text( document.document_name, document.summary, document.tags, document.content_text, ) document.updated_by = context.user_id document.updated_by_name = context.real_name document.updated_at = _now() document.row_version += 1 session.add( business_audit( action=AuditAction.EDIT_DOCUMENT, target=AuditTarget.ATTACHMENT, target_id=document.id, target_name=document.document_name, detail={"before": before, "rowVersion": document.row_version}, ) ) return _serialize(document) def delete_document(document_id: int, row_version: int): context = get_auth_context() now = _now() with transaction() as session: document = session.scalar( select(Document) .where( Document.id == document_id, Document.document_type.in_( [DocumentType.MAIN.value, DocumentType.SUB_PLAN.value] ), Document.is_deleted.is_(False), ) .with_for_update() ) if document is None: raise ResourceNotFoundError("方案文档不存在或类型不匹配") _check_version(document, row_version) if document.document_type == DocumentType.MAIN.value: child_count = session.scalar( select(func.count(Document.id)).where( or_( Document.parent_document_id == document.id, Document.root_document_id == document.id, ), Document.document_type == DocumentType.SUB_PLAN.value, Document.is_deleted.is_(False), ) ) if child_count: raise MainPlanHasChildrenError( details={"childCount": child_count} ) for relation in session.scalars( select(Permission).where( Permission.document_id == document.id, Permission.is_deleted.is_(False), ) ): relation.is_deleted = True relation.deleted_at = now relation.updated_at = now relation.updated_by = context.user_id relation.row_version += 1 for binding in session.scalars( select(AttachmentBinding).where( AttachmentBinding.main_document_id == document.id, AttachmentBinding.is_deleted.is_(False), ) ): binding.is_deleted = True binding.deleted_at = now binding.updated_at = now binding.updated_by = context.user_id binding.row_version += 1 document.attachment_count = 0 else: parent = session.get(Document, document.root_document_id) if parent is not None and not parent.is_deleted: parent.child_count = max(0, parent.child_count - 1) parent.updated_at = now parent.updated_by = context.user_id parent.updated_by_name = context.real_name parent.row_version += 1 category = session.get(Category, document.category_id) if category is not None and document.document_type == DocumentType.MAIN.value: category.document_count = max(0, category.document_count - 1) category.row_version += 1 category.updated_by = context.user_id document.is_deleted = True document.deleted_at = now document.updated_at = now document.updated_by = context.user_id document.updated_by_name = context.real_name document.row_version += 1 session.add( business_audit( action=AuditAction.DELETE_DOCUMENT, target=AuditTarget.DOCUMENT, target_id=document.id, target_name=document.document_name, detail={"documentType": document.document_type}, ) ) return {"id": serialize_id(document.id), "deleted": True} def delete_attachment(document_id: int, row_version: int): context = get_auth_context() now = _now() with transaction() as session: document = session.scalar( select(Document) .where( Document.id == document_id, Document.document_type == DocumentType.ATTACHMENT.value, Document.is_deleted.is_(False), ) .with_for_update() ) if document is None: raise ResourceNotFoundError("共享附件不存在或类型不匹配") _check_version(document, row_version) rows = session.execute( select(AttachmentBinding, Document) .join(Document, Document.id == AttachmentBinding.main_document_id) .where( AttachmentBinding.attachment_document_id == document.id, AttachmentBinding.is_deleted.is_(False), Document.document_type == DocumentType.MAIN.value, Document.is_deleted.is_(False), ) ).all() if rows: raise AttachmentInUseError( details={ "mountedPlanCount": len(rows), "mainPlans": [ { "id": serialize_id(main.id), "documentName": main.document_name, } for _, main in rows ], } ) document.is_deleted = True document.deleted_at = now document.updated_at = now document.updated_by = context.user_id document.updated_by_name = context.real_name document.row_version += 1 session.add( business_audit( action=AuditAction.DELETE_DOCUMENT, target=AuditTarget.ATTACHMENT, target_id=document.id, target_name=document.document_name, detail={"attachmentType": document.attachment_type}, ) ) return {"id": serialize_id(document.id), "deleted": True}