qsc20001102/astrbot_plugin_jx3
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

11

Browse Source
This commit is contained in:
qsc20001102
2025-09-27 20:45:08 +08:00
parent 77c0404520
commit eb0e59574a
4 changed files with 289 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/api_data.py
+85 -54
View File
@@ -20,14 +20,14 @@ class APIClient:
self.base_timeout = base_timeout
self.ssl_verify = ssl_verify
async def _make_request(self, method, url, **kwargs):
async def _make_request(self, method, url, params_data=None):
"""
内部请求方法，处理通用请求逻辑
内部请求方法，统一处理请求逻辑
Args:
method: 请求方法 ('GET', 'POST')
url: 请求URL
**kwargs: 其他请求参数
params_data: 统一的参数字典，如 {"name": "万花"}
Returns:
成功时返回解析后的数据，失败返回None
@@ -35,24 +35,77 @@ class APIClient:
timeout = ClientTimeout(total=self.base_timeout)
try:
# 添加请求日志
logger.debug(f"发起 {method} 请求: {url}")
logger.debug(f"参数数据: {params_data}")
async with aiohttp.ClientSession(timeout=timeout) as session:
async with session.request(method, url, ssl=self.ssl_verify, **kwargs) as response:
response.raise_for_status()
data = await response.json()
# 检查API响应状态
return self._check_response_data(data)
if method.upper() == 'GET':
# GET请求：参数作为查询字符串
async with session.get(
url,
params=params_data,
ssl=self.ssl_verify
) as response:
return await self._handle_response(response)
elif method.upper() == 'POST':
# POST请求：参数作为JSON数据
headers = {'Content-Type': 'application/json'}
async with session.post(
url,
json=params_data, # 使用json参数而不是data
headers=headers,
ssl=self.ssl_verify
) as response:
return await self._handle_response(response)
else:
logger.error(f"不支持的HTTP方法: {method}")
return None
except aiohttp.ClientError as e:
logger.error(f"请求出错: {e}")
return None
except json.JSONDecodeError as e:
logger.error(f"JSON解析错误: {e}")
return None
except Exception as e:
logger.error(f"未知错误: {e}")
return None
async def _handle_response(self, response):
"""
处理HTTP响应
Args:
response: aiohttp响应对象
Returns:
解析后的数据或None
"""
try:
# 添加响应日志
logger.debug(f"响应状态: {response.status}")
response.raise_for_status()
data = await response.json()
logger.debug(f"响应数据: {data}")
# 检查API响应状态
return self._check_response_data(data)
except aiohttp.ClientError as e:
logger.error(f"HTTP错误: {e}")
return None
except json.JSONDecodeError as e:
logger.error(f"JSON解析错误: {e}")
# 尝试读取原始文本内容
try:
text_content = await response.text()
logger.error(f"原始响应内容: {text_content}")
except:
pass
return None
def _check_response_data(self, data):
"""
检查API响应数据的通用逻辑
@@ -63,8 +116,16 @@ class APIClient:
Returns:
检查通过返回数据，否则返回None
"""
# 如果数据是字符串，尝试解析为JSON
if isinstance(data, str):
try:
data = json.loads(data)
except json.JSONDecodeError:
logger.error("响应数据是无效的JSON字符串")
return None
# 检查是否有code字段
if 'code' in data:
if data and 'code' in data:
# 有code字段时，检查是否成功
if data.get('code') not in [200, "0", 0]:
logger.error(f"API返回错误:{data.get('code', '未知状态')} {data.get('msg', '未知错误')}")
@@ -77,19 +138,19 @@ class APIClient:
return data
async def post(self, api_url, json_data=None, outdata=None):
async def post(self, api_url, params_data=None, outdata=None):
"""
POST请求方法
Args:
api_url: API地址
json_data: POST请求的JSON数据
params_data: 参数字典，如 {"name": "万花"}
outdata: 返回数据中要提取的字段
Returns:
成功时返回outdata字段的数据，失败返回None
"""
data = await self._make_request('POST', api_url, json=json_data)
data = await self._make_request('POST', api_url, params_data)
if data is None:
return None
@@ -98,42 +159,19 @@ class APIClient:
return data
return data.get(outdata, {})
async def get(self, api_url, params=None, outdata=None):
async def get(self, api_url, params_data=None, outdata=None):
"""
GET请求方法
Args:
api_url: API地址
params: 其他查询参数
params_data: 参数字典，如 {"name": "万花"}
outdata: 返回数据中要提取的字段
Returns:
成功时返回outdata字段的数据，失败返回None
"""
data = await self._make_request('GET', api_url, params=params)
if data is None:
return None
if outdata is None or outdata == "":
return data
return data.get(outdata, {})
async def request(self, api_config, outdata=None):
"""
通用请求方法，根据api_config中的method选择GET或POST
Args:
api_config: 包含url, method, params/json等信息的字典
outdata: 返回数据中要提取的字段
Returns:
成功时返回outdata字段的数据，失败返回None
"""
if 'url' not in api_config or 'method' not in api_config:
logger.error("api_config必须包含url和method字段")
return None
data = await self._make_request(api_config['method'], api_config['url'], params=api_config['params'])
data = await self._make_request('GET', api_url, params_data)
if data is None:
return None
@@ -143,20 +181,13 @@ class APIClient:
return data.get(outdata, {})
# 保持原有函数接口的兼容性（可选）
async def api_data_post(api_url, json_data=None, outdata=None):
# 保持原有函数接口的兼容性
async def api_data_post(api_url, params_data=None, outdata=None):
"""兼容原有函数的POST请求"""
client = APIClient()
return await client.post(api_url, json_data, outdata)
return await client.post(api_url, params_data, outdata)
async def api_data_get(api_url, params=None, outdata=None):
async def api_data_get(api_url, params_data=None, outdata=None):
"""兼容原有函数的GET请求"""
client = APIClient()
return await client.get(api_url, params, outdata)
return await client.get(api_url, params_data, outdata)
core/jx3_Function.py
+121
View File
@@ -0,0 +1,121 @@
from astrbot.api import logger
from astrbot.api.event import filter, AstrMessageEvent, MessageEventResult
from .api_data import APIClient
class JX3Function:
def __init__(self, api_config):
self.__api = APIClient()
self.__api_config = api_config
async def richang(self,server: str = "眉间雪",num: int = 0):
return_data = {
"code": 0,
"msg": "功能函数未执行",
"data": {}
}
#在配置文件中获取接口配置
api_config = self.__api_config["jx3_richang"]
#更新参数
api_config["params"]["server"] = server
api_config["params"]["num"] = num
# 获取数据
data = await self.__api.get(api_config["url"],api_config["params"],"data")
if not data:
return_data["msg"] = "获取接口信息失败"
return return_data
# 处理返回数据
try:
result_msg = f"{server}-{data.get('date')}-星期{data.get('week')}\n"
result_msg += f"大战：{data.get('war')}\n"
result_msg += f"战场：{data.get('battle')}\n"
result_msg += f"阵营：{data.get('orecar')}\n"
result_msg += f"宗门：{data.get('school')}\n"
result_msg += f"驰援：{data.get('rescue')}\n"
result_msg += f"画像：{data.get('draw')}\n"
result_msg += f"宠物福缘：\n{data.get('luck')}\n"
result_msg += f"家园声望：\n{data.get('card')}\n"
result_msg += f"武林通鉴：\n{data.get('team')}\n"
return_data["data"] = result_msg
return_data["code"] = 200
except Exception as e:
logger.error(f"处理数据时出错: {e}")
return_data["msg"] = "处理接口返回信息时出错"
return return_data
async def shapan(self,server: str = "眉间雪"):
return_data = {
"code": 0,
"msg": "功能函数未执行",
"data": {}
}
#在配置文件中获取接口配置
api_config = self.__api_config["aijx3_shapan"]
#更新参数
api_config["params"]["serverName"] = server
# 获取数据
data = await self.__api.post(api_config["url"],api_config["params"],"data")
if not data:
return_data["msg"] = "获取接口信息失败"
return return_data
# 处理返回数据
try:
return_data["data"] = data.get("picUrl")
return_data["code"] = 200
except Exception as e:
logger.error(f"处理数据时出错: {e}")
return_data["msg"] = "处理接口返回信息时出错"
return return_data
async def shaohua(self):
return_data = {
"code": 0,
"msg": "功能函数未执行",
"data": {}
}
#在配置文件中获取接口配置
api_config = self.__api_config["jx3_shaohua"]
# 获取数据
data = await self.__api.get(api_config["url"],api_config["params"],"data")
if not data:
return_data["msg"] = "获取接口信息失败"
return return_data
# 处理返回数据
try:
return_data["data"] = data.get("text")
return_data["code"] = 200
except Exception as e:
logger.error(f"处理数据时出错: {e}")
return_data["msg"] = "处理接口返回信息时出错"
return return_data
async def jigai(self):
return_data = {
"code": 0,
"msg": "功能函数未执行",
"data": {}
}
#在配置文件中获取接口配置
api_config = self.__api_config["jx3_jigai"]
# 获取数据
data = await self.__api.get(api_config["url"],api_config["params"],"data")
if not data:
return_data["msg"] = "获取接口信息失败"
return return_data
# 处理返回数据
try:
result_msg = f"剑网三最近技改\n"
for i, item in enumerate(data[:2], 1):
result_msg += f"{i}. {item.get('title', '无标题')}\n"
result_msg += f"时间：{item.get('time', '未知时间')}\n"
result_msg += f"链接：{item.get('url', '无链接')}\n\n"
return_data["data"] = result_msg
return_data["code"] = 200
except Exception as e:
logger.error(f"处理数据时出错: {e}")
return_data["msg"] = "处理接口返回信息时出错"
return return_data