277 lines
8.4 KiB
Python
277 lines
8.4 KiB
Python
"""智能算法展示平台客户端"""
|
||
|
||
import requests
|
||
import json
|
||
from typing import List, Optional, Dict, Any
|
||
|
||
from .models import Algorithm, AlgorithmVersion, AlgorithmCallRequest, AlgorithmCallResult
|
||
|
||
|
||
class AlgorithmShowcaseClient:
|
||
"""智能算法展示平台客户端类"""
|
||
|
||
def __init__(self, base_url: str = "http://localhost:8000/api/v1", api_key: Optional[str] = None):
|
||
"""初始化客户端
|
||
|
||
Args:
|
||
base_url: API基础URL
|
||
api_key: API密钥
|
||
"""
|
||
self.base_url = base_url.rstrip("/")
|
||
self.api_key = api_key
|
||
self.session = requests.Session()
|
||
|
||
# 设置默认请求头
|
||
self.session.headers.update({
|
||
"Content-Type": "application/json"
|
||
})
|
||
|
||
# 如果提供了API密钥，设置认证头
|
||
if self.api_key:
|
||
self.session.headers.update({
|
||
"Authorization": f"Bearer {self.api_key}"
|
||
})
|
||
|
||
def _request(self, method: str, endpoint: str, data: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
|
||
"""发送请求
|
||
|
||
Args:
|
||
method: 请求方法
|
||
endpoint: 端点
|
||
data: 请求数据
|
||
|
||
Returns:
|
||
响应数据
|
||
"""
|
||
url = f"{self.base_url}/{endpoint}"
|
||
|
||
try:
|
||
if method == "GET":
|
||
response = self.session.get(url, params=data)
|
||
elif method == "POST":
|
||
response = self.session.post(url, json=data)
|
||
elif method == "PUT":
|
||
response = self.session.put(url, json=data)
|
||
elif method == "DELETE":
|
||
response = self.session.delete(url, json=data)
|
||
else:
|
||
raise ValueError(f"Unsupported method: {method}")
|
||
|
||
# 检查响应状态
|
||
response.raise_for_status()
|
||
|
||
# 返回响应数据
|
||
return response.json()
|
||
except requests.RequestException as e:
|
||
raise Exception(f"API request failed: {e}")
|
||
|
||
def get_algorithms(self, algorithm_type: Optional[str] = None) -> List[Algorithm]:
|
||
"""获取算法列表
|
||
|
||
Args:
|
||
algorithm_type: 算法类型
|
||
|
||
Returns:
|
||
算法列表
|
||
"""
|
||
params = {}
|
||
if algorithm_type:
|
||
params["type"] = algorithm_type
|
||
|
||
response = self._request("GET", "algorithms", params)
|
||
algorithms = []
|
||
for algo_data in response.get("algorithms", []):
|
||
algorithm = Algorithm(**algo_data)
|
||
algorithms.append(algorithm)
|
||
|
||
return algorithms
|
||
|
||
def get_algorithm(self, algorithm_id: str) -> Algorithm:
|
||
"""获取算法详情
|
||
|
||
Args:
|
||
algorithm_id: 算法ID
|
||
|
||
Returns:
|
||
算法详情
|
||
"""
|
||
response = self._request("GET", f"algorithms/{algorithm_id}")
|
||
return Algorithm(**response)
|
||
|
||
def get_algorithm_versions(self, algorithm_id: str) -> List[AlgorithmVersion]:
|
||
"""获取算法版本列表
|
||
|
||
Args:
|
||
algorithm_id: 算法ID
|
||
|
||
Returns:
|
||
版本列表
|
||
"""
|
||
response = self._request("GET", f"algorithms/{algorithm_id}/versions")
|
||
versions = []
|
||
for version_data in response:
|
||
version = AlgorithmVersion(**version_data)
|
||
versions.append(version)
|
||
|
||
return versions
|
||
|
||
def call_algorithm(self, algorithm_id: str, version_id: str, input_data: Dict[str, Any], params: Optional[Dict[str, Any]] = None) -> AlgorithmCallResult:
|
||
"""调用算法
|
||
|
||
Args:
|
||
algorithm_id: 算法ID
|
||
version_id: 版本ID
|
||
input_data: 输入数据
|
||
params: 算法参数
|
||
|
||
Returns:
|
||
调用结果
|
||
"""
|
||
request_data = AlgorithmCallRequest(
|
||
algorithm_id=algorithm_id,
|
||
version_id=version_id,
|
||
input_data=input_data,
|
||
params=params or {}
|
||
)
|
||
|
||
response = self._request("POST", "algorithms/call", request_data.dict())
|
||
return AlgorithmCallResult(**response)
|
||
|
||
def get_call_result(self, call_id: str) -> AlgorithmCallResult:
|
||
"""获取调用结果
|
||
|
||
Args:
|
||
call_id: 调用ID
|
||
|
||
Returns:
|
||
调用结果
|
||
"""
|
||
response = self._request("GET", f"algorithms/calls/{call_id}")
|
||
return AlgorithmCallResult(**response)
|
||
|
||
def generate_simulation_data(self, prompt: str, data_type: str = "text") -> Dict[str, Any]:
|
||
"""生成仿真输入数据
|
||
|
||
Args:
|
||
prompt: 描述
|
||
data_type: 数据类型
|
||
|
||
Returns:
|
||
生成的数据
|
||
"""
|
||
response = self._request("POST", "openai/generate-data", {
|
||
"prompt": prompt,
|
||
"data_type": data_type
|
||
})
|
||
return response
|
||
|
||
def get_call_history(self, algorithm_id: Optional[str] = None, status: Optional[str] = None,
|
||
start_date: Optional[str] = None, end_date: Optional[str] = None) -> List[AlgorithmCallResult]:
|
||
"""获取调用历史
|
||
|
||
Args:
|
||
algorithm_id: 算法ID
|
||
status: 调用状态
|
||
start_date: 开始日期
|
||
end_date: 结束日期
|
||
|
||
Returns:
|
||
调用历史列表
|
||
"""
|
||
params = {}
|
||
if algorithm_id:
|
||
params["algorithm_id"] = algorithm_id
|
||
if status:
|
||
params["status"] = status
|
||
if start_date:
|
||
params["start_date"] = start_date
|
||
if end_date:
|
||
params["end_date"] = end_date
|
||
|
||
response = self._request("GET", "history/user-calls", params)
|
||
history = []
|
||
for call_data in response.get("history", []):
|
||
call_result = AlgorithmCallResult(**call_data)
|
||
history.append(call_result)
|
||
return history
|
||
|
||
def get_call_statistics(self, algorithm_id: Optional[str] = None) -> Dict[str, Any]:
|
||
"""获取调用统计
|
||
|
||
Args:
|
||
algorithm_id: 算法ID
|
||
|
||
Returns:
|
||
统计信息
|
||
"""
|
||
params = {}
|
||
if algorithm_id:
|
||
params["algorithm_id"] = algorithm_id
|
||
|
||
response = self._request("GET", "history/statistics", params)
|
||
return response
|
||
|
||
def export_history(self, algorithm_id: Optional[str] = None, start_date: Optional[str] = None,
|
||
end_date: Optional[str] = None, format_type: str = "json") -> Dict[str, Any]:
|
||
"""导出历史记录
|
||
|
||
Args:
|
||
algorithm_id: 算法ID
|
||
start_date: 开始日期
|
||
end_date: 结束日期
|
||
format_type: 导出格式
|
||
|
||
Returns:
|
||
导出结果
|
||
"""
|
||
params = {
|
||
"format_type": format_type
|
||
}
|
||
if algorithm_id:
|
||
params["algorithm_id"] = algorithm_id
|
||
if start_date:
|
||
params["start_date"] = start_date
|
||
if end_date:
|
||
params["end_date"] = end_date
|
||
|
||
response = self._request("GET", "history/export", params)
|
||
return response
|
||
|
||
def get_algorithm_performance(self, algorithm_id: str, days: int = 7) -> Dict[str, Any]:
|
||
"""获取算法性能指标
|
||
|
||
Args:
|
||
algorithm_id: 算法ID
|
||
days: 统计天数
|
||
|
||
Returns:
|
||
性能指标
|
||
"""
|
||
response = self._request("GET", f"monitoring/performance/algorithm/{algorithm_id}", {"days": days})
|
||
return response
|
||
|
||
def get_system_health(self) -> Dict[str, Any]:
|
||
"""获取系统健康状况
|
||
|
||
Returns:
|
||
系统健康状况
|
||
"""
|
||
response = self._request("GET", "monitoring/health")
|
||
return response
|
||
|
||
def check_permission(self, algorithm_id: str, permission_type: str = "execute") -> bool:
|
||
"""检查权限
|
||
|
||
Args:
|
||
algorithm_id: 算法ID
|
||
permission_type: 权限类型
|
||
|
||
Returns:
|
||
是否有权限
|
||
"""
|
||
response = self._request("POST", "permissions/check", {
|
||
"algorithm_id": algorithm_id,
|
||
"permission_type": permission_type
|
||
})
|
||
return response.get("has_permission", False)
|