"""B5文档与共享附件写操作及文件事务补偿。""" from __future__ import annotations from datetime import datetime, timezone import logging from pathlib import Path 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 ( AllowedAction, AttachmentType, AuditAction, AuditTarget, DocumentStatus, DocumentType, EnabledStatus, RoleCode, SecurityLevel, ) from dms.common.errors import ( AttachmentInUseError, CategoryNotLeafError, ConflictError, DocumentNameConflictError, DocumentNotInMoveScopeError, FileNotFoundError as DmsFileNotFoundError, FilePathInvalidError, ForbiddenError, InvalidArgumentError, MainPlanHasChildrenError, ResourceNotFoundError, ) 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, Category, Document, Organization, 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 ( can_upload_plan, evaluate_plan_access, require_plan_action, ) from dms.services.document_password_service import ( DocumentPasswordAction, _pop_password_spec, apply_password_lock, get_request_password, require_document_password, ) from dms.services.document_query_service import ( LOCK_DEFER_LOB, _category_scope, document_detail, ) from dms.services.document_content_service import ( ContentExtraction, build_search_text, extract_document_content, sanitize_extract_error, ) 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"} # 详细说明:独立于元数据编辑的备注字段(主案/子案/附件通用), # 仅 {"description"},不参与 rowVersion 乐观锁(备注型轻量更新) DESCRIPTION_EDIT_FIELDS = {"description"} 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 _resolve_organization( session, organization_id_input: int | None, context, current: tuple[int | None, str | None] | None = None, ) -> tuple[int | None, str | None]: """解析文档归属部门。 - 输入为空:新建(current为None)取当前用户的部门(与创建者当前部门一致); 编辑则沿用文档现有归属,不会被编辑人部门静默覆盖。 - 输入与当前用户部门相同:直接复用上下文冗余字段,免一次查询。 - 输入与文档现有归属相同:原样回传视为未变更,免一次查询 (允许非管理员原样提交其他部门文档的归属)。 - 输入为其他部门:仅管理员可调整,按 ID 校验部门可用性并取最新名称 (防止借编辑调整文档的权限范围)。 """ if organization_id_input is None: if current is not None: return current return context.organization_id, context.organization_name if ( context.organization_id is not None and organization_id_input == context.organization_id ): return context.organization_id, context.organization_name if current is not None and organization_id_input == current[0]: return current if context.role_code != RoleCode.ADMIN: raise InvalidArgumentError("仅管理员可将文档调整到其他部门") organization = session.scalar( select(Organization).where( Organization.id == organization_id_input, Organization.is_deleted.is_(False), Organization.status == EnabledStatus.ENABLED.value, ) ) if organization is None: raise ResourceNotFoundError("所属部门不存在或不可用") return organization.id, organization.org_name 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)) _OFFICE_CONVERT_EXTENSIONS = {"doc", "wps", "xls", "xlsx", "et"} def _extract_file_content( file_path: Path, extension: str, file_hash: str ) -> ContentExtraction: """对落盘文件提取正文;Office 类型先经 LibreOffice 转 PDF 再提取。 上传(_create_record)与重新解析(re_extract_content)共用,保证口径一致。 任何异常都收敛为 FAILED 并携带脱敏错误信息,不向上抛出。 """ if extension not in _OFFICE_CONVERT_EXTENSIONS: return extract_document_content(file_path, extension) temp_pdf: Path | None = None try: from dms.services.document_converter import create_converter converter = create_converter(current_app.config) if not converter.available: return ContentExtraction( text=None, status="FAILED", extracted_at=_now(), error="LibreOffice转换器不可用,无法解析Office文档", ) pdf_bytes = converter.convert_to_pdf(file_path, file_hash) with tempfile.NamedTemporaryFile( suffix=".pdf", delete=False ) as temp_file: temp_file.write(pdf_bytes) temp_pdf = Path(temp_file.name) return extract_document_content(temp_pdf, "pdf") except Exception as exc: logger.exception("DOC/WPS/XLS/XLSX/ET转换PDF提取正文失败") return ContentExtraction( text=None, status="FAILED", extracted_at=_now(), error=sanitize_extract_error(exc), ) finally: if temp_pdf is not None: temp_pdf.unlink(missing_ok=True) 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 organization_id_input: int | None = None if "organizationId" in payload: organization_id_input = _string_id( payload.pop("organizationId"), "organizationId" ) password, protected_actions = _pop_password_spec( payload, attachment=attachment ) if attachment: _exact(payload, ATTACHMENT_CREATE_FIELDS) attachment_type = _enum( payload["attachmentType"], AttachmentType, "attachmentType" ) security = SecurityLevel.PUBLIC.value status = DocumentStatus.PUBLISHED.value else: raw_type = payload.get("documentType") # 阶段11:配套方案并入共享资料库(0020),停止新建SUB_PLAN文档; # 历史子案已迁移为ACCOMPANYING_PLAN附件,编辑/挂载走附件链路。 if raw_type == DocumentType.SUB_PLAN.value: raise InvalidArgumentError( "配套方案已并入共享资料库,请以附件类型ACCOMPANYING_PLAN上传" ) try: document_type = DocumentType(raw_type).value except (ValueError, TypeError) as exc: raise InvalidArgumentError("documentType必须是MAIN") from exc if document_type not in { DocumentType.MAIN.value, }: raise InvalidArgumentError("documentType必须是MAIN") 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") 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("父文档必须是有效主案") 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"]) extraction = _extract_file_content( upload.temporary_path, upload.extension, upload.file_hash ) final_created = False committed = False old_file_path = None try: with transaction() as session: organization_id, organization_name = _resolve_organization( session, organization_id_input, context ) if not attachment: # 上传权限(契约1.31):主案上传按(密级×部门范围)方案权限 # 矩阵 can_upload 校验(超管恒可;归属他部门按 OTHER 行恒拒); # 共享附件挂载走挂载权限位,不受此限 if not can_upload_plan(context, security, organization_id): raise ForbiddenError("无该密级方案的上传权限") 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) .options(*LOCK_DEFER_LOB) .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.organization_id = organization_id document.organization_name = organization_name 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_extract_error = extraction.error 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: # 新建文档追加到同范围排序末尾:主案全局排序、子方案在所属主案内排序;共享附件不参与。 if document_type == DocumentType.MAIN.value: next_sort_no = (session.scalar( select(func.max(Document.sort_no)).where( Document.document_type == DocumentType.MAIN.value, Document.parent_document_id.is_(None), Document.is_deleted.is_(False), ) ) or 0) + 10 elif document_type == DocumentType.SUB_PLAN.value: next_sort_no = (session.scalar( select(func.max(Document.sort_no)).where( Document.parent_document_id == parent.id, Document.document_type == DocumentType.SUB_PLAN.value, Document.is_deleted.is_(False), ) ) or 0) + 10 else: next_sort_no = 0 document = Document( document_name=name, summary=summary, document_type=document_type, document_status=status, security_level=security, 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, organization_id=organization_id, organization_name=organization_name, 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, sort_no=next_sort_no, content_text=extraction.text, content_extract_status=extraction.status, content_extract_error=extraction.error, 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 password is not None: # 上传主案时同步启用密码锁(覆盖上传已在入口拒绝携带密码字段) apply_password_lock(document, password, protected_actions or []) session.add( business_audit( action=AuditAction.DOCUMENT_PASSWORD_SET, target=AuditTarget.DOCUMENT, target_id=document.id, target_name=document.document_name, detail={ "onUpload": True, "protectedActions": protected_actions or [], }, ) ) 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}, ) REORDER_FIELDS = {"orderedIds"} def _reorder_payload(payload: Any) -> list[int]: if not isinstance(payload, dict) or set(payload) != REORDER_FIELDS: raise InvalidArgumentError("请求字段必须且只能包含orderedIds") raw_ids = payload["orderedIds"] if not isinstance(raw_ids, list) or len(raw_ids) < 2: raise InvalidArgumentError("orderedIds必须是至少包含2个ID的数组") ordered_ids: list[int] = [] seen: set[int] = set() for index, raw_id in enumerate(raw_ids): document_id = _string_id(raw_id, f"orderedIds[{index}]") if document_id not in seen: ordered_ids.append(document_id) seen.add(document_id) return ordered_ids def reorder_documents(payload: Any) -> dict[str, int]: """管理员拖动重排方案:提交ID确定排序范围,槽位整体重排并按步长10重编号。 - 主案在全局范围排序,子方案在所属主案内排序,共享附件不支持; - ``orderedIds`` 为拖动后的目标顺序(允许只提交范围内的部分ID, 未提交的槽位保持相对顺序,仅被提交ID"置换"填充)。 """ ordered_ids = _reorder_payload(payload) context = get_auth_context() def operation() -> dict[str, int]: with transaction() as session: submitted = session.scalars( select(Document) .where(Document.id.in_(sorted(ordered_ids))) .order_by(Document.id.asc()) .options(*LOCK_DEFER_LOB) .with_for_update() ).all() by_id = {document.id: document for document in submitted} if any( document_id not in by_id or by_id[document_id].is_deleted for document_id in ordered_ids ): raise ResourceNotFoundError("待排序方案不存在或已删除") # 重排=改排序字段:对每个提交文档要求EDIT(密级+矩阵/ACL同源校验) for document in submitted: require_plan_action(document, context, AllowedAction.EDIT) first = by_id[ordered_ids[0]] document_type = first.document_type if document_type == DocumentType.MAIN.value: scope = ( Document.document_type == DocumentType.MAIN.value, Document.parent_document_id.is_(None), ) scope_key = "GLOBAL" elif document_type == DocumentType.SUB_PLAN.value: scope = ( Document.document_type == DocumentType.SUB_PLAN.value, Document.parent_document_id == first.parent_document_id, ) scope_key = serialize_id(first.parent_document_id) else: raise InvalidArgumentError("共享附件不支持拖动排序") if any( by_id[document_id].document_type != document_type for document_id in ordered_ids ): raise InvalidArgumentError("orderedIds包含不同范围的方案") siblings = session.scalars( select(Document) .where(Document.is_deleted.is_(False), *scope) .order_by(Document.sort_no.asc(), Document.id.asc()) .options(*LOCK_DEFER_LOB) .with_for_update() ).all() all_ids = [document.id for document in siblings] sibling_ids = set(all_ids) if any(document_id not in sibling_ids for document_id in ordered_ids): raise ResourceNotFoundError("待排序方案不存在或已删除") slot_ids = sorted(ordered_ids, key=all_ids.index) new_order = list(all_ids) for position, document_id in zip( (all_ids.index(did) for did in slot_ids), ordered_ids ): new_order[position] = document_id target_sort_by_id = { document_id: (index + 1) * 10 for index, document_id in enumerate(new_order) } now = _now() reordered_count = 0 for document in siblings: target = target_sort_by_id[document.id] if document.sort_no == target: continue document.sort_no = target document.updated_by = context.user_id document.updated_by_name = context.real_name document.updated_at = now document.row_version += 1 reordered_count += 1 session.add( business_audit( action=AuditAction.REORDER_DOCUMENT, target=AuditTarget.DOCUMENT, target_id=first.id, target_name=first.document_name, detail={ "documentType": document_type, "scope": scope_key, "orderedIds": [serialize_id(did) for did in ordered_ids], "reorderedCount": reordered_count, }, ) ) return {"reorderedCount": reordered_count} return execute_with_deadlock_retry(operation) MOVE_TARGET_VALUES = {"TOP", "BOTTOM"} def _move_target_position( payload: Any, *, extra_fields: set[str] = frozenset(), required_fields: set[str] = frozenset(), ) -> tuple[int | None, str | None]: """解析跨页移动请求:targetPosition/target 二选一,返回(数值位置, 枚举目标)。""" if not isinstance(payload, dict): raise InvalidArgumentError("请求体必须是JSON对象") allowed = {"targetPosition", "target"} | extra_fields | required_fields unknown = set(payload) - allowed if unknown: raise InvalidArgumentError( "请求字段只允许targetPosition、target" + "".join( f"、{field}" for field in sorted(extra_fields | required_fields) ) ) for field in sorted(required_fields): if field not in payload: raise InvalidArgumentError(f"请求必须包含{field}") if ("targetPosition" in payload) == ("target" in payload): raise InvalidArgumentError("targetPosition与target必须二选一") if "targetPosition" in payload: raw = payload["targetPosition"] if type(raw) is not int or raw < 1: raise InvalidArgumentError("targetPosition必须是大于等于1的整数") return raw, None target = payload["target"] if target not in MOVE_TARGET_VALUES: raise InvalidArgumentError("target只允许TOP或BOTTOM") return None, target def move_document(document_id: int, payload: Any) -> dict[str, int]: """管理员跨页调整方案顺序:移动到浏览范围内第 N 位(或置顶/末尾)。 - 范围与拖动重排一致:主案全局、子方案在所属主案内,共享附件不支持; - ``scopeCategoryId`` 仅主案生效:按分类子树过滤可见范围(与列表页口径 一致),位置为范围内 1-based 序号,范围外方案保持原相对顺序; - 位置计算在 ``FOR UPDATE`` 锁内快照上完成,随后按步长10重编号并仅写 变化行,与拖动重排共享同一套并发控制。 """ position_input, target_input, scope_category_id = _move_document_payload(payload) context = get_auth_context() def operation() -> dict[str, int]: 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), ) .options(*LOCK_DEFER_LOB) .with_for_update() ) if document is None: raise ResourceNotFoundError("方案文档不存在或类型不匹配") require_plan_action(document, context, AllowedAction.EDIT) if document.document_type == DocumentType.MAIN.value: scope = ( Document.document_type == DocumentType.MAIN.value, Document.parent_document_id.is_(None), ) scope_key = "GLOBAL" else: scope = ( Document.document_type == DocumentType.SUB_PLAN.value, Document.parent_document_id == document.parent_document_id, ) scope_key = serialize_id(document.parent_document_id) siblings = session.scalars( select(Document) .where(Document.is_deleted.is_(False), *scope) .order_by(Document.sort_no.asc(), Document.id.asc()) .options(*LOCK_DEFER_LOB) .with_for_update() ).all() if not any(item.id == document.id for item in siblings): raise ResourceNotFoundError("待排序方案不存在或已删除") if scope_category_id is not None: if document.document_type != DocumentType.MAIN.value: raise InvalidArgumentError("scopeCategoryId仅支持主案范围") category_ids = _category_scope(scope_category_id, True) visible = [ item for item in siblings if item.category_id in category_ids ] else: visible = list(siblings) if not any(item.id == document.id for item in visible): raise DocumentNotInMoveScopeError("方案不在指定分类范围内") current = next( index for index, item in enumerate(visible) if item.id == document.id ) + 1 total = len(visible) if target_input == "TOP": position = 1 elif target_input == "BOTTOM": position = total else: position = max(1, min(position_input, total)) if position == current: return {"position": position, "total": total} # 可见子序列内重排、范围外方案原位保持:可见槽位按新序列回填。 rest = [item for item in visible if item.id != document.id] insert_at = position - 1 new_visible = rest[:insert_at] + [document] + rest[insert_at:] visible_ids = {item.id for item in visible} fill = iter(new_visible) new_order = [ next(fill) if item.id in visible_ids else item for item in siblings ] target_sort_by_id = { item.id: (index + 1) * 10 for index, item in enumerate(new_order) } now = _now() moved_count = 0 for item in siblings: target_no = target_sort_by_id[item.id] if item.sort_no == target_no: continue item.sort_no = target_no item.updated_by = context.user_id item.updated_by_name = context.real_name item.updated_at = now item.row_version += 1 moved_count += 1 detail: dict[str, Any] = { "documentType": document.document_type, "scope": scope_key, } if scope_category_id is not None: detail["scopeCategoryId"] = serialize_id(scope_category_id) detail.update( { "fromPosition": current, "toPosition": position, "scopeTotal": total, "movedCount": moved_count, } ) session.add( business_audit( action=AuditAction.MOVE_DOCUMENT, target=AuditTarget.DOCUMENT, target_id=document.id, target_name=document.document_name, detail=detail, ) ) return {"position": position, "total": total} return execute_with_deadlock_retry(operation) def _move_document_payload(payload: Any) -> tuple[int | None, str | None, int | None]: """文档移动请求解析:targetPosition/target 二选一 + 可选 scopeCategoryId。""" position, target = _move_target_position( payload, extra_fields={"scopeCategoryId"} ) scope_category_id: int | None = None if isinstance(payload, dict) and "scopeCategoryId" in payload: scope_category_id = _string_id( payload["scopeCategoryId"], "scopeCategoryId" ) return position, target, scope_category_id def update_document(document_id: int, payload: dict[str, Any]): context = get_auth_context() organization_id_input: int | None = None if "organizationId" in payload: organization_id_input = _string_id( payload.pop("organizationId"), "organizationId" ) 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), ) .options(*LOCK_DEFER_LOB) .with_for_update() ) if document is None: raise ResourceNotFoundError("方案文档不存在或类型不匹配") require_plan_action(document, context, AllowedAction.EDIT) require_document_password( document, DocumentPasswordAction.EDIT.value, get_request_password(), context, ) _exact( payload, MAIN_EDIT_FIELDS if document.document_type == DocumentType.MAIN.value else SUB_PLAN_EDIT_FIELDS, ) _check_version(document, _version(payload["rowVersion"])) organization_id, organization_name = _resolve_organization( session, organization_id_input, context, current=(document.organization_id, document.organization_name), ) 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), "organizationId": serialize_id(document.organization_id), "securityLevel": document.security_level, "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.organization_id = organization_id document.organization_name = organization_name document.security_level = _enum( payload["securityLevel"], SecurityLevel, "securityLevel" ) 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_document_description(document_id: int, payload: dict[str, Any]): """维护文档详细说明("查看"对话框右侧备注区,主案/子案/附件通用)。""" context = get_auth_context() _exact(payload, DESCRIPTION_EDIT_FIELDS) description = _text( payload["description"], "description", 20000, nullable=True ) with transaction() as session: document = session.scalar( select(Document) .where( Document.id == document_id, Document.is_deleted.is_(False), ) .options(*LOCK_DEFER_LOB) .with_for_update() ) if document is None: raise ResourceNotFoundError("文档不存在或已删除") require_plan_action(document, context, AllowedAction.EDIT) require_document_password( document, DocumentPasswordAction.EDIT.value, get_request_password(), context, ) before = {"description": document.description} document.description = description 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.DOCUMENT, target_id=document.id, target_name=document.document_name, detail={"field": "description", "before": before, "rowVersion": document.row_version}, ) ) return _serialize(document) def update_attachment(document_id: int, payload: dict[str, Any]): context = get_auth_context() organization_id_input: int | None = None if "organizationId" in payload: organization_id_input = _string_id( payload.pop("organizationId"), "organizationId" ) 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), ) .options(*LOCK_DEFER_LOB) .with_for_update() ) if document is None: raise ResourceNotFoundError("共享附件不存在或类型不匹配") require_plan_action(document, context, AllowedAction.EDIT) _exact(payload, ATTACHMENT_EDIT_FIELDS) _check_version(document, _version(payload["rowVersion"])) organization_id, organization_name = _resolve_organization( session, organization_id_input, context, current=(document.organization_id, document.organization_name), ) before = { "documentName": document.document_name, "attachmentType": document.attachment_type, "organizationId": serialize_id(document.organization_id), } 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.organization_id = organization_id document.organization_name = organization_name 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 re_extract_content(document_id: int): """管理员对存量文档重新提取正文:按当前存储文件重新解析并回写提取结果。 主案/子案/共享附件通用;提取结果(含失败原因)覆盖写回, search_text 同步重建,审计复用 EDIT_DOCUMENT(scope=reExtractContent)。 """ context = get_auth_context() with transaction() as session: document = session.scalar( select(Document) .where( Document.id == document_id, Document.is_deleted.is_(False), ) .options(*LOCK_DEFER_LOB) .with_for_update() ) if document is None: raise ResourceNotFoundError("文档不存在") try: file_path = resolve_storage_path( document.file_relative_path, current_app.config["DMS_STORAGE_ROOT"], ) except UnsafeStoragePathError as exc: raise FilePathInvalidError() from exc if not file_path.is_file(): raise DmsFileNotFoundError() extraction = _extract_file_content( file_path, document.file_extension, document.file_hash ) document.content_text = extraction.text document.content_extract_status = extraction.status document.content_extract_error = extraction.error document.content_extracted_at = extraction.extracted_at document.search_text = build_search_text( document.document_name, document.summary, document.tags, extraction.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 if document.document_type == DocumentType.ATTACHMENT.value else AuditTarget.DOCUMENT ), target_id=document.id, target_name=document.document_name, detail={ "scope": "reExtractContent", "contentExtractStatus": extraction.status, "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), ) .options(*LOCK_DEFER_LOB) .with_for_update() ) if document is None: raise ResourceNotFoundError("方案文档不存在或类型不匹配") require_plan_action(document, context, AllowedAction.DELETE) require_document_password( document, DocumentPasswordAction.DELETE.value, get_request_password(), context, ) _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), ) .options(*LOCK_DEFER_LOB) .with_for_update() ) if document is None: raise ResourceNotFoundError("共享附件不存在或类型不匹配") require_plan_action(document, context, AllowedAction.DELETE) _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}