m2-mcp/m2_mcp/config_api.py

from __future__ import annotations

from typing import Any, TypedDict

from .auth import find_project_config, resolve_project_token, _request_json

CONFIG_OPERATOR = "Dataturing"


class SetPointValueItem(TypedDict):
    point_id: str
    value: str


def _post_config(project_key: str, path: str, payload: dict[str, Any]) -> Any:
    return _post_project_json(
        project_key,
        path,
        payload,
        success_key="state",
        message_keys=("state_info",),
    )


def _post_ai(project_key: str, path: str, payload: dict[str, Any]) -> Any:
    return _post_project_json(
        project_key,
        path,
        payload,
        success_key="errcode",
        message_keys=("msg", "message"),
    )


def _post_project_json(
    project_key: str,
    path: str,
    payload: dict[str, Any],
    *,
    success_key: str,
    message_keys: tuple[str, ...],
) -> Any:
    project = find_project_config(project_key)
    authorization = resolve_project_token(project)
    response_payload = _request_json(
        "POST",
        f"{project['base_url']}{path}",
        authorization,
        json_payload=payload,
    )

    if not isinstance(response_payload, dict):
        raise ValueError(f"API returned invalid payload for {path}: {response_payload}")

    status = response_payload.get(success_key)
    if str(status) not in {"0", "0.0"}:
        message = ""
        for key in message_keys:
            message = str(response_payload.get(key) or "").strip()
            if message:
                break
        raise ValueError(f"API failed for {path}: {message or response_payload}")

    return response_payload

def search_ai_systems(
    project_key: str,
    keyword: str | None = None,
    page_size: int = 20,
    page_num: int = 1,
    order_by: list[str] | None = None,
) -> Any:
    return _post_ai(
        project_key,
        "/api/ai/system/search",
        {
            "page": page_num,
            "page_size": page_size,
            "keyword": keyword or "",
            "order_by": order_by or ["-id"],
        },
    )


def search_ai_rcmd_operations(
    project_key: str,
    codes: list[str],
    end: str,
    page_size: int = 20,
    page_num: int = 1,
    order: str = "-create_time",
) -> Any:
    return _post_ai(
        project_key,
        "/api/ai/ai_rcmd_operation/search_ai_rcmd_operation",
        {
            "end": end,
            "codes": codes,
            "order": order,
            "page_size": page_size,
            "page_num": page_num,
        },
    )


def get_ai_online_v2(project_key: str, codes: list[str]) -> Any:
    return _post_ai(
        project_key,
        "/api/ai/ai_rcmd_operation/ai_online_v2",
        {
            "codes": codes,
        },
    )


def set_multi_values(project_key: str, points: list[SetPointValueItem], from_: str = "M2_BACKEND") -> Any:
    return _post_config(
        project_key,
        "/basedataportal/value/set_multi_values",
        {
            "points": points,
            "from": from_,
        },
    )