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

"""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,
    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 (
    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
            )
            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}