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

"""主案、子方案只读查询、DTO和查看审计。"""

from __future__ import annotations

import json
import logging
from datetime import datetime, timezone
from typing import Any, Mapping

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

from dms.common.enums import (
    AuditAction,
    AuditTarget,
    DocumentStatus,
    DocumentType,
    RoleCode,
    SecurityLevel,
    VisibilityType,
)
from dms.common.errors import (
    DocumentViewForbiddenError,
    InvalidArgumentError,
    MainPlanNotFoundError,
    ResourceNotFoundError,
    SecurityLevelForbiddenError,
)
from dms.common.pagination import PageRequest, page_result
from dms.common.response import serialize_id
from dms.common.time_range import parse_datetime, parse_updated_range
from dms.database.transaction import transaction
from dms.extensions import db
from dms.models import (
    AttachmentBinding,
    Category,
    Document,
    Permission,
    SubPlanBinding,
)
from dms.security.auth_context import AuthContext, get_auth_context
from dms.services.audit_service import business_audit
from dms.services.authorization_service import (
    PlanAccess,
    evaluate_plan_access,
    plan_allowed_actions,
)
from dms.services.document_password_service import (
    is_view_locked,
    protected_actions_of,
    view_password_verified,
)

logger = logging.getLogger(__name__)

PLAN_TYPES = (DocumentType.MAIN.value, DocumentType.SUB_PLAN.value)
DOCUMENT_SORTS = {
    "documentName": Document.document_name,
    "createdAt": Document.created_at,
    "updatedAt": Document.updated_at,
    "viewCount": Document.view_count,
    "downloadCount": Document.download_count,
    "sortNo": Document.sort_no,
}


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


def _integer(
    params: Mapping[str, str],
    name: str,
    default: int,
) -> int:
    raw = params.get(name)
    if raw is None:
        return default
    if not raw.isdecimal():
        raise InvalidArgumentError(f"{name}必须是正整数")
    return int(raw)


def _page(params: Mapping[str, str]) -> PageRequest:
    return PageRequest(
        page=_integer(params, "page", 1),
        page_size=_integer(params, "pageSize", 20),
    )


def _enum(value: str | None, enum_type, name: str) -> str | None:
    if value is None or value == "":
        return None
    try:
        return enum_type(value).value
    except ValueError as exc:
        raise InvalidArgumentError(f"{name}不是有效枚举值") from exc


def _date(value: str | None, name: str) -> datetime | None:
    return parse_datetime(value, name=name)


def _sort(
    params: Mapping[str, str],
    *,
    allowed: dict[str, Any] = DOCUMENT_SORTS,
    default: str = "updatedAt",
) -> tuple[Any, str]:
    sort_by = params.get("sortBy", default)
    direction = params.get("sortDirection", "desc").lower()
    if sort_by not in allowed:
        raise InvalidArgumentError("sortBy不是允许的排序字段")
    if direction not in {"asc", "desc"}:
        raise InvalidArgumentError("sortDirection必须是asc或desc")
    column = allowed[sort_by]
    return (column.asc() if direction == "asc" else column.desc()), direction


def _iso(value: datetime) -> str:
    return value.replace(tzinfo=timezone.utc).isoformat(timespec="milliseconds").replace(
        "+00:00", "Z"
    )


def _permission_summary(
    access: PlanAccess,
    document: Document,
) -> tuple[dict[str, object], str]:
    source = access.source
    if source is None:
        return {
            "organizationCount": 0,
            "userCount": 0,
            "inheritedFromMainPlan": False,
        }, "未配置"
    permissions = db.session.scalars(
        select(Permission).where(
            Permission.document_id == source.id,
            Permission.is_deleted.is_(False),
        )
    ).all()
    organization_count = sum(item.subject_type == "ORG" for item in permissions)
    user_count = sum(item.subject_type == "USER" for item in permissions)
    visibility = VisibilityType(source.visibility_type)
    if visibility == VisibilityType.ALL_AUTHENTICATED:
        visibility_summary = "全部已登录用户"
    else:
        names = [item.subject_name for item in permissions if item.can_view]
        visibility_summary = "、".join(names) if names else "未配置"
    return {
        "organizationCount": organization_count,
        "userCount": user_count,
        "inheritedFromMainPlan": document.document_type
        == DocumentType.SUB_PLAN.value,
    }, visibility_summary


_SNIPPET_BEFORE = 40
_SNIPPET_AFTER = 80


def _content_snippet(content_text: str | None, keyword: str | None) -> str | None:
    """当 keyword 命中正文时，截取包含命中区域的片段。

    前后各保留若干字符作为上下文；若未触达正文首尾则以"……"标识截断。
    未命中或任一参数为空时返回 None。
    """
    if not content_text or not keyword or not keyword.strip():
        return None
    kw = keyword.strip()
    pos = content_text.lower().find(kw.lower())
    if pos < 0:
        return None
    start = max(pos - _SNIPPET_BEFORE, 0)
    end = min(pos + len(kw) + _SNIPPET_AFTER, len(content_text))
    snippet = content_text[start:end]
    if start > 0:
        snippet = "……" + snippet
    if end < len(content_text):
        snippet += "……"
    return snippet


def document_summary(
    document: Document,
    context: AuthContext,
    access: PlanAccess,
    keyword: str | None = None,
) -> dict[str, object]:
    # 无权限方案也在列表展示（需求：全员可见列表），但授权名单属敏感信息，
    # visibilitySummary 以占位符返回；正文片段同样只在有权限时生成。
    if access.allowed:
        _, visibility_summary = _permission_summary(access, document)
    else:
        visibility_summary = "-"
    source = access.source or document
    view_locked = is_view_locked(document, context)
    result: dict[str, object] = {
        "id": serialize_id(document.id),
        "documentName": document.document_name,
        "summary": document.summary,
        "description": document.description if access.allowed else None,
        "documentType": document.document_type,
        "status": document.document_status,
        "securityLevel": document.security_level,
        "visibilityType": source.visibility_type,
        "visibilitySummary": visibility_summary,
        "attachmentType": document.attachment_type,
        "categoryId": serialize_id(document.category_id),
        "categoryName": document.category_name,
        "categoryPath": document.category_path,
        "organizationId": serialize_id(document.organization_id),
        "organizationName": document.organization_name,
        "parentDocumentId": serialize_id(document.parent_document_id),
        "rootDocumentId": serialize_id(
            document.id
            if document.document_type == DocumentType.MAIN.value
            else document.root_document_id
        ),
        "tags": document.tags or [],
        "fileExtension": document.file_extension,
        "childCount": document.child_count,
        "attachmentCount": document.attachment_count,
        "viewCount": document.view_count,
        "downloadCount": document.download_count,
        "createdByName": document.created_by_name,
        "createdAt": _iso(document.created_at),
        "updatedAt": _iso(document.updated_at),
        "rowVersion": document.row_version,
        "allowedActions": plan_allowed_actions(document, context, access),
        "hasPassword": document.password_hash is not None,
        "passwordProtectedActions": protected_actions_of(document),
        "viewLocked": view_locked,
    }
    if not view_locked:
        snippet = _content_snippet(document.content_text, keyword)
        if snippet is not None:
            result["contentSnippet"] = snippet
    return result


def document_detail(
    document: Document,
    context: AuthContext,
    access: PlanAccess,
) -> dict[str, object]:
    result = document_summary(document, context, access)
    permission_summary, _ = _permission_summary(access, document)
    verified = view_password_verified(document, context)
    result.update(
        {
            "originalFileName": document.original_file_name,
            "mimeType": document.mime_type,
            "fileSize": document.file_size,
            "fileHash": document.file_hash,
            "permissionSummary": permission_summary,
            "viewPasswordVerified": verified,
            "contentText": document.content_text if verified else None,
        }
    )
    # 提取状态/失败原因仅供管理员排障（"编辑"界面提示与重新解析）
    if context.role_code == RoleCode.ADMIN:
        result.update(
            {
                "contentExtractStatus": document.content_extract_status,
                "contentExtractError": document.content_extract_error,
                "contentExtractedAt": (
                    _iso(document.content_extracted_at)
                    if document.content_extracted_at
                    else None
                ),
            }
        )
    return result


def _require_plan_access(document: Document, context: AuthContext) -> PlanAccess:
    access = evaluate_plan_access(document, context)
    if access.allowed:
        return access
    if access.reason == "SECURITY":
        raise SecurityLevelForbiddenError()
    raise DocumentViewForbiddenError()


def _escape_like(keyword: str) -> str:
    return keyword.strip().replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def _title_keyword(statement, keyword: str | None):
    """简单关键词：仅模糊匹配标题（document_name）。"""
    if not keyword or not keyword.strip():
        return statement
    pattern = f"%{_escape_like(keyword)}%"
    return statement.where(Document.document_name.like(pattern, escape="\\"))


def _security_levels(params: Mapping[str, str]) -> list[str] | None:
    """解析 ``securityLevels`` 逗号分隔多选（勾选项内 OR）。"""
    raw = params.get("securityLevels")
    if raw is None or not raw.strip():
        return None
    values = [item.strip() for item in raw.split(",") if item.strip()]
    if not values:
        return None
    for item in values:
        try:
            SecurityLevel(item)
        except ValueError as exc:
            raise InvalidArgumentError("securityLevels包含无效密级值") from exc
    if len(set(values)) != len(values):
        raise InvalidArgumentError("securityLevels不允许重复值")
    return values


def _advanced_filters(statement, params: Mapping[str, str]):
    """高级检索三条件（正文关键词 / 密级多选 / 部门模糊），全部 AND 叠加。"""
    content_keyword = params.get("contentKeyword")
    if content_keyword and content_keyword.strip():
        pattern = f"%{_escape_like(content_keyword)}%"
        statement = statement.where(
            or_(
                Document.content_text.like(pattern, escape="\\"),
                Document.search_text.like(pattern, escape="\\"),
            )
        )
    levels = _security_levels(params)
    if levels:
        statement = statement.where(Document.security_level.in_(levels))
    organization_keyword = params.get("organizationKeyword")
    if organization_keyword and organization_keyword.strip():
        pattern = f"%{_escape_like(organization_keyword)}%"
        statement = statement.where(
            Document.organization_name.like(pattern, escape="\\")
        )
    return statement


def _title_contains(document_name: str | None, keyword: str) -> bool:
    return keyword.lower() in (document_name or "").lower()


def _child_title_hit_counts(
    context: AuthContext,
    main_ids: list[int],
    keyword: str,
) -> dict[int, list[int]]:
    """统计各主案下标题命中简单关键词的可见子案/附件数量。

    - 子案经 ``evaluate_plan_access`` 逐条校验，不可见的命中不计入；
    - 附件随挂载主案权限可见（主案本身已通过校验），直接计数；
    - 返回 ``{main_id: [子案命中数, 附件命中数]}``。
    """
    if not main_ids or not keyword.strip():
        return {}
    pattern = f"%{_escape_like(keyword)}%"
    counts: dict[int, list[int]] = {main_id: [0, 0] for main_id in main_ids}
    children = db.session.scalars(
        select(Document).where(
            Document.parent_document_id.in_(main_ids),
            Document.document_type == DocumentType.SUB_PLAN.value,
            Document.is_deleted.is_(False),
            Document.document_name.like(pattern, escape="\\"),
        )
    ).all()
    for child in children:
        if evaluate_plan_access(child, context).allowed:
            counts[child.parent_document_id][0] += 1
    bindings = db.session.execute(
        select(AttachmentBinding.main_document_id, Document.id)
        .join(Document, Document.id == AttachmentBinding.attachment_document_id)
        .where(
            AttachmentBinding.main_document_id.in_(main_ids),
            AttachmentBinding.is_deleted.is_(False),
            Document.document_type == DocumentType.ATTACHMENT.value,
            Document.is_deleted.is_(False),
            Document.document_name.like(pattern, escape="\\"),
        )
    ).all()
    for main_id, _ in bindings:
        counts[main_id][1] += 1
    return counts


def _tags_filter(
    statement,
    tags_value: str | None,
    match_value: str | None,
):
    """按 ``tags`` JSON 数组精确匹配；``tagsMatch=any|all`` 决定 OR/AND 语义。"""
    if not tags_value or not tags_value.strip():
        return statement
    tags = [item.strip() for item in tags_value.split(",") if item.strip()]
    if not tags:
        return statement
    match = (match_value or "any").strip().lower()
    if match not in {"any", "all"}:
        raise InvalidArgumentError("tagsMatch只允许any或all")
    # 达梦无 JSON_CONTAINS：对 JSON 编码后的标签（含引号边界）做 LIKE 包含。
    # ensure_ascii=False 保持原文中文，与达梦存量数据及 Java 版写入格式一致；
    # _escape_like 处理 %、_ 与反斜杠。
    conditions = [
        Document.tags.like(
            f"%{_escape_like(json.dumps(tag, ensure_ascii=False))}%", escape="\\"
        )
        for tag in tags
    ]
    combiner = and_ if match == "all" else or_
    return statement.where(combiner(*conditions))


def _boolean(value: str | None, name: str, default: bool = False) -> bool:
    if value is None:
        return default
    if value not in {"true", "false"}:
        raise InvalidArgumentError(f"{name}必须是true或false")
    return value == "true"


def _category_scope(category_id: int, include_descendants: bool) -> set[int]:
    categories = db.session.scalars(
        select(Category).where(
            Category.is_deleted.is_(False),
            Category.status == "ENABLED",
        )
    ).all()
    if not any(category.id == category_id for category in categories):
        raise ResourceNotFoundError("方案分类不存在或不可用")
    if not include_descendants:
        return {category_id}
    children: dict[int | None, list[int]] = {}
    for category in categories:
        children.setdefault(category.parent_id, []).append(category.id)
    result = {category_id}
    pending = [category_id]
    while pending:
        current = pending.pop()
        for child_id in children.get(current, []):
            if child_id not in result:
                result.add(child_id)
                pending.append(child_id)
    return result


def list_documents(params: Mapping[str, str]) -> dict[str, object]:
    context = get_auth_context()
    page = _page(params)
    raw_types = params.get("documentType", "MAIN")
    document_types = [item.strip() for item in raw_types.split(",") if item.strip()]
    if (
        not document_types
        or any(item not in PLAN_TYPES for item in document_types)
        or len(set(document_types)) != len(document_types)
    ):
        raise InvalidArgumentError("documentType只允许MAIN、SUB_PLAN或二者组合")
    statement = select(Document).where(
        Document.is_deleted.is_(False),
        Document.document_type.in_(document_types),
    )
    include_descendants = _boolean(
        params.get("includeDescendants"), "includeDescendants"
    )
    category_id = params.get("categoryId")
    if category_id:
        parsed_category_id = parse_string_id(category_id, field="categoryId")
        statement = statement.where(
            Document.category_id.in_(
                _category_scope(parsed_category_id, include_descendants)
            )
        )
    aggregate = _boolean(params.get("aggregateHits"), "aggregateHits")
    if aggregate and document_types != ["MAIN"]:
        raise InvalidArgumentError("aggregateHits只允许在documentType=MAIN时使用")
    if not aggregate:
        statement = _title_keyword(statement, params.get("keyword"))
    statement = _advanced_filters(statement, params)
    statement = _tags_filter(
        statement, params.get("tags"), params.get("tagsMatch")
    )
    visibility_filter = _enum(
        params.get("visibilityType"), VisibilityType, "visibilityType"
    )
    for name, enum_type, column in (
        ("securityLevel", SecurityLevel, Document.security_level),
        ("status", DocumentStatus, Document.document_status),
    ):
        value = _enum(params.get(name), enum_type, name)
        if value:
            statement = statement.where(column == value)
    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)
    documents = db.session.scalars(statement.order_by(order, Document.id.asc())).all()
    # 列表全量可见（需求3）：无权限方案同样返回，allowedActions 为空集，
    # 前端据此禁用查看类按钮；预览/详情/下载接口仍独立拦截。
    listed: list[tuple[Document, PlanAccess]] = []
    for document in documents:
        access = evaluate_plan_access(document, context)
        if visibility_filter is not None:
            source_type = (
                access.source.visibility_type
                if access.source is not None
                else document.visibility_type
            )
            if source_type != visibility_filter:
                continue
        listed.append((document, access))
    total = len(listed)
    keyword = (params.get("keyword") or "").strip()
    hit_counts: dict[int, list[int]] = {}
    if aggregate:
        if keyword:
            # 子案/附件命中统计只覆盖有权限主案，避免借计数泄露无权限主案的挂载信息。
            hit_counts = _child_title_hit_counts(
                context, [document.id for document, access in listed if access.allowed], keyword
            )
            listed = [
                (document, access)
                for document, access in listed
                if _title_contains(document.document_name, keyword)
                or any(hit_counts.get(document.id))
            ]
            total = len(listed)
    selected = listed[page.offset : page.offset + page.page_size]
    items = [
        document_summary(
            document, context, access,
            keyword=params.get("contentKeyword") if access.allowed else None,
        )
        for document, access in selected
    ]
    if aggregate:
        for item, (document, access) in zip(items, selected, strict=True):
            counts = hit_counts.get(document.id, [0, 0]) if access.allowed else [0, 0]
            item["titleHit"] = not keyword or _title_contains(
                document.document_name, keyword
            )
            item["subHitCount"] = counts[0]
            item["attachmentHitCount"] = counts[1]
    return page_result(
        items,
        page=page.page,
        page_size=page.page_size,
        total=total,
    )


def _active_plan(document_id: int) -> Document:
    document = db.session.scalar(
        select(Document).where(
            Document.id == document_id,
            Document.is_deleted.is_(False),
            Document.document_type.in_(PLAN_TYPES),
        )
    )
    if document is None:
        raise ResourceNotFoundError("方案文档不存在")
    return document


def _record_view(document: Document) -> int:
    old_count = document.view_count
    document_id = document.id
    document_name = document.document_name
    target = (
        AuditTarget.ATTACHMENT
        if document.document_type == DocumentType.ATTACHMENT.value
        else AuditTarget.DOCUMENT
    )
    db.session.rollback()
    try:
        with transaction() as session:
            current = session.scalar(
                select(Document)
                .where(
                    Document.id == document_id,
                    Document.is_deleted.is_(False),
                )
                .with_for_update()
            )
            if current is None:
                raise ResourceNotFoundError("文档不存在")
            current.view_count += 1
            session.add(
                business_audit(
                    action=AuditAction.VIEW_DOCUMENT,
                    target=target,
                    target_id=document_id,
                    target_name=document_name,
                    detail={"documentType": current.document_type},
                )
            )
            new_count = current.view_count
        return new_count
    except Exception:
        db.session.rollback()
        logger.exception("文档查看计数或审计写入失败：document_id=%s", document_id)
        return old_count


def get_document(document_id: int) -> dict[str, object]:
    context = get_auth_context()
    document = _active_plan(document_id)
    access = _require_plan_access(document, context)
    result = document_detail(document, context, access)
    result["viewCount"] = _record_view(document)
    return result


def list_sub_plans(
    main_document_id: int,
    params: Mapping[str, str],
) -> dict[str, object]:
    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()
    # 主案无权限时也返回子案列表（列表全量可见，与主案列表语义一致）；
    # 子案权限继承自原主案（root_document_id 不变），allowedActions 自然为
    # 空集，内容由预览/详情接口拦截。
    evaluate_plan_access(main, context)
    page = _page(params)

    def _apply_filters(statement):
        statement = _title_keyword(statement, params.get("keyword"))
        statement = _advanced_filters(statement, params)
        status = _enum(params.get("status"), DocumentStatus, "status")
        if status:
            statement = statement.where(Document.document_status == status)
        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)
        return statement

    order, _ = _sort(params)
    # 自有子案：parent_document_id 指向该主案，按请求排序（默认 sortNo）。
    own_children = db.session.scalars(
        _apply_filters(
            select(Document).where(
                Document.parent_document_id == main.id,
                Document.document_type == DocumentType.SUB_PLAN.value,
                Document.is_deleted.is_(False),
            )
        ).order_by(order, Document.id.asc())
    ).all()
    # 挂载子案：其他主案的子案经 doc_sub_plan_binding 建立展示关系，
    # 按挂载顺序（binding.sort_no）固定追加在自有子案之后，不参与拖动排序。
    mount_rows = db.session.execute(
        select(SubPlanBinding, Document)
        .join(Document, Document.id == SubPlanBinding.sub_document_id)
        .where(
            SubPlanBinding.main_document_id == main.id,
            SubPlanBinding.is_deleted.is_(False),
            Document.is_deleted.is_(False),
            Document.document_type == DocumentType.SUB_PLAN.value,
        )
        .order_by(SubPlanBinding.sort_no.asc(), SubPlanBinding.id.asc())
    ).all()
    mounted: list[tuple[SubPlanBinding, Document]] = []
    if mount_rows:
        filtered_ids = set(
            db.session.scalars(
                _apply_filters(
                    select(Document.id).where(
                        Document.id.in_([doc.id for _, doc in mount_rows])
                    )
                )
            ).all()
        )
        mounted = [row for row in mount_rows if row[0].sub_document_id in filtered_ids]

    merged: list[tuple[Document, int | None]] = [
        *[(child, None) for child in own_children],
        *[(doc, binding.id) for binding, doc in mounted],
    ]
    total = len(merged)
    selected = merged[page.offset : page.offset + page.page_size]
    items: list[dict[str, object]] = []
    for child, mount_id in selected:
        access = evaluate_plan_access(child, context)
        summary = document_summary(
            child, context, access,
            keyword=params.get("contentKeyword") if access.allowed else None,
        )
        if mount_id is not None:
            summary["mountId"] = serialize_id(mount_id)
        items.append(summary)
    return page_result(
        items,
        page=page.page,
        page_size=page.page_size,
        total=total,
    )


def list_main_plan_tags() -> dict[str, object]:
    """聚合当前用户可见的所有主案标签。

    - 全局视角：标签集合不随分类、关键词等其他筛选变化；
    - 权限：通过 ``evaluate_plan_access`` 在 Python 层逐条过滤，
      与 ``list_documents`` 保持一致的可见性语义，避免泄露被
      ACL/密级隔离的主案标签；
    - 顺序：按 Unicode 升序，便于前端稳定渲染。
    """
    context = get_auth_context()
    statement = select(Document).where(
        Document.is_deleted.is_(False),
        Document.document_type == DocumentType.MAIN.value,
    )
    documents = db.session.scalars(statement).all()
    tags_set: set[str] = set()
    for document in documents:
        access = evaluate_plan_access(document, context)
        if not access.allowed:
            continue
        if document.tags:
            tags_set.update(document.tags)
    tags = sorted(tags_set)
    return {"items": tags, "total": len(tags)}


__all__ = [
    "DOCUMENT_SORTS",
    "_advanced_filters",
    "_content_snippet",
    "_date",
    "_escape_like",
    "_enum",
    "_iso",
    "_page",
    "_record_view",
    "_sort",
    "_tags_filter",
    "_title_keyword",
    "document_detail",
    "document_summary",
    "get_document",
    "list_documents",
    "list_main_plan_tags",
    "list_sub_plans",
    "parse_string_id",
]