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

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

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

导言

通过外部API与Engee合作为自动化技术计算、开发和优化模型、使用计算机建模测试复杂技术系统以及最重要的是与外部工程环境集成提供了更广泛的可能性。
要开始使用，您需要阅读[документашиц]（https://engee.com/helpcenter/stable/ru/external-software/external-software-interface-for-engee.html ）到Engee 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已被编辑以保护个人数据。

EngeeManager实现

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

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):
    """Engee API的抽象接口。"""

    @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):
    """实现与Engee交互的API。"""

    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。文本}")
                    return True
                else:
                    print(f"错误：{响应。status_code}-{响应。文本}")
            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）
        返回：成功时为真，错误时为假
        """

        # 确定是否已通过一个路径或多个路径
        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/'开头。 指定：{路径}")
                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}-{响应。文本}")
                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}")

                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:
    """管理工程师的主要课程"""

    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)

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

print(f"当前状态：{经理。get_status()}")

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

工程师发射

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

manager.start()
print(f"当前状态：{经理。get_status()}")

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

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

manager.command("pwd()")

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

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

manager.start()

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

执行命令

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

print(f"当前状态：{经理。get_status()}")
pwd = manager.command("pwd()")
engee_version = manager.command("engee.version()")

Текущий статус: ServerStatus.RUNNING
/user
25.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"当前状态：{经理。get_status()}")

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

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

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

结论

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

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