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:
qsc
2025-09-29 11:52:41 +08:00
parent e1ced9098f
commit 4398ad32bc
4 changed files with 278 additions and 205 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api_config.json
+24
View File
@@ -7,5 +7,29 @@
"server": "眉间雪",
"num": 0
}
},
"jx3_shaohua":{
"url":"https://www.jx3api.com/data/saohua/random",
"method":"GET",
"description":"获取剑网三骚话",
"params":{
"name": "万花"
}
},
"jx3_jigai":{
"url":"https://www.jx3api.com/data/skills/records",
"method":"GET",
"description":"获取剑网三技能记录",
"params":{
}
},
"aijx3_shapan":{
"url":"https://www.jianxiachaguan.cn/api2/aijx3-jxcg/game/get-sand-table-img",
"method":"POST",
"description":"获取剑网三沙盘状态",
"params":{
"serverName": "眉间雪"
}
}
}
core/api_data.py
+81 -49
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,22 +35,75 @@ 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:
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 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"请求出错: {e}")
logger.error(f"HTTP错误: {e}")
return None
except json.JSONDecodeError as e:
logger.error(f"JSON解析错误: {e}")
return None
except Exception as e:
logger.error(f"未知错误: {e}")
# 尝试读取原始文本内容
try:
text_content = await response.text()
logger.error(f"原始响应内容: {text_content}")
except:
pass
return None
def _check_response_data(self, data):
@@ -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,19 +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
main.py
+47 -151
View File
@@ -11,6 +11,7 @@ from .core.load_template import load_template
from .core.jx3_data import jx3_data_jiaoyihang,jx3_data_wujia
from .core.api_data import APIClient,api_data_get, api_data_post
from .core.sql_data import sql_data_searchdata,sql_data_select
from .core.jx3_Function import JX3Function
@register("astrbot_plugin_jx3api",
@@ -36,32 +37,43 @@ class Jx3ApiPlugin(Star):
async def initialize(self):
"""可选择实现异步的插件初始化方法，当实例化该插件类之后会自动调用该方法。"""
#创建类实例
self.api_data = APIClient()
self.jx3fun = JX3Function(self.api_config)
logger.info("jx3api插件创建实例完成")
async def message_str(self, event: AstrMessageEvent):
"""可选择实现异步的消息预处理方法，在消息分发到具体命令前会调用该方法。"""
# 预处理消息内容
message_str = event.message_str.strip()
return message_str.split()
@filter.command("剑三日常")
async def jx3_richang(self, event: AstrMessageEvent):
"""剑三日常"""
# 接口URL
custom_url = "https://www.jx3api.com/data/active/calendar"
# 接口参数
params = {
"server": "眉间雪", # 服务器
"num": 0 # 当天
}
# 获取消息内容
parts = await self.message_str(event)
# 解析消息内容
if len(parts) > 1:
params["server"] = parts[1]
<<<<<<< HEAD
@filter.command_group("剑三")
def jx3(self):
pass
@jx3.command("日常")
async def jx3_richang(self, event: AstrMessageEvent,server: str = "眉间雪",num: int = 0):
"""剑三 日常 服务器 天数"""
try:
data= await self.jx3fun.richang(server,num)
if data["code"] == 200:
yield event.plain_result(data["data"])
else:
yield event.plain_result("msg")
return
except Exception as e:
logger.error(f"功能函数执行错误: {e}")
yield event.plain_result("猪脑过载，请稍后再试")
@jx3.command("沙盘")
async def jx3_shapan(self, event: AstrMessageEvent,server: str = "眉间雪"):
"""剑三 沙盘 服务器"""
try:
data= await self.jx3fun.shapan(server)
if data["code"] == 200:
yield event.image_result(data["data"])
else:
yield event.plain_result(data["msg"])
return
except Exception as e:
logger.error(f"功能函数执行错误: {e}")
yield event.plain_result("猪脑过载，请稍后再试")
@jx3.command("骚话")
async def jx3_shaohua(self, event: AstrMessageEvent,):
@@ -72,110 +84,25 @@ class Jx3ApiPlugin(Star):
yield event.plain_result(data["data"])
else:
yield event.plain_result(data["msg"])
=======
if len(parts) > 2:
try:
params["num"] = int(parts[2])
except ValueError:
params["num"] = 0
# 获取数据
data = await self.api_data.request(self.api_config["jx3_richang"],"data")
if not data:
yield event.plain_result("获取获取接口信息失败，请稍后再试")
>>>>>>> ecb573761ec04986a07da4a19511ef52e9f36ec6
return
# 处理返回数据
try:
# 构建回复消息
result_msg = f"{params['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"
yield event.plain_result(result_msg)
except Exception as e:
logger.error(f"处理数据时出错: {e}")
yield event.plain_result("处理接口返回信息时出错")
logger.error(f"功能函数执行错误: {e}")
yield event.plain_result("猪脑过载，请稍后再试")
@filter.command("剑三骚话")
async def jx3_shaohua(self, event: AstrMessageEvent):
"""剑三骚话"""
# 接口URL
custom_url = "https://www.jx3api.com/data/saohua/random"
# 接口参数
params = {
"name": "万花", # 默认值
}
# 获取消息内容
# 解析消息内容
# 获取数据
data = api_data_get(custom_url, params,"data")
if not data:
yield event.plain_result("获取获取接口信息失败，请稍后再试")
@jx3.command("技改")
async def jx3_jigai(self, event: AstrMessageEvent,):
"""剑三 技改"""
try:
data= await self.jx3fun.jigai()
if data["code"] == 200:
yield event.plain_result(data["data"])
else:
yield event.plain_result(data["msg"])
return
# 处理返回数据
try:
# 构建回复消息
result_msg = f"{data.get('text')}\n"
yield event.plain_result(result_msg)
except Exception as e:
logger.error(f"处理数据时出错: {e}")
yield event.plain_result("处理接口返回信息时出错")
@filter.command("剑三技改")
async def jx3_jigai(self, event: AstrMessageEvent):
"""剑三技改"""
# 接口URL
custom_url = "https://www.jx3api.com/data/skills/records"
# 接口参数
params = {
}
# 获取消息内容
# 解析消息内容
# 获取数据
data = api_data_get(custom_url,params,"data")
if not data:
yield event.plain_result("获取获取接口信息失败，请稍后再试")
return
# 处理返回数据
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"
yield event.plain_result(result_msg)
except Exception as e:
logger.error(f"处理数据时出错: {e}")
yield event.plain_result("处理接口返回信息时出错")
logger.error(f"功能函数执行错误: {e}")
yield event.plain_result("猪脑过载，请稍后再试")
@filter.command("剑三交易行")
@@ -234,38 +161,7 @@ class Jx3ApiPlugin(Star):
yield event.plain_result("查询交易行数据时出错，请稍后再试")
@filter.command("剑三沙盘")
async def jx3_shapan(self, event: AstrMessageEvent):
"""剑三沙盘 服务器"""
# 接口URL
custom_url = "https://www.jianxiachaguan.cn/api2/aijx3-jxcg/game/get-sand-table-img"
# 接口参数
params = {
"serverName": "眉间雪" # 默认服务器
}
# 获取消息内容
message_str = event.message_str.strip()
parts = message_str.split()
# 解析消息内容
if len(parts) > 1:
params["serverName"] = parts[1] # 第二个参数为服务器
# 获取数据
data = api_data_post(custom_url,params,"data")
if not data:
yield event.plain_result("获取获取接口信息失败，请稍后再试")
return
# 处理返回数据
try:
# 构建回复消息
yield event.image_result(data.get("picUrl"))
except Exception as e:
logger.error(f"处理数据时出错: {e}")
yield event.plain_result("处理接口返回信息时出错")
@filter.command("剑三金价")