LongArticleTaskServer/app/domains/mcp/_utils.py

from __future__ import annotations

import math
from typing import Any, Dict, Tuple

from ._const import LongArticlesMcpConst


class LongArticlesMcpUtils(LongArticlesMcpConst):
    """MCP 工具层：参数解析、校验、分页计算与统一响应构造。"""

    # ---------- 分页计算 ----------

    @staticmethod
    def normalize_pagination(page: int | None, page_size: int | None) -> Tuple[int, int]:
        page = page or LongArticlesMcpConst.DEFAULT_PAGE
        page = max(page, 1)

        page_size = page_size or LongArticlesMcpConst.DEFAULT_PAGE_SIZE
        page_size = max(1, min(page_size, LongArticlesMcpConst.MAX_PAGE_SIZE))
        return page, page_size

    @staticmethod
    def calc_total_pages(total: int, page_size: int) -> int:
        if total <= 0:
            return 0
        return math.ceil(total / page_size)

    # ---------- 参数解析 ----------

    @classmethod
    def parse_pagination(cls, params: Dict[str, Any]) -> Tuple[int, int]:
        page_raw = params.get("page")
        page_size_raw = params.get("page_size")

        try:
            page = int(page_raw) if page_raw is not None else None
            page_size = int(page_size_raw) if page_size_raw is not None else None
        except (TypeError, ValueError) as exc:
            raise ValueError("page and page_size must be integers") from exc

        return cls.normalize_pagination(page, page_size)

    @classmethod
    def parse_sort(cls, params: Dict[str, Any]) -> Tuple[str | None, str | None]:
        sort_by = params.get("sort_by")
        sort_order = params.get("sort_order")

        if sort_by is not None:
            if sort_by not in cls.SORT_FIELDS:
                raise ValueError(f"invalid sort_by: {sort_by}")
        if sort_order is not None:
            sort_order = str(sort_order).lower()
            if sort_order not in cls.SORT_ORDERS:
                raise ValueError(f"invalid sort_order: {sort_order}")

        return sort_by, sort_order

    # ---------- 响应构造 ----------

    @staticmethod
    def build_success(data: Any) -> Dict[str, Any]:
        return {
            "code": 0,
            "message": "success",
            "data": data,
        }

    @staticmethod
    def build_error(code: int, message: str, errors: list[str] | None = None) -> Dict[str, Any]:
        body: Dict[str, Any] = {
            "code": code,
            "message": message,
            "data": None,
        }
        if errors is not None:
            body["errors"] = errors
        return body


__all__ = ["LongArticlesMcpUtils"]