使用AnyMath外部API（个人访问令牌）

在示例中，我们考虑使用pat（个人访问令牌）使用外部AnyMath API-AnyMathManager控件类实现，检查状态，启动，停止，执行命令，上传和下载文件的功能。

此示例不考虑异步代码执行，以及使用OAuth2.0访问AnyMath API的实现。

导言

通过外部API与AnyMath合作为自动化技术计算、开发和优化模型、使用计算机建模测试复杂技术系统以及最重要的是与外部工程环境集成提供了更广泛的可能性。
要开始使用，您需要阅读[документахиц]（https://engee.com/helpcenter/stable/cn/external-software/external-software-interface-for-engee.html ）到AnyMath API。它还告诉您如何获取个人访问令牌（PAT）。

环境准备

在开发工作示例时，使用了Python3.13版本。
安装必要的依赖项:

%pip install -r requirements.txt

Requirement already satisfied: requests in c:\users\alexe\appdata\local\programs\python\python313\lib\site-packages (from -r requirements.txt (line 1)) (2.32.5)Requirement already satisfied: charset_normalizer<4,>=2 in c:\users\alexe\appdata\local\programs\python\python313\lib\site-packages (from requests->-r requirements.txt (line 1)) (3.4.2)Requirement already satisfied: idna<4,>=2.5 in c:\users\alexe\appdata\local\programs\python\python313\lib\site-packages (from requests->-r requirements.txt (line 1)) (3.10)Requirement already satisfied: urllib3<3,>=1.21.1 in c:\users\alexe\appdata\local\programs\python\python313\lib\site-packages (from requests->-r requirements.txt (line 1)) (2.2.1)Requirement already satisfied: certifi>=2017.4.17 in c:\users\alexe\appdata\local\programs\python\python313\lib\site-packages (from requests->-r requirements.txt (line 1)) (2024.12.14)Note: you may need to restart the kernel to use updated packages.

收到此项目的令牌后，只需在文件中指定它就足够了。 .env 附项目。将项目发布到公共存储库时，不要忘记删除此文件或添加它。 .gitignore，并且还在存储库secrets中指定您的PAT。

项目中指定的PAT。env已被编辑以保护个人数据。

AnyMathManager实现

下面是所有必要的类和函数的实现。此外，他们提供了解释，并在难以理解的逻辑的地方-与评论。

import os
from abc import ABC, abstractmethod
from typing import Optional, Dict, Any, Union
from dataclasses import dataclass
import requests
import json
from enum import Enum
from dotenv import load_dotenv

class ServerStatus(str, Enum):
    RUNNING = "running"
    STOPPED = "stopped"
    STARTING = "starting"

@dataclass
class EngeeConfig:
    """Конфигурация подключения к Engee."""
    pat: str
    base_url: str = "https://engee.com"
    timeout: float = 36000.0  # 默认空闲超时，秒
    # serverName: str = "demo00000000-engeeid"


class IEngeeAPI(ABC):
    """Абстрактный интерфейс для API Engee."""

    @abstractmethod
    def get_server_info(self) -> Optional[Dict[str, Any]]:
        pass

    @abstractmethod
    def start_server(self) -> Optional[Dict[str, Any]]:
        pass

    @abstractmethod
    def stop_server(self) -> bool:
        pass

    @abstractmethod
    def execute_command(self, command: str) -> Optional[str]:
        pass

    @abstractmethod
    def upload_file(self, destination_path: str, local_path: str) -> bool:
        pass


class EngeeAPI(IEngeeAPI):
    """Реализация API для взаимодействия с Engee."""

    def __init__(self, config: EngeeConfig):
        self.config = config
        self._headers = {"Authorization": f"Bearer {config.pat}"}

    def _make_request(self, method: str, url: str, **kwargs) -> Optional[requests.Response]:
        """Внутренний метод для выполнения HTTP-запросов."""
        try:
            response = requests.request(method, url, headers=self._headers, **kwargs)
            return response if response.status_code < 400 else None
        except requests.exceptions.RequestException as e:
            print(f"Ошибка запроса: {e}")
            return None

    def get_server_info(self) -> Optional[Dict[str, Any]]:
        """Получить информацию о состоянии сервера."""
        info_url = f"{self.config.base_url}/account/api/engee/info"
        response = self._make_request("GET", info_url)
        return response.json() if response else None

    def start_server(self) -> Optional[Dict[str, Any]]:
        """Запустить сервер."""
        info = self.get_server_info()
        if not info or info.get("serverStatus") != ServerStatus.STOPPED:
            print("服务器没有处于\"停止\"状态，启动是不可能的。")
            return None

        start_url = f"{self.config.base_url}/account/api/engee/start"
        payload = {
            "url": info["url"],
            "inactivityTimeout": self.config.timeout
        }
        headers = {**self._headers, "Content-Type": "application/json"}

        response = self._make_request("POST", start_url, json=payload)
        return response.json() if response else None

    def stop_server(self) -> bool:
        """Остановить сервер."""
        stop_url = f"{self.config.base_url}/account/api/engee/stop"
        response = self._make_request("DELETE", stop_url)
        success = response is not None and response.status_code == 204
        if success:
            print("服务器已停止")
        return success

    def execute_command(self, command: str) -> Optional[str]:
        """Выполнить команду на сервере."""
        info = self.get_server_info()
        if not info:
            print("无法获取有关服务器的信息。")
            return None

        if info.get("serverStatus") != ServerStatus.RUNNING:
            status = info.get("serverStatus")
            if status == ServerStatus.STOPPED:
                print("服务器未运行。 使用'start_server()'命令启动服务器。")
            else:
                print("服务器正在启动。 请稍后再试。")
            return None

        server_url = f"{self.config.base_url}{info['url']}"
        eval_url = f"{server_url}/external/command/eval"
        payload = {"command": command}
        headers = {**self._headers, "Content-Type": "application/json"}

        response = self._make_request("POST", eval_url, json=payload)
        if response and response.status_code == 201:
            result = response.json().get("result")
            print(result)
            return result
        else:
            print("No result")
            return None

    def upload_file(self, destination_path: str, local_path: str) -> bool:
        """Загрузить файл на сервер."""
        if not os.path.isfile(local_path):
            raise FileNotFoundError(f"Файл не найден: {local_path}")

        info = self.get_server_info()
        if not info:
            print("无法获取有关服务器的信息。")
            return False

        if info.get("serverStatus") != ServerStatus.RUNNING:
            status = info.get("serverStatus")
            if status == ServerStatus.STOPPED:
                print("服务器未运行。 使用'start_server()'命令启动服务器。")
            else:
                print("服务器正在启动。 请稍后再试。")
            return False

        server_url = f"{self.config.base_url}{info['url']}"
        upload_url = f"{server_url}/external/file/upload"
        filename = os.path.basename(local_path)

        files = {
            'paths': (None, json.dumps({"sources": [destination_path]})),
            'upload_files': (filename, open(local_path, 'rb'))
        }

        try:
            headers = {**self._headers} 
            response = self._make_request("POST", upload_url, files=files)
            if response:
                if response.status_code == 201:
                    print("上传成功")
                    return True
                elif response.status_code == 200:
                    print(f"Файл с таким именем существует. Загруженному файлу присвоено имя {response.text}")
                    return True
                else:
                    print(f"Ошибка: {response.status_code} – {response.text}")
            return False
        finally:
            if 'upload_files' in files and hasattr(files['upload_files'][1], 'close'):
                files['upload_files'][1].close()
    
    def download(self, remote_path: Union[str, list[str]], local_save_path: str) -> bool:
        """
        Универсальный метод скачивания.
        :param remote_path: строка (один файл) или список строк (несколько файлов)
        :param local_save_path: локальный путь для сохранения (файл или ZIP)
        :return: True при успехе, False при ошибке
        """

        # 确定是否已通过一个路径或多个路径
        if isinstance(remote_path, str):
            paths_to_download = [remote_path]
            single_file_mode = True
        elif isinstance(remote_path, list):
            paths_to_download = remote_path
            single_file_mode = False
        else:
            print("错误：remote_path必须是字符串或字符串列表。")
            return False

        # 我们检查所有路径都以'/user/'开头
        for path in paths_to_download:
            if not path.startswith('/user/'):
                print(f"Ошибка: Все пути должны начинаться с '/user/'. Указан: {path}")
                return False

        info = self.get_server_info()
        if not info:
            print("无法获取有关服务器的信息。")
            return False

        if info.get("serverStatus") != ServerStatus.RUNNING:
            status = info.get("serverStatus")
            if status == ServerStatus.STOPPED:
                print("服务器未运行。 使用'start_server()'命令启动服务器。")
            else:
                print("服务器正在启动。 请稍后再试。")
            return False

        server_url = f"{self.config.base_url}{info['url']}"
        download_url = f"{server_url}/external/file/download"

        payload = {"sources": paths_to_download}

        try:
            response = self._make_request("POST", download_url, json=payload)
            if not response:
                print("下载请求失败。")
                return False

            if response.status_code == 404:
                print(f"Один или несколько файлов из списка не существуют на сервере: {paths_to_download}")
                return False

            if response.status_code == 403:
                print(f"Один или несколько файлов нарушают правило: должны быть внутри /user/. Проверьте: {paths_to_download}")
                return False

            if response.status_code != 200:
                print(f"Ошибка при скачивании: {response.status_code} – {response.text}")
                return False

            # 创建要保存的目录
            os.makedirs(os.path.dirname(local_save_path), exist_ok=True)

            content_type = response.headers.get('Content-Type', '')

            if single_file_mode:
                # 下载为单个文件
                with open(local_save_path, 'wb') as f:
                    f.write(response.content)
                print(f"Файл '{remote_path}' успешно скачан в '{local_save_path}'")
            else:
                # 下载为ZIP档案
                if 'application/x-zip-compressed' not in content_type and 'zip' not in content_type:
                    print(f"Предупреждение: Ответ не является ZIP-архивом. Content-Type: {content_type}")

                with open(local_save_path, 'wb') as f:
                    f.write(response.content)
                print(f"Архив с файлами успешно скачан в '{local_save_path}'")

            return True

        except Exception as e:
            print(f"Ошибка при скачивании: {e}")
            return False

class EngeeManager:
    """Основной класс для управления Engee"""

    def __init__(self, config: EngeeConfig):
        self.api = EngeeAPI(config)

    def get_status(self) -> Optional[ServerStatus]:
        """Получить текущий статус сервера."""
        info = self.api.get_server_info()
        if info:
            status_str = info.get("serverStatus")
            try:
                return ServerStatus(status_str)
            except ValueError:
                return None
        return None

    def start(self) -> Optional[Dict[str, Any]]:
        """Запустить сервер."""
        return self.api.start_server()

    def stop(self) -> bool:
        """Остановить сервер."""
        return self.api.stop_server()

    def command(self, command: str) -> Optional[str]:
        """Выполнить команду на сервере."""
        return self.api.execute_command(command)

    def upload(self, destination_path: str, local_path: str) -> bool:
        """Загрузить файл на сервер."""
        return self.api.upload_file(destination_path, local_path)
    
    def download(self, remote_path: Union[str, list[str]], local_save_path: str) -> bool:
        """
        Универсальный метод скачивания: один файл или несколько.
        :param remote_path: строка или список строк
        :param local_save_path: куда сохранить
        """
        return self.api.download(remote_path, local_save_path)


def load_config_from_env(env_file: str = ".env") -> EngeeConfig:
    """Загрузить конфигурацию из файла .env."""
    try:
        load_dotenv()
    except ImportError:
        print("未安装python-dotenv库。 跳过下载。恩维")

    pat = os.environ.get("ENGEE_PAT")
    if not pat:
        raise ValueError("Переменная окружения ENGEE_PAT не найдена.")

    return EngeeConfig(pat=pat)

使用示例

从文件下载配置 .env 并创建一个示例类 EngeeManager:

config = load_config_from_env()
manager = EngeeManager(config)

AnyMath已停止，让我们检查状态:

print(f"Текущий статус: {manager.get_status()}")

Текущий статус: ServerStatus.STOPPED

工程师发射

让我们启动AnyMath并检查状态:

manager.start()
print(f"Текущий статус: {manager.get_status()}")

Текущий статус: ServerStatus.STARTING

当会话未运行时，我们将无法执行命令。:

manager.command("pwd()")

Сервер запускается. Повторите попытку позднее.

尝试再次运行它也不会给出结果。:

manager.start()

Сервер не находится в состоянии 'stopped', запуск невозможен.

执行命令

启动AnyMath会话后，我们可以立即通过API发送命令来执行:

print(f"Текущий статус: {manager.get_status()}")
pwd = manager.command("pwd()")
engee_version = manager.command("engee.version()")

Текущий статус: ServerStatus.RUNNING/user25.12.2-H1

上传文件

将文件上传到服务器:

upload_status = manager.upload("/user/upload_test/", r'D:\engee_api\upload_test\upload_test.txt')

Файл с таким именем существует. Загруженному файлу присвоено имя {"paths":["/user/upload_test/upload_test.txt"]}

该函数在成功加载时返回一个布尔值True。让我们检查上传的文件是否在服务器上。:

files = manager.command('readdir("/user/upload_test/")')

["upload_test.txt"]

我们将尝试重新上传同名文件。:

upload_status = manager.upload("/user/upload_test/", r'D:\engee_api\upload_test\upload_test.txt')

Файл с таким именем существует. Загруженному файлу присвоено имя {"paths":["/user/upload_test/upload_test_1.txt"]}

服务器上目标目录的内容:

files = manager.command('readdir("/user/upload_test/")')

["upload_test.txt", "upload_test_1.txt"]

下载文件

将以前上传的文件之一下载到计算机上的文件夹中:

manager.download("/user/upload_test/upload_test_1.txt", r"D:\engee_api\download_test\download_test.txt")

Файл '/user/upload_test/upload_test_1.txt' успешно скачан в 'D:\engee_api\download_test\download_test.txt'

True

让我们尝试一次下载几个文件。我们记得在变量的前面 files 我们在服务器上的结果目录中记录了一个文件列表。将此列表传递给相应的函数:

manager.download(
        files,
        r"D:\engee_api\download_test\download_test.zip"
    )

Ошибка: Все пути должны начинаться с '/user/'. Указан: ["upload_test.txt", "upload_test_1.txt"]

False

我们收到了预期的错误-根据文档，我们必须指定下载文件的路径，child /user/.

manager.download(
        ["/user/upload_test/upload_test.txt", "/user/upload_test/upload_test_1.txt"],
        r"D:\engee_api\download_test\download_test.zip"
    )

Архив с файлами успешно скачан в 'D:\engee_api\download_test\download_test.zip'

True

这些文件被下载到计算机上，您可以在本例的文件夹中检查它们的存在。

工程师停止

stop_status = manager.stop()
print(f"Текущий статус: {manager.get_status()}")

Сервер остановленТекущий статус: ServerStatus.STOPPED

pwd = manager.command("pwd()")

Сервер не запущен. Используйте команду `start_server()` для запуска сервера.

结论

我们已经回顾了通过PAT使用外部AnyMath API的类的实现，以及用于检查状态、启动、停止AnyMath、发送命令和上传/下载文件的用例和函数结果。

但是，这些并不是AnyMath API的所有功能，即使在当前版本中也可用-例如，我们在本例中没有考虑[异步执行]。 кода](https://engee.com/helpcenter/stable/cn/external-software/external-software-interface-for-engee.html#асинхронное-исполнение-кода)。但这是一个值得单独和详细考虑的话题。