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-26 21:04:08 +08:00
parent cd49f2b0d0
commit 3368baa0cb
3 changed files with 174 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api_config.json
+11
View File
@@ -0,0 +1,11 @@
{
"jx3_richang":{
"url":"https://www.jx3api.com/data/active/calendar",
"method":"GET",
"description":"获取剑网三日常任务",
"params":{
"server": "眉间雪",
"num": 0
}
}
}
core/api_data.py
+123 -45
View File
@@ -1,11 +1,85 @@
# core/request.py
import requests
import aiohttp
import json
from aiohttp import ClientTimeout
from astrbot.api import logger
def api_data_post(api_url, json_data=None, outdata=None):
class APIClient:
"""
获取数据的POST请求函数
API客户端类，封装GET和POST请求功能
"""
def __init__(self, base_timeout=10, ssl_verify=False):
"""
初始化APIClient
Args:
base_timeout: 默认超时时间（秒）
ssl_verify: SSL证书验证开关
"""
self.base_timeout = base_timeout
self.ssl_verify = ssl_verify
async def _make_request(self, method, url, **kwargs):
"""
内部请求方法，处理通用请求逻辑
Args:
method: 请求方法 ('GET', 'POST')
url: 请求URL
**kwargs: 其他请求参数
Returns:
成功时返回解析后的数据，失败返回None
"""
timeout = ClientTimeout(total=self.base_timeout)
try:
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)
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
def _check_response_data(self, data):
"""
检查API响应数据的通用逻辑
Args:
data: API返回的数据
Returns:
检查通过返回数据，否则返回None
"""
# 检查是否有code字段
if 'code' in data:
# 有code字段时，检查是否成功
if data.get('code') not in [200, "0", 0]:
logger.error(f"API返回错误:{data.get('code', '未知状态')} {data.get('msg', '未知错误')}")
return None
else:
# 无code字段时，检查数据是否为空
if not data:
logger.error("API返回空数据")
return None
return data
async def post(self, api_url, json_data=None, outdata=None):
"""
POST请求方法
Args:
api_url: API地址
@@ -15,70 +89,74 @@ def api_data_post(api_url, json_data=None, outdata=None):
Returns:
成功时返回outdata字段的数据，失败返回None
"""
try:
response = requests.post(api_url, json=json_data, timeout=10, verify=False)
response.raise_for_status()
data = response.json()
data = await self._make_request('POST', api_url, json=json_data)
# 检查是否有code字段
if 'code' in data:
# 有code字段时，检查是否成功
if data.get('code') not in [200, "0", 0]:
logger.error(f"API返回错误:{data.get('code', '未知状态')} {data.get('msg', '未知错误')}")
return None
else:
# 无code字段时，检查数据是否为空
if not data:
logger.error("API返回空数据")
if data is None:
return None
if outdata is None or outdata == "":
return data
return data.get(outdata, {})
except requests.exceptions.RequestException as e:
logger.error(f"请求出错: {e}")
return None
except json.JSONDecodeError as e:
logger.error(f"JSON解析错误: {e}")
return None
def api_data_get(api_url,params=None,outdata=None):
async def get(self, api_url, params=None, outdata=None):
"""
获取数据的GET请求函数
GET请求方法
Args:
api_url: API地址
params: 其他查询参数
outdata: 返回数据中要提取的字段
Returns:
成功时返回outdata字段的数据，失败返回None
"""
try:
response = requests.get(api_url, params, timeout=10, verify=False)
response.raise_for_status()
data = response.json()
data = await self._make_request('GET', api_url, params=params)
# 检查是否有code字段
if 'code' in data:
# 有code字段时，检查是否成功
if data.get('code') not in [200, "0", 0]:
logger.error(f"API返回错误:{data.get('code', '未知状态')} {data.get('msg', '未知错误')}")
return None
else:
# 无code字段时，检查数据是否为空
if not data:
logger.error("API返回空数据")
if data is None:
return None
if outdata is None or outdata == "":
return data
return data.get(outdata, {})
except requests.exceptions.RequestException as e:
logger.error(f"请求出错: {e}")
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
except json.JSONDecodeError as e:
logger.error(f"JSON解析错误: {e}")
data = await self._make_request(api_config['method'], api_config['url'], api_config['params'])
if data is None:
return None
if outdata is None or outdata == "":
return data
return data.get(outdata, {})
# 保持原有函数接口的兼容性（可选）
async def api_data_post(api_url, json_data=None, outdata=None):
"""兼容原有函数的POST请求"""
client = APIClient()
return await client.post(api_url, json_data, outdata)
async def api_data_get(api_url, params=None, outdata=None):
"""兼容原有函数的GET请求"""
client = APIClient()
return await client.get(api_url, params, outdata)
main.py
+31 -17
View File
@@ -1,29 +1,45 @@
import urllib3
from pathlib import Path
from datetime import datetime
from astrbot.api.event import filter, AstrMessageEvent, MessageEventResult
from astrbot.api.star import Context, Star, register
from astrbot.api.star import Context, Star, register, StarTools
from astrbot.api import logger
from astrbot.api import AstrBotConfig
from .core.load_template import load_template
from .core.jx3_data import jx3_data_jiaoyihang,jx3_data_wujia
from .core.api_data import api_data_get, api_data_post
from .core.api_data import APIClient,api_data_get, api_data_post
from .core.sql_data import sql_data_searchdata,sql_data_select
# 禁用 SSL 警告
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
@register("jx3api", "fxdyz", "通过接口调用剑网三API接口获取游戏数据", "1.0.0")
@register("astrbot_plugin_jx3api",
"fxdyz",
"通过接口调用剑网三API接口获取游戏数据",
"1.0.0",
"https://github.com/qsc20001102/astrbot_plugin_jx3api.git"
)
class Jx3ApiPlugin(Star):
def __init__(self, context: Context, config: AstrBotConfig):
super().__init__(context)
#获取配置
self.conf = config
print(self.conf)
# 本地数据存储路径
self.local_data_dir = StarTools.get_data_dir("astrbot_plugin_jx3api")
# api数据文件
self.api_file = Path(__file__).parent / "api_config.json"
logger.info("jx3api插件初始化完成")
async def initialize(self):
"""可选择实现异步的插件初始化方法，当实例化该插件类之后会自动调用该方法。"""
#创建类实例
self.api_data = APIClient()
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):
@@ -32,25 +48,23 @@ class Jx3ApiPlugin(Star):
custom_url = "https://www.jx3api.com/data/active/calendar"
# 接口参数
params = {
"server": "眉间雪", # 默认服务器
"num": 0 # 默认当天
"server": "眉间雪", # 服务器
"num": 0 # 当天
}
# 获取消息内容
message_str = event.message_str.strip()
parts = message_str.split()
parts = await self.message_str(event)
# 解析消息内容
if len(parts) > 1:
params["server"] = parts[1] # 第二个参数为服务器
params["server"] = parts[1]
if len(parts) > 2:
try:
params["num"] = int(parts[2]) # 第三个参数为日期偏移
params["num"] = int(parts[2])
except ValueError:
params["num"] = 0 # 参数为日期偏移
params["num"] = 0
# 获取数据
data = api_data_get(custom_url, params,"data")
data = await self.api_data.request(self.api_file['jx3_richang'], params,"data")
if not data:
yield event.plain_result("获取获取接口信息失败，请稍后再试")