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

"""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}