135 lines
4.0 KiB
Python
135 lines
4.0 KiB
Python
import requests
|
||
import json
|
||
from .config_loader import ConfigLoader
|
||
|
||
class ApiUtils:
|
||
def __init__(self):
|
||
"""初始化 API 工具类"""
|
||
self.config_loader = ConfigLoader.get_instance()
|
||
self.base_url = self.config_loader.get_value("app.base_url", "")
|
||
self.apis = self.config_loader.get_value("apis", {})
|
||
|
||
def get_full_url(self, url):
|
||
"""
|
||
根据 API 键获取完整 URL
|
||
|
||
Args:
|
||
url: API 配置中的键名
|
||
|
||
Returns:
|
||
str: 完整的 URL
|
||
"""
|
||
if url not in self.apis:
|
||
return None
|
||
|
||
return f"{self.base_url}{self.apis[url]}"
|
||
|
||
def request(self, method, url, params=None, data=None, json_data=None, headers=None):
|
||
"""
|
||
发送 HTTP 请求
|
||
|
||
Args:
|
||
method: 请求方法，如 'GET', 'POST' 等
|
||
url: API 配置中的键名
|
||
params: URL 参数
|
||
data: 表单数据
|
||
json_data: JSON 数据
|
||
headers: 请求头
|
||
|
||
Returns:
|
||
dict: 响应数据
|
||
"""
|
||
full_url = self.get_full_url(url)
|
||
if not full_url:
|
||
return {"success": False, "message": f"未找到 API 配置: {url}"}
|
||
|
||
# 处理托盘号参数
|
||
# 如果是托盘查询接口，并且有tp_note参数，将其附加到URL末尾
|
||
if url == "get_tray_info" and params and "tp_note" in params:
|
||
full_url = f"{full_url}{params['tp_note']}"
|
||
# 从params中移除tp_note，因为已经添加到URL中
|
||
params = {k: v for k, v in params.items() if k != "tp_note"}
|
||
|
||
default_headers = {
|
||
"Content-Type": "application/json"
|
||
}
|
||
|
||
if headers:
|
||
default_headers.update(headers)
|
||
|
||
try:
|
||
response = requests.request(
|
||
method=method,
|
||
url=full_url,
|
||
params=params,
|
||
data=data,
|
||
json=json_data,
|
||
headers=default_headers
|
||
)
|
||
|
||
response.raise_for_status()
|
||
return response.json()
|
||
except requests.exceptions.RequestException as e:
|
||
return {"success": False, "message": f"请求异常: {str(e)}"}
|
||
except json.JSONDecodeError:
|
||
return {"success": False, "message": "响应数据不是有效的 JSON 格式"}
|
||
|
||
def get(self, url, params=None, headers=None):
|
||
"""
|
||
发送 GET 请求
|
||
|
||
Args:
|
||
url: API 配置中的键名
|
||
params: URL 参数
|
||
headers: 请求头
|
||
|
||
Returns:
|
||
dict: 响应数据
|
||
"""
|
||
return self.request("GET", url, params=params, headers=headers)
|
||
|
||
def post(self, url, json_data=None, data=None, headers=None):
|
||
"""
|
||
发送 POST 请求
|
||
|
||
Args:
|
||
url: API 配置中的键名
|
||
json_data: JSON 数据
|
||
data: 表单数据
|
||
headers: 请求头
|
||
|
||
Returns:
|
||
dict: 响应数据
|
||
"""
|
||
return self.request("POST", url, data=data, json_data=json_data, headers=headers)
|
||
|
||
def put(self, url, json_data=None, data=None, headers=None):
|
||
"""
|
||
发送 PUT 请求
|
||
|
||
Args:
|
||
url: API 配置中的键名
|
||
json_data: JSON 数据
|
||
data: 表单数据
|
||
headers: 请求头
|
||
|
||
Returns:
|
||
dict: 响应数据
|
||
"""
|
||
return self.request("PUT", url, data=data, json_data=json_data, headers=headers)
|
||
|
||
def delete(self, url, params=None, headers=None):
|
||
"""
|
||
发送 DELETE 请求
|
||
|
||
Args:
|
||
url: API 配置中的键名
|
||
params: URL 参数
|
||
headers: 请求头
|
||
|
||
Returns:
|
||
dict: 响应数据
|
||
"""
|
||
return self.request("DELETE", url, params=params, headers=headers)
|
||
|