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

"""B6共享附件挂载和解除挂载事务。"""

from __future__ import annotations

from datetime import datetime, timezone
from typing import Any

from sqlalchemy import func, select
from sqlalchemy.exc import IntegrityError

from dms.common.enums import (
    AttachmentType,
    AuditAction,
    AuditTarget,
    DocumentType,
)
from dms.common.errors import (
    ConflictError,
    DocumentNotInMoveScopeError,
    InvalidArgumentError,
    ResourceNotFoundError,
)
from dms.common.response import serialize_id
from dms.database.transaction import execute_with_deadlock_retry, transaction
from dms.extensions import db
from dms.models import LOCK_DEFER_LOB, AttachmentBinding, Document
from dms.security.auth_context import get_auth_context
from dms.services.audit_service import business_audit
from dms.common.enums import AllowedAction
from dms.services.authorization_service import (
    require_plan_action,
    require_security_clearance,
)
from dms.services.document_password_service import (
    DocumentPasswordAction,
    get_request_password,
    require_document_password,
)

BIND_FIELDS = {"attachmentIds", "mainPlanRowVersion"}


def _now() -> datetime:
    return datetime.now(timezone.utc).replace(tzinfo=None)


def _version(value: Any, name: str) -> int:
    if type(value) is not int or value < 0:
        raise InvalidArgumentError(f"{name}必须是非负整数")
    return value


def _string_id(value: Any, name: str) -> int:
    if not isinstance(value, str) or not value.isdecimal() or int(value) <= 0:
        raise InvalidArgumentError(f"{name}必须是正整数形式的字符串ID")
    return int(value)


def _bind_payload(payload: Any) -> tuple[list[int], int]:
    if not isinstance(payload, dict) or set(payload) != BIND_FIELDS:
        raise InvalidArgumentError(
            "请求字段必须且只能包含attachmentIds和mainPlanRowVersion"
        )
    raw_ids = payload["attachmentIds"]
    if not isinstance(raw_ids, list) or not raw_ids:
        raise InvalidArgumentError("attachmentIds必须是非空数组")
    attachment_ids: list[int] = []
    seen: set[int] = set()
    for index, raw_id in enumerate(raw_ids):
        attachment_id = _string_id(raw_id, f"attachmentIds[{index}]")
        if attachment_id not in seen:
            attachment_ids.append(attachment_id)
            seen.add(attachment_id)
    return attachment_ids, _version(
        payload["mainPlanRowVersion"], "mainPlanRowVersion"
    )


def _locked_main(session, main_document_id: int) -> Document:
    main = session.scalar(
        select(Document)
        .where(
            Document.id == main_document_id,
            Document.is_deleted.is_(False),
        )
        .options(*LOCK_DEFER_LOB)
        .with_for_update()
    )
    if main is None or main.document_type != DocumentType.MAIN.value:
        raise ResourceNotFoundError("主案不存在或类型不匹配")
    require_security_clearance(main, get_auth_context())
    return main


def _check_version(main: Document, expected: int) -> None:
    if main.row_version != expected:
        raise ConflictError(
            "数据已被其他用户修改，请刷新后重试",
            details={"currentRowVersion": main.row_version},
        )


def _locked_attachments(session, attachment_ids: list[int]) -> dict[int, Document]:
    documents = session.scalars(
        select(Document)
        .where(Document.id.in_(sorted(attachment_ids)))
        .order_by(Document.id.asc())
        .options(*LOCK_DEFER_LOB)
        .with_for_update()
    ).all()
    by_id = {document.id: document for document in documents}
    if any(
        attachment_id not in by_id
        or by_id[attachment_id].is_deleted
        or by_id[attachment_id].document_type != DocumentType.ATTACHMENT.value
        for attachment_id in attachment_ids
    ):
        raise ResourceNotFoundError("共享附件不存在或类型不匹配")
    return by_id


def _active_count(session, main_document_id: int) -> int:
    return (
        session.scalar(
            select(func.count(AttachmentBinding.id)).where(
                AttachmentBinding.main_document_id == main_document_id,
                AttachmentBinding.is_deleted.is_(False),
            )
        )
        or 0
    )


def _current_version(main_document_id: int) -> int | None:
    return db.session.scalar(
        select(Document.row_version).where(
            Document.id == main_document_id,
            Document.is_deleted.is_(False),
        )
    )


def bind_attachments(main_document_id: int, payload: Any) -> dict[str, int]:
    """去重后批量挂载；同一主案上的写入由主案行锁串行化。"""
    attachment_ids, expected_version = _bind_payload(payload)
    context = get_auth_context()

    def operation() -> dict[str, int]:
        with transaction() as session:
            main = _locked_main(session, main_document_id)
            require_plan_action(main, context, AllowedAction.BIND_ATTACHMENT)
            require_document_password(
                main,
                DocumentPasswordAction.BIND_ATTACHMENT.value,
                get_request_password(),
                context,
            )
            _check_version(main, expected_version)
            _locked_attachments(session, attachment_ids)

            bindings = session.scalars(
                select(AttachmentBinding)
                .where(AttachmentBinding.main_document_id == main.id)
                .order_by(AttachmentBinding.id.asc())
                .with_for_update()
            ).all()
            active_by_attachment = {
                binding.attachment_document_id: binding
                for binding in bindings
                if not binding.is_deleted
            }
            existing_ids = [
                attachment_id
                for attachment_id in attachment_ids
                if attachment_id in active_by_attachment
            ]
            created_ids = [
                attachment_id
                for attachment_id in attachment_ids
                if attachment_id not in active_by_attachment
            ]
            if not created_ids:
                return {
                    "createdCount": 0,
                    "existingCount": len(existing_ids),
                    "attachmentCount": _active_count(session, main.id),
                    "mainPlanRowVersion": main.row_version,
                }

            max_sort_no = max(
                (
                    binding.sort_no
                    for binding in bindings
                    if not binding.is_deleted
                ),
                default=0,
            )
            for offset, attachment_id in enumerate(created_ids, start=1):
                session.add(
                    AttachmentBinding(
                        main_document_id=main.id,
                        attachment_document_id=attachment_id,
                        sort_no=max_sort_no + offset * 10,
                        created_by=context.user_id,
                        updated_by=context.user_id,
                    )
                )
            session.flush()
            main.attachment_count = _active_count(session, main.id)
            main.row_version += 1
            main.updated_by = context.user_id
            main.updated_by_name = context.real_name
            main.updated_at = _now()
            session.add(
                business_audit(
                    action=AuditAction.BIND_ATTACHMENT,
                    target=AuditTarget.ATTACHMENT_BINDING,
                    target_id=main.id,
                    target_name=main.document_name,
                    detail={
                        "mainPlanId": serialize_id(main.id),
                        "requestedAttachmentIds": [
                            serialize_id(value) for value in attachment_ids
                        ],
                        "createdAttachmentIds": [
                            serialize_id(value) for value in created_ids
                        ],
                        "existingAttachmentIds": [
                            serialize_id(value) for value in existing_ids
                        ],
                        "createdCount": len(created_ids),
                        "existingCount": len(existing_ids),
                        "attachmentCount": main.attachment_count,
                        "mainPlanRowVersion": main.row_version,
                    },
                )
            )
            return {
                "createdCount": len(created_ids),
                "existingCount": len(existing_ids),
                "attachmentCount": main.attachment_count,
                "mainPlanRowVersion": main.row_version,
            }

    try:
        return execute_with_deadlock_retry(operation)
    except IntegrityError as exc:
        db.session.rollback()
        raise ConflictError(
            "挂载关系已被并发修改，请刷新后重试",
            details={"currentRowVersion": _current_version(main_document_id)},
        ) from exc


def unbind_attachment(
    main_document_id: int,
    attachment_document_id: int,
    expected_version: int,
) -> dict[str, int]:
    """逻辑删除一条有效挂载关系并重新计算主案计数。"""
    expected_version = _version(expected_version, "mainPlanRowVersion")
    context = get_auth_context()

    def operation() -> dict[str, int]:
        with transaction() as session:
            main = _locked_main(session, main_document_id)
            require_plan_action(main, context, AllowedAction.UNBIND_ATTACHMENT)
            require_document_password(
                main,
                DocumentPasswordAction.UNBIND_ATTACHMENT.value,
                get_request_password(),
                context,
            )
            _check_version(main, expected_version)
            _locked_attachments(session, [attachment_document_id])
            binding = session.scalar(
                select(AttachmentBinding)
                .where(
                    AttachmentBinding.main_document_id == main.id,
                    AttachmentBinding.attachment_document_id
                    == attachment_document_id,
                    AttachmentBinding.is_deleted.is_(False),
                )
                .with_for_update()
            )
            if binding is None:
                raise ResourceNotFoundError("有效挂载关系不存在")

            now = _now()
            binding.is_deleted = True
            binding.deleted_at = now
            binding.updated_at = now
            binding.updated_by = context.user_id
            binding.row_version += 1
            session.flush()
            main.attachment_count = _active_count(session, main.id)
            main.row_version += 1
            main.updated_by = context.user_id
            main.updated_by_name = context.real_name
            main.updated_at = now
            session.add(
                business_audit(
                    action=AuditAction.UNBIND_ATTACHMENT,
                    target=AuditTarget.ATTACHMENT_BINDING,
                    target_id=binding.id,
                    target_name=main.document_name,
                    detail={
                        "bindingId": serialize_id(binding.id),
                        "mainPlanId": serialize_id(main.id),
                        "attachmentId": serialize_id(attachment_document_id),
                        "attachmentCount": main.attachment_count,
                        "mainPlanRowVersion": main.row_version,
                    },
                )
            )
            return {
                "attachmentCount": main.attachment_count,
                "mainPlanRowVersion": main.row_version,
            }

    return execute_with_deadlock_retry(operation)


REORDER_BINDING_FIELDS = {"orderedBindingIds", "mainPlanRowVersion"}


def _reorder_bindings_payload(payload: Any) -> tuple[list[int], int]:
    if not isinstance(payload, dict) or set(payload) != REORDER_BINDING_FIELDS:
        raise InvalidArgumentError(
            "请求字段必须且只能包含orderedBindingIds和mainPlanRowVersion"
        )
    raw_ids = payload["orderedBindingIds"]
    if not isinstance(raw_ids, list) or len(raw_ids) < 2:
        raise InvalidArgumentError("orderedBindingIds必须是至少包含2个ID的数组")
    binding_ids: list[int] = []
    seen: set[int] = set()
    for index, raw_id in enumerate(raw_ids):
        binding_id = _string_id(raw_id, f"orderedBindingIds[{index}]")
        if binding_id not in seen:
            binding_ids.append(binding_id)
            seen.add(binding_id)
    return binding_ids, _version(
        payload["mainPlanRowVersion"], "mainPlanRowVersion"
    )


def reorder_main_plan_attachments(
    main_document_id: int,
    payload: Any,
) -> dict[str, int]:
    """调整主案下挂载附件的显示顺序；槽位重排语义与方案重排一致。

    不同主案可对同一附件给出各自独立的顺序（顺序存于挂载关系而非附件本身）。
    """
    ordered_ids, expected_version = _reorder_bindings_payload(payload)
    context = get_auth_context()

    def operation() -> dict[str, int]:
        with transaction() as session:
            main = _locked_main(session, main_document_id)
            # 挂载重排=调整主案挂载区顺序：按主案 EDIT 判定（移动=编辑，契约1.18修订）
            require_plan_action(main, context, AllowedAction.EDIT)
            _check_version(main, expected_version)
            bindings = session.scalars(
                select(AttachmentBinding)
                .where(
                    AttachmentBinding.main_document_id == main.id,
                    AttachmentBinding.is_deleted.is_(False),
                )
                .order_by(
                    AttachmentBinding.sort_no.asc(), AttachmentBinding.id.asc()
                )
                .with_for_update()
            ).all()
            all_ids = [binding.id for binding in bindings]
            active_ids = set(all_ids)
            if any(binding_id not in active_ids for binding_id in ordered_ids):
                raise ResourceNotFoundError("挂载关系不存在或已解除")
            slot_ids = sorted(ordered_ids, key=all_ids.index)
            new_order = list(all_ids)
            for position, binding_id in zip(
                (all_ids.index(bid) for bid in slot_ids), ordered_ids
            ):
                new_order[position] = binding_id
            target_sort_by_id = {
                binding_id: (index + 1) * 10
                for index, binding_id in enumerate(new_order)
            }
            now = _now()
            reordered_count = 0
            for binding in bindings:
                target = target_sort_by_id[binding.id]
                if binding.sort_no == target:
                    continue
                binding.sort_no = target
                binding.updated_at = now
                binding.updated_by = context.user_id
                binding.row_version += 1
                reordered_count += 1
            if reordered_count:
                main.row_version += 1
                main.updated_by = context.user_id
                main.updated_by_name = context.real_name
                main.updated_at = now
            session.add(
                business_audit(
                    action=AuditAction.REORDER_ATTACHMENT,
                    target=AuditTarget.ATTACHMENT_BINDING,
                    target_id=main.id,
                    target_name=main.document_name,
                    detail={
                        "mainPlanId": serialize_id(main.id),
                        "orderedBindingIds": [
                            serialize_id(bid) for bid in ordered_ids
                        ],
                        "reorderedCount": reordered_count,
                        "mainPlanRowVersion": main.row_version,
                    },
                )
            )
            return {
                "reorderedCount": reordered_count,
                "mainPlanRowVersion": main.row_version,
            }

    return execute_with_deadlock_retry(operation)


def _move_binding_payload(
    payload: Any,
) -> tuple[int | None, str | None, str | None, int]:
    """挂载附件移动请求解析：targetPosition/target 二选一 +
    必填 mainPlanRowVersion + 可选 scopeAttachmentType（页签类型）。
    """
    if not isinstance(payload, dict):
        raise InvalidArgumentError("请求体必须是JSON对象")
    allowed = {
        "targetPosition",
        "target",
        "scopeAttachmentType",
        "mainPlanRowVersion",
    }
    if set(payload) - allowed:
        raise InvalidArgumentError(
            "请求字段只允许targetPosition、target、scopeAttachmentType或mainPlanRowVersion"
        )
    if "mainPlanRowVersion" not in payload:
        raise InvalidArgumentError("请求必须包含mainPlanRowVersion")
    if ("targetPosition" in payload) == ("target" in payload):
        raise InvalidArgumentError("targetPosition与target必须二选一")
    if "targetPosition" in payload:
        raw_position = payload["targetPosition"]
        if type(raw_position) is not int or raw_position < 1:
            raise InvalidArgumentError("targetPosition必须是大于等于1的整数")
        position: int | None = raw_position
        target: str | None = None
    else:
        target = payload["target"]
        if target not in {"TOP", "BOTTOM"}:
            raise InvalidArgumentError("target只允许TOP或BOTTOM")
        position = None
    scope_attachment_type: str | None = None
    if "scopeAttachmentType" in payload:
        raw_type = payload["scopeAttachmentType"]
        try:
            scope_attachment_type = AttachmentType(raw_type).value
        except ValueError:
            raise InvalidArgumentError("scopeAttachmentType不是有效附件类型")
    return (
        position,
        target,
        scope_attachment_type,
        _version(payload["mainPlanRowVersion"], "mainPlanRowVersion"),
    )


def move_main_plan_attachment(
    main_document_id: int,
    binding_id: int,
    payload: Any,
) -> dict[str, int]:
    """跨页调整主案下挂载附件顺序：移动到页签类型范围内指定位置/置顶/末尾。

    - ``scopeAttachmentType`` 给定时位置口径为该类型挂载的 1-based 序号
      （与列表页签过滤口径一致），其他类型挂载保持原相对顺序；
    - 槽位移动语义与挂载重排一致，顺序存于挂载关系（各主案独立）。
    """
    position_input, target_input, scope_type, expected_version = (
        _move_binding_payload(payload)
    )
    context = get_auth_context()

    def operation() -> dict[str, int]:
        with transaction() as session:
            main = _locked_main(session, main_document_id)
            # 挂载移动与重排同权限口径：按主案 EDIT 判定（移动=编辑，契约1.18修订）
            require_plan_action(main, context, AllowedAction.EDIT)
            _check_version(main, expected_version)
            bindings = session.scalars(
                select(AttachmentBinding)
                .where(
                    AttachmentBinding.main_document_id == main.id,
                    AttachmentBinding.is_deleted.is_(False),
                )
                .order_by(
                    AttachmentBinding.sort_no.asc(), AttachmentBinding.id.asc()
                )
                .with_for_update()
            ).all()
            moved = next(
                (item for item in bindings if item.id == binding_id), None
            )
            if moved is None:
                raise ResourceNotFoundError("挂载关系不存在或已解除")
            if scope_type is not None:
                attachment_ids = [
                    item.attachment_document_id for item in bindings
                ]
                documents = session.scalars(
                    select(Document).where(
                        Document.id.in_(attachment_ids),
                        Document.is_deleted.is_(False),
                    )
                ).all()
                type_by_attachment = {
                    document.id: document.attachment_type
                    for document in documents
                }
                visible = [
                    item
                    for item in bindings
                    if type_by_attachment.get(item.attachment_document_id)
                    == scope_type
                ]
            else:
                visible = list(bindings)
            if not any(item.id == binding_id for item in visible):
                raise DocumentNotInMoveScopeError("挂载附件不在指定类型范围内")
            current = next(
                index for index, item in enumerate(visible)
                if item.id == binding_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,
                    "mainPlanRowVersion": main.row_version,
                }
            # 类型内重排、类型外挂载原位保持：可见槽位按新序列回填。
            rest = [item for item in visible if item.id != binding_id]
            insert_at = position - 1
            new_visible = rest[:insert_at] + [moved] + 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 bindings
            ]
            target_sort_by_id = {
                item.id: (index + 1) * 10
                for index, item in enumerate(new_order)
            }
            now = _now()
            moved_count = 0
            for item in bindings:
                target_no = target_sort_by_id[item.id]
                if item.sort_no == target_no:
                    continue
                item.sort_no = target_no
                item.updated_at = now
                item.updated_by = context.user_id
                item.row_version += 1
                moved_count += 1
            if moved_count:
                main.row_version += 1
                main.updated_by = context.user_id
                main.updated_by_name = context.real_name
                main.updated_at = now
            detail: dict[str, Any] = {
                "mainPlanId": serialize_id(main.id),
                "bindingId": serialize_id(moved.id),
            }
            if scope_type is not None:
                detail["scopeAttachmentType"] = scope_type
            detail.update(
                {
                    "fromPosition": current,
                    "toPosition": position,
                    "scopeTotal": total,
                    "movedCount": moved_count,
                    "mainPlanRowVersion": main.row_version,
                }
            )
            session.add(
                business_audit(
                    action=AuditAction.MOVE_ATTACHMENT,
                    target=AuditTarget.ATTACHMENT_BINDING,
                    target_id=moved.id,
                    target_name=main.document_name,
                    detail=detail,
                )
            )
            return {
                "position": position,
                "total": total,
                "mainPlanRowVersion": main.row_version,
            }

    return execute_with_deadlock_retry(operation)


__all__ = [
    "bind_attachments",
    "move_main_plan_attachment",
    "reorder_main_plan_attachments",
    "unbind_attachment",
]