"""B6共享附件挂载和解除挂载事务。""" from __future__ import annotations from datetime import datetime, timezone from typing import Any from sqlalchemy import func, select from sqlalchemy.exc import IntegrityError from dms.common.enums import AuditAction, AuditTarget, DocumentType from dms.common.errors import ( ConflictError, InvalidArgumentError, 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, Document from dms.security.auth_context import get_auth_context from dms.services.audit_service import business_audit from dms.services.authorization_service import require_security_clearance BIND_FIELDS = {"attachmentIds", "mainPlanRowVersion"} def _now() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) def _version(value: Any, name: str) -> int: if type(value) is not int or value < 0: raise InvalidArgumentError(f"{name}必须是非负整数") return value 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 _bind_payload(payload: Any) -> tuple[list[int], int]: if not isinstance(payload, dict) or set(payload) != BIND_FIELDS: raise InvalidArgumentError( "请求字段必须且只能包含attachmentIds和mainPlanRowVersion" ) raw_ids = payload["attachmentIds"] if not isinstance(raw_ids, list) or not raw_ids: raise InvalidArgumentError("attachmentIds必须是非空数组") attachment_ids: list[int] = [] seen: set[int] = set() for index, raw_id in enumerate(raw_ids): attachment_id = _string_id(raw_id, f"attachmentIds[{index}]") if attachment_id not in seen: attachment_ids.append(attachment_id) seen.add(attachment_id) return attachment_ids, _version( payload["mainPlanRowVersion"], "mainPlanRowVersion" ) def _locked_main(session, main_document_id: int) -> Document: main = session.scalar( select(Document) .where( Document.id == main_document_id, Document.is_deleted.is_(False), ) .with_for_update() ) if main is None or main.document_type != DocumentType.MAIN.value: raise ResourceNotFoundError("主案不存在或类型不匹配") require_security_clearance(main, get_auth_context()) return main def _check_version(main: Document, expected: int) -> None: if main.row_version != expected: raise ConflictError( "数据已被其他用户修改,请刷新后重试", details={"currentRowVersion": main.row_version}, ) def _locked_attachments(session, attachment_ids: list[int]) -> dict[int, Document]: documents = session.scalars( select(Document) .where(Document.id.in_(sorted(attachment_ids))) .order_by(Document.id.asc()) .with_for_update() ).all() by_id = {document.id: document for document in documents} if any( attachment_id not in by_id or by_id[attachment_id].is_deleted or by_id[attachment_id].document_type != DocumentType.ATTACHMENT.value for attachment_id in attachment_ids ): raise ResourceNotFoundError("共享附件不存在或类型不匹配") return by_id def _active_count(session, main_document_id: int) -> int: return ( session.scalar( select(func.count(AttachmentBinding.id)).where( AttachmentBinding.main_document_id == main_document_id, AttachmentBinding.is_deleted.is_(False), ) ) or 0 ) def _current_version(main_document_id: int) -> int | None: return db.session.scalar( select(Document.row_version).where( Document.id == main_document_id, Document.is_deleted.is_(False), ) ) def bind_attachments(main_document_id: int, payload: Any) -> dict[str, int]: """去重后批量挂载;同一主案上的写入由主案行锁串行化。""" attachment_ids, expected_version = _bind_payload(payload) context = get_auth_context() def operation() -> dict[str, int]: with transaction() as session: main = _locked_main(session, main_document_id) _check_version(main, expected_version) _locked_attachments(session, attachment_ids) bindings = session.scalars( select(AttachmentBinding) .where(AttachmentBinding.main_document_id == main.id) .order_by(AttachmentBinding.id.asc()) .with_for_update() ).all() active_by_attachment = { binding.attachment_document_id: binding for binding in bindings if not binding.is_deleted } existing_ids = [ attachment_id for attachment_id in attachment_ids if attachment_id in active_by_attachment ] created_ids = [ attachment_id for attachment_id in attachment_ids if attachment_id not in active_by_attachment ] if not created_ids: return { "createdCount": 0, "existingCount": len(existing_ids), "attachmentCount": _active_count(session, main.id), "mainPlanRowVersion": main.row_version, } max_sort_no = max( ( binding.sort_no for binding in bindings if not binding.is_deleted ), default=0, ) for offset, attachment_id in enumerate(created_ids, start=1): session.add( AttachmentBinding( main_document_id=main.id, attachment_document_id=attachment_id, sort_no=max_sort_no + offset * 10, created_by=context.user_id, updated_by=context.user_id, ) ) session.flush() main.attachment_count = _active_count(session, main.id) main.row_version += 1 main.updated_by = context.user_id main.updated_by_name = context.real_name main.updated_at = _now() session.add( business_audit( action=AuditAction.BIND_ATTACHMENT, target=AuditTarget.ATTACHMENT_BINDING, target_id=main.id, target_name=main.document_name, detail={ "mainPlanId": serialize_id(main.id), "requestedAttachmentIds": [ serialize_id(value) for value in attachment_ids ], "createdAttachmentIds": [ serialize_id(value) for value in created_ids ], "existingAttachmentIds": [ serialize_id(value) for value in existing_ids ], "createdCount": len(created_ids), "existingCount": len(existing_ids), "attachmentCount": main.attachment_count, "mainPlanRowVersion": main.row_version, }, ) ) return { "createdCount": len(created_ids), "existingCount": len(existing_ids), "attachmentCount": main.attachment_count, "mainPlanRowVersion": main.row_version, } try: return execute_with_deadlock_retry(operation) except IntegrityError as exc: db.session.rollback() raise ConflictError( "挂载关系已被并发修改,请刷新后重试", details={"currentRowVersion": _current_version(main_document_id)}, ) from exc def unbind_attachment( main_document_id: int, attachment_document_id: int, expected_version: int, ) -> dict[str, int]: """逻辑删除一条有效挂载关系并重新计算主案计数。""" expected_version = _version(expected_version, "mainPlanRowVersion") context = get_auth_context() def operation() -> dict[str, int]: with transaction() as session: main = _locked_main(session, main_document_id) _check_version(main, expected_version) _locked_attachments(session, [attachment_document_id]) binding = session.scalar( select(AttachmentBinding) .where( AttachmentBinding.main_document_id == main.id, AttachmentBinding.attachment_document_id == attachment_document_id, AttachmentBinding.is_deleted.is_(False), ) .with_for_update() ) if binding is None: raise ResourceNotFoundError("有效挂载关系不存在") now = _now() binding.is_deleted = True binding.deleted_at = now binding.updated_at = now binding.updated_by = context.user_id binding.row_version += 1 session.flush() main.attachment_count = _active_count(session, main.id) main.row_version += 1 main.updated_by = context.user_id main.updated_by_name = context.real_name main.updated_at = now session.add( business_audit( action=AuditAction.UNBIND_ATTACHMENT, target=AuditTarget.ATTACHMENT_BINDING, target_id=binding.id, target_name=main.document_name, detail={ "bindingId": serialize_id(binding.id), "mainPlanId": serialize_id(main.id), "attachmentId": serialize_id(attachment_document_id), "attachmentCount": main.attachment_count, "mainPlanRowVersion": main.row_version, }, ) ) return { "attachmentCount": main.attachment_count, "mainPlanRowVersion": main.row_version, } return execute_with_deadlock_retry(operation) __all__ = ["bind_attachments", "unbind_attachment"]