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

"""B5文档与共享附件写操作及文件事务补偿。"""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any

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,
    ConflictError,
    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.storage.uploads import StagedUpload, stage_upload

PLAN_CREATE_FIELDS = {
    "documentName",
    "documentType",
    "summary",
    "categoryId",
    "securityLevel",
    "visibilityType",
    "status",
    "tags",
}
PLAN_EDIT_FIELDS = {
    "documentName",
    "summary",
    "categoryId",
    "securityLevel",
    "visibilityType",
    "status",
    "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 _search(name: str, summary: str | None, tags: list[str]) -> str:
    return " ".join([name, summary or "", *tags]).strip()


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")
        fields = PLAN_CREATE_FIELDS | (
            {"parentDocumentId"}
            if document_type == DocumentType.SUB_PLAN.value
            else set()
        )
        _exact(payload, fields)
        category = _category(_string_id(payload["categoryId"], "categoryId"))
        security = _enum(payload["securityLevel"], SecurityLevel, "securityLevel")
        visibility = _enum(
            payload["visibilityType"], VisibilityType, "visibilityType"
        )
        status = _enum(payload["status"], DocumentStatus, "status")
        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

    name = _text(payload["documentName"], "documentName", 255)
    summary = _text(payload["summary"], "summary", 20000, nullable=True)
    tags = _tags(payload["tags"])
    final_created = False
    committed = False
    try:
        with transaction() as session:
            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,
                search_text=_search(name, summary, tags),
                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:
                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
    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]):
    _exact(payload, PLAN_EDIT_FIELDS)
    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("方案文档不存在或类型不匹配")
        _check_version(document, _version(payload["rowVersion"]))
        category = _category(_string_id(payload["categoryId"], "categoryId"))
        before = {
            "documentName": document.document_name,
            "categoryId": serialize_id(document.category_id),
            "securityLevel": document.security_level,
            "visibilityType": document.visibility_type,
            "status": document.document_status,
        }
        if document.category_id != category.id:
            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"
        )
        requested_visibility = _enum(
            payload["visibilityType"], VisibilityType, "visibilityType"
        )
        if document.document_type == DocumentType.MAIN.value:
            document.visibility_type = requested_visibility
        document.document_status = _enum(
            payload["status"], DocumentStatus, "status"
        )
        document.search_text = _search(
            document.document_name, document.summary, document.tags
        )
        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={"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 = _search(
            document.document_name, document.summary, document.tags
        )
        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:
            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}