"""智能算法展示平台客户端""" 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:8001/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)