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

"""共享附件及其有效挂载关系只读查询。"""

from __future__ import annotations

from typing import Mapping

from sqlalchemy import String, cast, func, or_, select

from dms.common.enums import AttachmentType, DocumentType
from dms.common.errors import ResourceNotFoundError
from dms.common.pagination import page_result
from dms.common.response import serialize_id
from dms.common.time_range import parse_updated_range
from dms.extensions import db
from dms.models import AttachmentBinding, Document
from dms.security.auth_context import AuthContext, get_auth_context
from dms.services.authorization_service import (
    attachment_allowed_actions,
    evaluate_plan_access,
    require_attachment_access,
)
from dms.services.document_query_service import (
    DOCUMENT_SORTS,
    _enum,
    _iso,
    _page,
    _record_view,
    _sort,
    parse_string_id,
)


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


def attachment_summary(
    attachment: Document,
    context: AuthContext,
    *,
    plan_access=None,
) -> dict[str, object]:
    return {
        "id": serialize_id(attachment.id),
        "documentName": attachment.document_name,
        "summary": attachment.summary,
        "documentType": attachment.document_type,
        "status": attachment.document_status,
        "securityLevel": attachment.security_level,
        "visibilityType": attachment.visibility_type,
        "visibilitySummary": "全部已登录用户",
        "attachmentType": attachment.attachment_type,
        "categoryId": None,
        "categoryName": None,
        "categoryPath": None,
        "parentDocumentId": None,
        "rootDocumentId": None,
        "tags": attachment.tags or [],
        "fileExtension": attachment.file_extension,
        "childCount": attachment.child_count,
        "attachmentCount": attachment.attachment_count,
        "viewCount": attachment.view_count,
        "downloadCount": attachment.download_count,
        "createdByName": attachment.created_by_name,
        "createdAt": _iso(attachment.created_at),
        "updatedAt": _iso(attachment.updated_at),
        "rowVersion": attachment.row_version,
        "allowedActions": attachment_allowed_actions(context, plan_access),
        "mountedPlanCount": mounted_plan_count(attachment.id),
    }


def attachment_detail(
    attachment: Document,
    context: AuthContext,
) -> dict[str, object]:
    result = attachment_summary(attachment, context)
    result.update(
        {
            "originalFileName": attachment.original_file_name,
            "mimeType": attachment.mime_type,
            "fileSize": attachment.file_size,
            "fileHash": attachment.file_hash,
            "permissionSummary": {
                "organizationCount": 0,
                "userCount": 0,
                "inheritedFromMainPlan": False,
            },
        }
    )
    return result


def _attachment(attachment_id: int) -> Document:
    attachment = db.session.scalar(
        select(Document).where(
            Document.id == attachment_id,
            Document.document_type == DocumentType.ATTACHMENT.value,
            Document.is_deleted.is_(False),
        )
    )
    if attachment is None:
        raise ResourceNotFoundError("共享附件不存在")
    return attachment


def _filtered_attachments(params: Mapping[str, str]):
    statement = select(Document).where(
        Document.document_type == DocumentType.ATTACHMENT.value,
        Document.is_deleted.is_(False),
    )
    keyword = params.get("keyword")
    if keyword and keyword.strip():
        escaped = keyword.strip().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
        pattern = f"%{escaped}%"
        statement = statement.where(
            or_(
                Document.document_name.like(pattern, escape="\\"),
                Document.summary.like(pattern, escape="\\"),
                Document.search_text.like(pattern, escape="\\"),
                cast(Document.tags, String).like(pattern, escape="\\"),
            )
        )
    attachment_type = _enum(
        params.get("attachmentType"), AttachmentType, "attachmentType"
    )
    if attachment_type:
        statement = statement.where(Document.attachment_type == attachment_type)
    extension = params.get("fileExtension")
    if extension:
        statement = statement.where(
            Document.file_extension == extension.strip().lower().lstrip(".")
        )
    updated_from, updated_to = parse_updated_range(params)
    if updated_from:
        statement = statement.where(Document.updated_at >= updated_from)
    if updated_to:
        statement = statement.where(Document.updated_at <= updated_to)
    order, _ = _sort(params, allowed=DOCUMENT_SORTS)
    return statement.order_by(order, Document.id.asc())


def list_attachments(params: Mapping[str, str]) -> dict[str, object]:
    context = get_auth_context()
    page = _page(params)
    attachments = db.session.scalars(_filtered_attachments(params)).all()
    total = len(attachments)
    selected = attachments[page.offset : page.offset + page.page_size]
    return page_result(
        [attachment_summary(item, context) for item in selected],
        page=page.page,
        page_size=page.page_size,
        total=total,
    )


def get_attachment(attachment_id: int) -> dict[str, object]:
    context = get_auth_context()
    attachment = _attachment(attachment_id)
    can_view, can_download = require_attachment_access(
        attachment, context, download=False
    )
    result = attachment_detail(attachment, context)
    if context.role_code.value == "USER":
        result["allowedActions"] = ["VIEW"] + (
            ["DOWNLOAD"] if can_download else []
        )
    result["viewCount"] = _record_view(attachment)
    return result


def list_attachment_main_plans(attachment_id: int) -> dict[str, object]:
    context = get_auth_context()
    _attachment(attachment_id)
    rows = db.session.execute(
        select(AttachmentBinding, Document)
        .join(Document, Document.id == AttachmentBinding.main_document_id)
        .where(
            AttachmentBinding.attachment_document_id == attachment_id,
            AttachmentBinding.is_deleted.is_(False),
            Document.document_type == DocumentType.MAIN.value,
            Document.is_deleted.is_(False),
        )
        .order_by(Document.document_name.asc(), Document.id.asc())
    ).all()
    items: list[dict[str, object]] = []
    for _, main in rows:
        if not evaluate_plan_access(main, context).allowed:
            continue
        items.append(
            {
                "id": serialize_id(main.id),
                "documentName": main.document_name,
                "categoryName": main.category_name,
                "securityLevel": main.security_level,
            }
        )
    return {"items": items, "total": len(items)}


def list_main_plan_attachments(
    main_document_id: int,
    params: Mapping[str, str],
) -> dict[str, object]:
    from dms.common.errors import MainPlanNotFoundError
    from dms.services.document_query_service import _require_plan_access

    context = get_auth_context()
    main = db.session.scalar(
        select(Document).where(
            Document.id == main_document_id,
            Document.document_type == DocumentType.MAIN.value,
            Document.is_deleted.is_(False),
        )
    )
    if main is None:
        raise MainPlanNotFoundError()
    main_access = _require_plan_access(main, context)
    page = _page(params)
    statement = (
        select(AttachmentBinding, Document)
        .join(Document, Document.id == AttachmentBinding.attachment_document_id)
        .where(
            AttachmentBinding.main_document_id == main.id,
            AttachmentBinding.is_deleted.is_(False),
            Document.document_type == DocumentType.ATTACHMENT.value,
            Document.is_deleted.is_(False),
        )
    )
    keyword = params.get("keyword")
    if keyword and keyword.strip():
        escaped = keyword.strip().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
        pattern = f"%{escaped}%"
        statement = statement.where(
            or_(
                Document.document_name.like(pattern, escape="\\"),
                Document.summary.like(pattern, escape="\\"),
                Document.search_text.like(pattern, escape="\\"),
            )
        )
    attachment_type = _enum(
        params.get("attachmentType"), AttachmentType, "attachmentType"
    )
    if attachment_type:
        statement = statement.where(Document.attachment_type == attachment_type)
    updated_from, updated_to = parse_updated_range(params)
    if updated_from:
        statement = statement.where(Document.updated_at >= updated_from)
    if updated_to:
        statement = statement.where(Document.updated_at <= updated_to)
    rows = db.session.execute(
        statement.order_by(
            AttachmentBinding.sort_no.asc(),
            AttachmentBinding.id.asc(),
        )
    ).all()
    total = len(rows)
    selected = rows[page.offset : page.offset + page.page_size]
    items: list[dict[str, object]] = []
    for binding, attachment in selected:
        item = attachment_summary(
            attachment, context, plan_access=main_access
        )
        item.update(
            {
                "bindingId": serialize_id(binding.id),
                "bindingSortNo": binding.sort_no,
            }
        )
        items.append(item)
    return page_result(
        items,
        page=page.page,
        page_size=page.page_size,
        total=total,
    )


__all__ = [
    "get_attachment",
    "list_attachment_main_plans",
    "list_attachments",
    "list_main_plan_attachments",
    "parse_string_id",
]