11
This commit is contained in:
@@ -1,150 +0,0 @@
|
||||
# core/request.py
|
||||
import aiohttp
|
||||
import json
|
||||
from aiohttp import ClientTimeout
|
||||
from astrbot.api import logger
|
||||
|
||||
class APIClient:
|
||||
"""
|
||||
API客户端类，支持GET/POST请求，同时兼容JSON和二进制数据
|
||||
"""
|
||||
|
||||
def __init__(self, base_timeout=10, ssl_verify=False):
|
||||
self.base_timeout = base_timeout
|
||||
self.ssl_verify = ssl_verify
|
||||
|
||||
async def _make_request(self, method, url, params_data=None):
|
||||
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:
|
||||
if method.upper() == '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':
|
||||
headers = {'Content-Type': 'application/json'}
|
||||
async with session.post(url, json=params_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):
|
||||
"""
|
||||
处理响应：支持 JSON 和二进制数据
|
||||
"""
|
||||
try:
|
||||
logger.debug(f"响应状态: {response.status}")
|
||||
response.raise_for_status()
|
||||
content_type = response.headers.get('Content-Type', '')
|
||||
|
||||
if 'image' in content_type or 'octet-stream' in content_type:
|
||||
# 返回二进制数据
|
||||
data = await response.read()
|
||||
return data
|
||||
|
||||
# 尝试解析为 JSON
|
||||
try:
|
||||
data = await response.json(content_type=None)
|
||||
except Exception:
|
||||
text = await response.text()
|
||||
logger.debug(f"原始文本响应: {text}")
|
||||
try:
|
||||
data = json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
logger.error("无法解析为 JSON 数据")
|
||||
return None
|
||||
|
||||
logger.debug(f"响应数据: {data}")
|
||||
return self._check_response_data(data)
|
||||
except aiohttp.ClientError as e:
|
||||
logger.error(f"HTTP错误: {e}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"未知错误: {e}")
|
||||
return None
|
||||
|
||||
def _check_response_data(self, data):
|
||||
"""
|
||||
检查 API 返回 JSON 数据
|
||||
"""
|
||||
if isinstance(data, str):
|
||||
try:
|
||||
data = json.loads(data)
|
||||
except json.JSONDecodeError:
|
||||
logger.error("响应数据是无效的JSON字符串")
|
||||
return None
|
||||
|
||||
if data and isinstance(data, dict) and 'code' in data:
|
||||
if data.get('code') not in [200, "0", 0, 1]:
|
||||
logger.error(f"API返回错误:{data.get('code', '未知状态')} {data.get('msg', '未知错误')}")
|
||||
return None
|
||||
elif not data:
|
||||
logger.error("API返回空数据")
|
||||
return None
|
||||
|
||||
return data
|
||||
|
||||
async def post(self, api_url, params_data=None, outdata=None):
|
||||
data = await self._make_request('POST', api_url, params_data)
|
||||
if data is None:
|
||||
return None
|
||||
if isinstance(data, bytes):
|
||||
return data
|
||||
if not outdata:
|
||||
return data
|
||||
return data.get(outdata, {})
|
||||
|
||||
async def get(self, api_url, params_data=None, outdata=None):
|
||||
data = await self._make_request('GET', api_url, params_data)
|
||||
if data is None:
|
||||
return None
|
||||
if isinstance(data, bytes):
|
||||
return data
|
||||
if not outdata:
|
||||
return data
|
||||
return data.get(outdata, {})
|
||||
|
||||
async def all_pages(self, http, api_url, params_data=None, outdata: str = "", listdata: str = "list", max_pages: int = 10):
|
||||
all_data = []
|
||||
current_page = 1
|
||||
while True:
|
||||
params = params_data.copy() if params_data else {}
|
||||
params["page"] = str(current_page)
|
||||
|
||||
if http.upper() == "POST":
|
||||
data = await self.post(api_url, params, outdata)
|
||||
else:
|
||||
data = await self.get(api_url, params, outdata)
|
||||
|
||||
if not data or isinstance(data, bytes):
|
||||
# 二进制数据或者空数据，不分页
|
||||
break
|
||||
|
||||
if not data.get(listdata):
|
||||
break
|
||||
|
||||
all_data.extend(data[listdata])
|
||||
|
||||
if max_pages and current_page >= max_pages:
|
||||
break
|
||||
|
||||
current_page += 1
|
||||
logger.info(f"已获取第 {current_page} 页数据")
|
||||
return all_data
|
||||
|
||||
# 保持原接口兼容
|
||||
async def api_data_post(api_url, params_data=None, outdata=None):
|
||||
client = APIClient()
|
||||
return await client.post(api_url, params_data, outdata)
|
||||
|
||||
async def api_data_get(api_url, params_data=None, outdata=None):
|
||||
client = APIClient()
|
||||
return await client.get(api_url, params_data, outdata)
|
||||
@@ -1,536 +0,0 @@
|
||||
from datetime import datetime
|
||||
|
||||
from astrbot.api import logger
|
||||
|
||||
from .APIClient import APIClient
|
||||
from .AsyncMySQL import AsyncMySQL
|
||||
from .function_basic import load_template,extract_field,flatten_field,extract_fields,gold_to_string,plot_line_chart_base64
|
||||
|
||||
class JX3Function:
|
||||
def __init__(self, api_config,db: AsyncMySQL ):
|
||||
self.__api = APIClient()
|
||||
self.__db = db
|
||||
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"宠物福缘：{data.get('luck')[0]},{data.get('luck')[1]},{data.get('luck')[2]}\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 kaifu(self,server: str = "梦江南"):
|
||||
"""
|
||||
开服状态查询
|
||||
"""
|
||||
return_data = {
|
||||
"code": 0,
|
||||
"msg": "功能函数未执行",
|
||||
"data": {}
|
||||
}
|
||||
#在配置文件中获取接口配置
|
||||
api_config = self.__api_config["jx3_kaifu"]
|
||||
#更新参数
|
||||
api_config["params"]["server"] = server
|
||||
# 获取数据
|
||||
data = await self.__api.get(api_config["url"],api_config["params"],"data")
|
||||
if not data:
|
||||
return_data["msg"] = "获取接口信息失败"
|
||||
return return_data
|
||||
# 处理返回数据
|
||||
try:
|
||||
status = data.get("status")
|
||||
status_time = datetime.fromtimestamp(data.get("time")).strftime("%Y-%m-%d %H:%M:%S")
|
||||
if status == 1:
|
||||
status_str = f"{server}服务器已开服，快冲，快冲！\n开服时间：{status_time}"
|
||||
status_bool = True
|
||||
else:
|
||||
status_str = f"{server}服务器当前维护中，等会再来吧！\n维护时间：{status_time}"
|
||||
status_bool = False
|
||||
return_data["status"] = status_bool
|
||||
return_data["data"] = status_str
|
||||
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[:1], 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
|
||||
|
||||
|
||||
async def jinjia(self, server: str = "眉间雪"):
|
||||
"""
|
||||
区服金价
|
||||
"""
|
||||
return_data = {
|
||||
"code": 0,
|
||||
"msg": "功能函数未执行",
|
||||
"data": {}
|
||||
}
|
||||
#在配置文件中获取接口配置
|
||||
api_config = self.__api_config["aijx3_jinjia"]
|
||||
#更新参数
|
||||
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
|
||||
chart_data = data
|
||||
data = data[:15]
|
||||
# 加载模板
|
||||
try:
|
||||
return_data["temp"] = load_template("jinjia.html")
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"加载模板失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
return return_data
|
||||
# 生成折线图
|
||||
try:
|
||||
chart_base64 = plot_line_chart_base64(chart_data, "date", "priceWanbaolou", "万宝楼金价走势",True)
|
||||
except Exception as e:
|
||||
logger.error(f"生成折线图失败: {e}")
|
||||
return_data["msg"] = "系统错误：生成折线图失败"
|
||||
return return_data
|
||||
# 准备模板渲染数据
|
||||
try:
|
||||
|
||||
return_data["data"] = {
|
||||
"items": data,
|
||||
"server": api_config["params"]["serverName"],
|
||||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"img": chart_base64
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"处理数据时出错: {e}")
|
||||
return_data["msg"] = "系统错误：模板渲染数据准备失败"
|
||||
return return_data
|
||||
return_data["code"] = 200
|
||||
return return_data
|
||||
|
||||
|
||||
async def qiyu(self, adventureName: str = "阴阳两界", serverName: str = "眉间雪"):
|
||||
"""
|
||||
区服奇遇
|
||||
"""
|
||||
return_data = {
|
||||
"code": 0,
|
||||
"msg": "功能函数未执行",
|
||||
"data": {}
|
||||
}
|
||||
#在配置文件中获取接口配置
|
||||
api_config = self.__api_config["aijx3_qiyu"]
|
||||
#更新参数
|
||||
api_config["params"]["adventureName"] = adventureName
|
||||
api_config["params"]["serverName"] = serverName
|
||||
# 获取数据
|
||||
data = await self.__api.post(api_config["url"],api_config["params"],"data")
|
||||
if not data:
|
||||
return_data["msg"] = "获取接口信息失败"
|
||||
return return_data
|
||||
# 格式化时间
|
||||
for item in data:
|
||||
if "time" in item:
|
||||
item["time"] = datetime.fromtimestamp(item["time"]/1000).strftime("%Y-%m-%d %H:%M:%S")
|
||||
# 加载模板
|
||||
try:
|
||||
return_data["temp"] = load_template("qiyuliebiao.html")
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"加载模板失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
return return_data
|
||||
# 准备模板渲染数据
|
||||
try:
|
||||
return_data["data"] = {
|
||||
"items": data,
|
||||
"server": serverName,
|
||||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
,"qiyuname": adventureName
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"处理数据时出错: {e}")
|
||||
return_data["msg"] = "系统错误：模板渲染数据准备失败"
|
||||
return return_data
|
||||
return_data["code"] = 200
|
||||
return return_data
|
||||
|
||||
|
||||
async def SearchData(self):
|
||||
"""
|
||||
外观数据插入数据库
|
||||
"""
|
||||
return_data = {
|
||||
"code": 0,
|
||||
"msg": "功能函数未执行",
|
||||
"data": {}
|
||||
}
|
||||
#在配置文件中获取接口配置
|
||||
api_config = self.__api_config["aijx3_SearchData"]
|
||||
# 获取数据
|
||||
data = await self.__api.post(api_config["url"],api_config["params"],"data")
|
||||
if not data:
|
||||
return_data["msg"] = "获取接口信息失败"
|
||||
return return_data
|
||||
# 提取数据
|
||||
try:
|
||||
extracted_data = flatten_field(data, "dataModels")
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"提取数据失败: {e}")
|
||||
return_data["msg"] = "提取指定数据字段失败"
|
||||
return return_data
|
||||
# 准备SQL和批量数据
|
||||
sql = """
|
||||
INSERT INTO searchdata
|
||||
(typeName, name, showName, picUrl, searchId, searchDescType)
|
||||
VALUES (%s, %s, %s, %s, %s, %s)
|
||||
"""
|
||||
values_list = [
|
||||
(
|
||||
item['typeName'],
|
||||
item['name'],
|
||||
item['showName'],
|
||||
item['picUrl'],
|
||||
item['searchId'],
|
||||
item['searchDescType']
|
||||
)
|
||||
for item in extracted_data
|
||||
]
|
||||
# 插入数据
|
||||
try:
|
||||
await self.__db.truncate_table("searchdata")
|
||||
await self.__db.executemany(sql, values_list)
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"数据插入失败: {e}")
|
||||
return_data["msg"] = "数据插入数据库失败"
|
||||
return return_data
|
||||
return_data["msg"] = f"成功批量插入 {len(extracted_data)} 条数据！"
|
||||
return_data["code"] = 200
|
||||
return return_data
|
||||
|
||||
|
||||
async def wujia(self,Name: str = "秃盒"):
|
||||
"""
|
||||
物价查询
|
||||
"""
|
||||
return_data = {
|
||||
"code": 0,
|
||||
"msg": "功能函数未执行",
|
||||
"data": {}
|
||||
}
|
||||
#在配置文件中获取接口配置
|
||||
api_config = self.__api_config["jx3box_exterior"]
|
||||
#更新参数
|
||||
api_config["params"]["keyword"] = Name
|
||||
# 获取查找信息
|
||||
data = await self.__api.get(api_config["url"],api_config["params"],"data")
|
||||
if data["total"] == 0:
|
||||
try:
|
||||
sql = "SELECT showName, searchId FROM searchdata WHERE name = %s OR showName = %s"
|
||||
sqldata = await self.__db.fetch_one(sql, (Name,Name))
|
||||
searchId=sqldata["searchId"]
|
||||
showName=sqldata["showName"]
|
||||
return_data["data"]["searchId"]=searchId
|
||||
return_data["data"]["showName"]=showName
|
||||
except Exception as e:
|
||||
logger.error(f"获取外观信息错误: {e}")
|
||||
return_data["msg"] = "未找到该外观信息"
|
||||
return return_data
|
||||
else:
|
||||
# 查询魔盒获取外观名称
|
||||
try:
|
||||
showName=data["list"][0]["name"]
|
||||
return_data["data"]["showName"]=showName
|
||||
except Exception as e:
|
||||
logger.error(f"获取外观名称错误: {e}")
|
||||
return_data["msg"] = "未找到该外观信息"
|
||||
return return_data
|
||||
# 查询爱剑三获取外观ID
|
||||
try:
|
||||
sql = "SELECT searchId FROM searchdata WHERE showName=%s"
|
||||
sqldata = await self.__db.fetch_one(sql, (return_data["data"]["showName"],))
|
||||
searchId=sqldata["searchId"]
|
||||
return_data["data"]["searchId"]=searchId
|
||||
except Exception as e:
|
||||
logger.error(f"获取外观ID错误: {e}\n{return_data['data']['showName']}")
|
||||
return_data["msg"] = "未找到该外观信息"
|
||||
return return_data
|
||||
|
||||
#在配置文件中获取接口配置
|
||||
api_config = self.__api_config["aijx3_GoodsDetail"]
|
||||
#更新参数
|
||||
api_config["params"]["goodsName"] = showName
|
||||
# 获取数据外观详细数据
|
||||
data = await self.__api.post(api_config["url"],api_config["params"],"data")
|
||||
if not data:
|
||||
return_data["msg"] = "获取外观详细数据失败"
|
||||
return return_data
|
||||
# 提取外观详细数据
|
||||
try:
|
||||
imgs = data.get("imgs", [])
|
||||
return_data["data"]["goodsDesc"]=data.get("goodsDesc", "无描述")
|
||||
return_data["data"]["publishTime"]=data.get("publishTime", "无价格")
|
||||
return_data["data"]["priceNum"]=data.get("priceNum", 0)
|
||||
return_data["data"]["goodsId"]=data.get("goodsId", "无数据")
|
||||
return_data["data"]["imgs"]=imgs[0] if imgs else ""
|
||||
return_data["data"]["goodsAlias"]=data.get("goodsAlias", "无别名")
|
||||
except Exception as e:
|
||||
logger.error(f"提取外观详细数据失败: {e}")
|
||||
return_data["msg"] = "提取外观详细数据失败"
|
||||
return return_data
|
||||
|
||||
# 查询万宝楼数据（公示和在售）
|
||||
wbl_data = await self.__get_wbl_data(showName)
|
||||
if not wbl_data:
|
||||
return_data["msg"] = "获取万宝楼数据失败"
|
||||
return return_data
|
||||
if wbl_data:
|
||||
return_data["data"]["wblgs"]=wbl_data["wblgs"]
|
||||
return_data["data"]["wblzs"]=wbl_data["wblzs"]
|
||||
return_data["msg"] = "获取万宝楼数据完成"
|
||||
# 加载模板
|
||||
try:
|
||||
return_data["temp"] = load_template("wujia.html")
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"加载模板失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
return return_data
|
||||
return_data["code"] = 200
|
||||
return return_data
|
||||
|
||||
|
||||
async def __get_wbl_data(self,showName):
|
||||
"""获取万宝楼数据（公示和在售）"""
|
||||
try:
|
||||
#在配置文件中获取接口配置
|
||||
#api_config = self.__api_config["aijx3_wblwg"]
|
||||
api_config = self.__api_config["wbl_waiguan"]
|
||||
#更新参数
|
||||
api_config["params"]["filter[role_appearance]"] = showName
|
||||
# 获取公示数据
|
||||
api_config["params"]["filter[state]"] = "1"
|
||||
datawblgs = await self.__api.get(api_config["url"], api_config["params"], "data")
|
||||
# 获取在售数据
|
||||
api_config["params"]["filter[state]"] = "2"
|
||||
datawblzs = await self.__api.get(api_config["url"], api_config["params"], "data")
|
||||
|
||||
return {
|
||||
"wblgs": await self.__process_wbl_records(datawblgs.get("list", [])),
|
||||
"wblzs": await self.__process_wbl_records(datawblzs.get("list", []))
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"获取万宝楼数据出错: {e}")
|
||||
return None
|
||||
|
||||
|
||||
async def __process_wbl_records(self,records):
|
||||
"""处理万宝楼记录数据"""
|
||||
processed = []
|
||||
for record in records:
|
||||
try:
|
||||
# 转换时间戳为可读格式
|
||||
timestamp = record.get("remaining_time", 0)
|
||||
days = timestamp // 86400 # 每天有 86400 秒
|
||||
hours = (timestamp % 86400) // 3600 # 每小时有 3600 秒
|
||||
minutes = (timestamp % 3600) // 60 # 每分钟有 60 秒
|
||||
result = ""
|
||||
if days > 0:
|
||||
result = (f"{days}天")
|
||||
if hours > 0:
|
||||
result += (f"{hours}时")
|
||||
if minutes > 0:
|
||||
result += (f"{minutes}分钟")
|
||||
single_unit_price = record.get("single_unit_price", 0)
|
||||
processed.append({
|
||||
"priceNum": "{:.2f}".format( single_unit_price / 100),
|
||||
"belongQf2": record.get("server_name", "无数据"),
|
||||
"replyTime": result,
|
||||
"discountRate": record.get("discountRate", 0.0),
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error(f"处理万宝楼记录出错: {e}")
|
||||
continue
|
||||
|
||||
return processed
|
||||
|
||||
|
||||
async def jiaoyihang(self,Name: str = "守缺式",server: str = "梦江南"):
|
||||
"""
|
||||
区服交易行
|
||||
"""
|
||||
return_data = {
|
||||
"code": 0,
|
||||
"msg": "功能函数未执行",
|
||||
"data": {}
|
||||
}
|
||||
#在配置文件中获取接口配置
|
||||
api_config = self.__api_config["jx3box_item"]
|
||||
#更新参数
|
||||
api_config["params"]["keyword"] = Name
|
||||
# 获取多页查找信息
|
||||
data = await self.__api.all_pages("GET",api_config["url"],api_config["params"],"data","data")
|
||||
if not data:
|
||||
logger.error(f"获取多页数据失败")
|
||||
return_data["msg"] = "未找到该物品"
|
||||
return return_data
|
||||
# 提取指定字段
|
||||
fields = ["IconID", "Name","id"]
|
||||
result = extract_fields(data, fields)
|
||||
if not data:
|
||||
logger.error(f"提取字段失败")
|
||||
return_data["msg"] = "未找到该物品"
|
||||
return return_data
|
||||
# 提取id列表
|
||||
lists_id = extract_field(result, "id")
|
||||
strlists_id = ",".join(str(x) for x in lists_id)
|
||||
logger.info(f"{strlists_id}")
|
||||
#在配置文件中获取接口配置
|
||||
api_config = self.__api_config["jx3box_itemprice"]
|
||||
#更新参数
|
||||
api_config["params"]["itemIds"] = strlists_id
|
||||
api_config["params"]["server"] = server
|
||||
# 获取数据
|
||||
data = await self.__api.get(api_config["url"],api_config["params"],"data")
|
||||
if not data:
|
||||
return_data["msg"] = "未找到在售物品"
|
||||
return return_data
|
||||
#提取需要的字段
|
||||
fieldsjyh = ["ItemId", "SampleSize", "LowestPrice", "AvgPrice", "Date"]
|
||||
resultjyh = [{f: v[f] for f in fieldsjyh} for v in data.values()]
|
||||
#合并表格数据
|
||||
# 先建立一个字典映射，加快查找
|
||||
map_result = {item["id"]: {"IconID": item["IconID"], "Name": item["Name"]} for item in result}
|
||||
# 遍历 resultjyh，合并字段
|
||||
for item in resultjyh:
|
||||
if item["ItemId"] in map_result:
|
||||
item.update(map_result[item["ItemId"]])
|
||||
#处理数据
|
||||
for item in resultjyh:
|
||||
item["LowestPrice"] = gold_to_string(item["LowestPrice"])
|
||||
item["AvgPrice"] = gold_to_string(item["AvgPrice"])
|
||||
item["IconID"] = f"https://icon.jx3box.com/icon/{item['IconID']}.png"
|
||||
# 准备模板渲染数据
|
||||
return_data["data"] = {
|
||||
"items": resultjyh,
|
||||
"server": server,
|
||||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
}
|
||||
# 加载模板
|
||||
try:
|
||||
return_data["temp"] = load_template("jiaoyihang.html")
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"加载模板失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
return return_data
|
||||
return_data["code"] = 200
|
||||
return return_data
|
||||
@@ -21,7 +21,7 @@ class AsyncMySQL:
|
||||
async def fetch_one(self, sql: str, params=None):
|
||||
"""查询单条数据"""
|
||||
await self.init_pool()
|
||||
async with self.pool.acquire() as conn:
|
||||
async with self.pool.acquire() as conn: # type: ignore
|
||||
async with conn.cursor(aiomysql.DictCursor) as cursor:
|
||||
await cursor.execute(sql, params or ())
|
||||
return await cursor.fetchone()
|
||||
@@ -29,7 +29,7 @@ class AsyncMySQL:
|
||||
async def fetch_all(self, sql: str, params=None):
|
||||
"""查询多条数据"""
|
||||
await self.init_pool()
|
||||
async with self.pool.acquire() as conn:
|
||||
async with self.pool.acquire() as conn: # type: ignore
|
||||
async with conn.cursor(aiomysql.DictCursor) as cursor:
|
||||
await cursor.execute(sql, params or ())
|
||||
return await cursor.fetchall()
|
||||
@@ -37,7 +37,7 @@ class AsyncMySQL:
|
||||
async def execute(self, sql: str, params=None):
|
||||
"""执行 SQL（insert/update/delete）"""
|
||||
await self.init_pool()
|
||||
async with self.pool.acquire() as conn:
|
||||
async with self.pool.acquire() as conn: # type: ignore
|
||||
async with conn.cursor() as cursor:
|
||||
await cursor.execute(sql, params or ())
|
||||
await conn.commit()
|
||||
@@ -46,7 +46,7 @@ class AsyncMySQL:
|
||||
async def executemany(self, sql: str, params_list):
|
||||
"""批量执行 SQL"""
|
||||
await self.init_pool()
|
||||
async with self.pool.acquire() as conn:
|
||||
async with self.pool.acquire() as conn: # type: ignore
|
||||
async with conn.cursor() as cursor:
|
||||
await cursor.executemany(sql, params_list)
|
||||
await conn.commit()
|
||||
@@ -55,7 +55,7 @@ class AsyncMySQL:
|
||||
async def truncate_table(self, table_name: str):
|
||||
"""清空指定表"""
|
||||
await self.init_pool()
|
||||
async with self.pool.acquire() as conn:
|
||||
async with self.pool.acquire() as conn: # type: ignore
|
||||
async with conn.cursor() as cursor:
|
||||
sql = f"TRUNCATE TABLE `{table_name}`"
|
||||
await cursor.execute(sql)
|
||||
@@ -0,0 +1,593 @@
|
||||
from datetime import datetime
|
||||
from typing import Dict, Any, Optional, List, Union
|
||||
|
||||
from astrbot.api import logger
|
||||
|
||||
from .request import APIClient
|
||||
from .AsyncMySQL import AsyncMySQL
|
||||
from .function_basic import load_template,flatten_field,extract_fields,gold_to_string
|
||||
|
||||
class JX3Service:
|
||||
def __init__(self, api_config,db: AsyncMySQL ):
|
||||
self._api = APIClient()
|
||||
self._db = db
|
||||
self._api_config = api_config
|
||||
|
||||
|
||||
def _init_return_data(self) -> Dict[str, Any]:
|
||||
"""初始化标准的返回数据结构"""
|
||||
return {
|
||||
"code": 0,
|
||||
"msg": "功能函数未执行",
|
||||
"data": {}
|
||||
}
|
||||
|
||||
|
||||
async def _base_request(
|
||||
self,
|
||||
config_key: str,
|
||||
method: str,
|
||||
params: Optional[Dict[str, Any]] = None,
|
||||
out_key: Optional[str] = "data"
|
||||
) -> Optional[Any]:
|
||||
"""
|
||||
基础请求封装，处理配置获取和API调用。
|
||||
|
||||
:param config_key: 配置字典中对应 API 的键名。
|
||||
:param method: HTTP方法 ('GET' 或 'POST')。
|
||||
:param params: 请求参数或 Body 数据。
|
||||
:param out_key: 响应数据中需要提取的字段。
|
||||
:return: 成功时返回提取后的数据，失败时返回 None。
|
||||
"""
|
||||
try:
|
||||
api_config = self._api_config.get(config_key)
|
||||
if not api_config:
|
||||
logger.error(f"配置文件中未找到 key: {config_key}")
|
||||
return None
|
||||
|
||||
# 复制 params，避免修改原始配置模板
|
||||
request_params = api_config.get("params", {}).copy()
|
||||
if params:
|
||||
request_params.update(params)
|
||||
|
||||
url = api_config.get("url", "")
|
||||
if not url:
|
||||
logger.error(f"API配置缺少 URL: {config_key}")
|
||||
return None
|
||||
|
||||
if method.upper() == 'POST':
|
||||
data = await self._api.post(url, data=request_params, out_key=out_key)
|
||||
else: # 默认为 GET
|
||||
data = await self._api.get(url, params=request_params, out_key=out_key)
|
||||
|
||||
if not data:
|
||||
logger.warning(f"获取接口信息失败或返回空数据: {config_key}")
|
||||
|
||||
return data
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"基础请求调用出错 ({config_key}): {e}")
|
||||
return None
|
||||
|
||||
# --- 业务功能函数 ---
|
||||
|
||||
async def richang(self,server: str, num: int = 0) -> Dict[str, Any]:
|
||||
"""日常活动"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
# 1. 构造请求参数
|
||||
params = {"server": server, "num": num}
|
||||
|
||||
# 2. 调用基础请求
|
||||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||||
"jx3_richang", "GET", params=params
|
||||
)
|
||||
|
||||
if not data:
|
||||
return_data["msg"] = "获取接口信息失败"
|
||||
return return_data
|
||||
|
||||
# 3. 处理返回数据
|
||||
try:
|
||||
# 格式化字符串，利用字典的 get 方法提供默认值
|
||||
result_msg = (
|
||||
f"{server}\n{data.get('date', '未知日期')}-星期{data.get('week', '未知')}\n"
|
||||
f"大战：{data.get('war', '无')}\n"
|
||||
f"战场：{data.get('battle', '无')}\n"
|
||||
f"阵营：{data.get('orecar', '无')}\n"
|
||||
f"宗门：{data.get('school', '无')}\n"
|
||||
f"驰援：{data.get('rescue', '无')}\n"
|
||||
f"画像：{data.get('draw', '无')}\n"
|
||||
)
|
||||
|
||||
# 安全地处理列表索引
|
||||
luck = data.get('luck', [None, None, None])
|
||||
luck_msg = f"[宠物福缘]：\n{luck[0] or '无'},{luck[1] or '无'},{luck[2] or '无'}\n"
|
||||
card = data.get('card', [None, None, None])
|
||||
card_msg = f"[家园声望·加倍道具]：\n{card[0] or '无'},{card[1] or '无'},{card[2] or '无'}\n"
|
||||
team = data.get('team', [None, None, None])
|
||||
team_msg = f"[武林通鉴·公共任务]：\n{team[0] or '无'}\n[武林通鉴·团队秘境]：\n{team[2] or '无'}\n"
|
||||
|
||||
return_data["data"] = result_msg + luck_msg + card_msg + team_msg
|
||||
return_data["code"] = 200
|
||||
except Exception as e:
|
||||
logger.error(f"richang 数据处理时出错: {e}")
|
||||
return_data["msg"] = "处理接口返回信息时出错"
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def shapan(self, server: str ) -> Dict[str, Any]:
|
||||
"""区服沙盘"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
# 1. 构造请求参数
|
||||
params = {"serverName": server}
|
||||
|
||||
# 2. 调用基础请求
|
||||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||||
"aijx3_shapan", "POST", params=params
|
||||
)
|
||||
|
||||
if not data:
|
||||
return_data["msg"] = "获取接口信息失败"
|
||||
return return_data
|
||||
|
||||
# 3. 处理返回数据 (直接提取图片 URL)
|
||||
pic_url = data.get("picUrl")
|
||||
if pic_url:
|
||||
return_data["data"] = pic_url
|
||||
return_data["code"] = 200
|
||||
else:
|
||||
return_data["msg"] = "接口未返回图片URL"
|
||||
|
||||
return return_data
|
||||
|
||||
async def kaifu(self, server: str) -> Dict[str, Any]:
|
||||
"""开服状态查询"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
# 1. 构造请求参数
|
||||
params = {"server": server}
|
||||
|
||||
# 2. 调用基础请求
|
||||
data: Optional[Dict[str, Union[int, str]]] = await self._base_request(
|
||||
"jx3_kaifu", "GET", params=params
|
||||
)
|
||||
|
||||
if not data:
|
||||
return_data["msg"] = "获取接口信息失败"
|
||||
return return_data
|
||||
|
||||
# 3. 处理返回数据
|
||||
try:
|
||||
status = data.get("status", 0)
|
||||
timestamp = data.get("time", 0)
|
||||
|
||||
status_time = datetime.fromtimestamp(float(timestamp)).strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
if status == 1:
|
||||
status_str = f"{server}服务器已开服，快冲，快冲！\n开服时间：{status_time}"
|
||||
status_bool = True
|
||||
else:
|
||||
status_str = f"{server}服务器当前维护中，等会再来吧！\n维护时间：{status_time}"
|
||||
status_bool = False
|
||||
|
||||
return_data["status"] = status_bool
|
||||
return_data["data"] = status_str
|
||||
return_data["code"] = 200
|
||||
except Exception as e:
|
||||
logger.error(f"kaifu 数据处理时出错: {e}")
|
||||
return_data["msg"] = "处理接口返回信息时出错"
|
||||
|
||||
return return_data
|
||||
|
||||
async def shaohua(self) -> Dict[str, Any]:
|
||||
"""骚话"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
# 因为没有参数，所以 params=None
|
||||
data: Optional[Dict[str, Any]] = await self._base_request("jx3_shaohua", "GET")
|
||||
|
||||
if not data:
|
||||
return_data["msg"] = "获取接口信息失败"
|
||||
return return_data
|
||||
|
||||
text = data.get("text")
|
||||
if text:
|
||||
return_data["data"] = text
|
||||
return_data["code"] = 200
|
||||
else:
|
||||
return_data["msg"] = "接口未返回文本"
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def jigai(self) -> Dict[str, Any]:
|
||||
"""技改记录"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
# 提取字段可能返回列表
|
||||
data: Optional[List[Dict[str, Any]]] = await self._base_request("jx3_jigai", "GET")
|
||||
|
||||
if not data or not isinstance(data, list):
|
||||
return_data["msg"] = "获取接口信息失败或数据格式错误"
|
||||
return return_data
|
||||
|
||||
try:
|
||||
result_msg = "剑网三最近技改\n"
|
||||
# 仅展示前1条，避免消息过长
|
||||
for i, item in enumerate(data[:1], 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"jigai 数据处理时出错: {e}")
|
||||
return_data["msg"] = "处理接口返回信息时出错"
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def jinjia(self, server: str) -> Dict[str, Any]:
|
||||
"""区服金价"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
params = {"serverName": server}
|
||||
data_list: Optional[List[Dict[str, Any]]] = await self._base_request("aijx3_jinjia", "POST", params=params)
|
||||
|
||||
if not data_list or not isinstance(data_list, list):
|
||||
return_data["msg"] = "获取接口信息失败或数据格式错误"
|
||||
return return_data
|
||||
|
||||
# 安全处理列表切片
|
||||
display_data = data_list[:15]
|
||||
|
||||
# 加载模板
|
||||
try:
|
||||
return_data["temp"] = load_template("jinjia.html")
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"加载模板失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
return return_data
|
||||
|
||||
# 准备模板渲染数据
|
||||
try:
|
||||
return_data["data"] = {
|
||||
"items": display_data,
|
||||
"server": server,
|
||||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
}
|
||||
return_data["code"] = 200
|
||||
except Exception as e:
|
||||
logger.error(f"jinjia 模板数据准备失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板渲染数据准备失败"
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def qiyu(self, adventureName: str = "阴阳两界", serverName: str = "眉间雪") -> Dict[str, Any]:
|
||||
"""区服奇遇"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
params = {"adventureName": adventureName, "serverName": serverName}
|
||||
data_list: Optional[List[Dict[str, Any]]] = await self._base_request("aijx3_qiyu", "POST", params=params)
|
||||
|
||||
if not data_list or not isinstance(data_list, list):
|
||||
return_data["msg"] = "获取接口信息失败或数据格式错误"
|
||||
return return_data
|
||||
|
||||
# 格式化时间
|
||||
for item in data_list:
|
||||
timestamp = item.get("time")
|
||||
if timestamp and isinstance(timestamp, (int, float)):
|
||||
# 修复时间戳：原代码显示这里是毫秒级，除以 1000
|
||||
item["time"] = datetime.fromtimestamp(timestamp / 1000).strftime("%Y-%m-%d %H:%M:%S")
|
||||
else:
|
||||
item["time"] = "未知时间" # 确保即使 time 字段缺失也不会报错
|
||||
|
||||
# 加载模板
|
||||
try:
|
||||
return_data["temp"] = load_template("qiyuliebiao.html")
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"加载模板失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
return return_data
|
||||
|
||||
# 准备模板渲染数据
|
||||
try:
|
||||
return_data["data"] = {
|
||||
"items": data_list,
|
||||
"server": serverName,
|
||||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"qiyuname": adventureName
|
||||
}
|
||||
return_data["code"] = 200
|
||||
except Exception as e:
|
||||
logger.error(f"qiyu 模板数据准备失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板渲染数据准备失败"
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def SearchData(self) -> Dict[str, Any]:
|
||||
"""外观数据插入数据库"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
# 获取数据
|
||||
data: Optional[List[Dict[str, Any]]] = await self._base_request("aijx3_SearchData", "POST")
|
||||
|
||||
if not data:
|
||||
return_data["msg"] = "获取接口信息失败"
|
||||
return return_data
|
||||
|
||||
# 提取数据
|
||||
try:
|
||||
extracted_data = flatten_field(data, "dataModels")
|
||||
except Exception as e:
|
||||
logger.error(f"提取数据失败: {e}")
|
||||
return_data["msg"] = "提取指定数据字段失败"
|
||||
return return_data
|
||||
|
||||
# 准备SQL和批量数据
|
||||
sql = """
|
||||
INSERT INTO searchdata
|
||||
(typeName, name, showName, picUrl, searchId, searchDescType)
|
||||
VALUES (%s, %s, %s, %s, %s, %s)
|
||||
"""
|
||||
# 增强数据提取的安全性
|
||||
values_list = [
|
||||
(
|
||||
item.get('typeName'),
|
||||
item.get('name'),
|
||||
item.get('showName'),
|
||||
item.get('picUrl'),
|
||||
item.get('searchId'),
|
||||
item.get('searchDescType')
|
||||
)
|
||||
for item in extracted_data
|
||||
]
|
||||
|
||||
# 插入数据
|
||||
try:
|
||||
await self._db.truncate_table("searchdata")
|
||||
await self._db.executemany(sql, values_list)
|
||||
except Exception as e:
|
||||
logger.error(f"数据插入失败: {e}")
|
||||
return_data["msg"] = "数据插入数据库失败"
|
||||
return return_data
|
||||
|
||||
return_data["msg"] = f"成功批量插入 {len(values_list)} 条数据！"
|
||||
return_data["code"] = 200
|
||||
return return_data
|
||||
|
||||
|
||||
async def wujia(self, Name: str) -> Dict[str, Any]:
|
||||
"""物价查询"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
# 1. 查询魔盒获取外观名称 (jx3box_exterior)
|
||||
params_search = {"keyword": Name}
|
||||
search_data: Optional[Dict[str, Any]] = await self._base_request("jx3box_exterior", "GET", params=params_search)
|
||||
|
||||
showName = ""
|
||||
searchId = None
|
||||
|
||||
# 2. 确定外观名称和 ID
|
||||
if search_data and search_data.get("total", 0) > 0 and search_data.get("list"):
|
||||
# 从 API 获取名称
|
||||
showName = search_data["list"][0].get("name", "")
|
||||
|
||||
# 进一步从 DB 查找 ID
|
||||
try:
|
||||
sql = "SELECT searchId FROM searchdata WHERE showName=%s"
|
||||
sqldata = await self._db.fetch_one(sql, (showName,))
|
||||
searchId = sqldata["searchId"] if sqldata else None
|
||||
except Exception as e:
|
||||
logger.warning(f"从 DB 获取外观ID错误: {e}") # 警告而非错误，因为可能数据库同步失败
|
||||
|
||||
elif search_data and search_data.get("total", 0) == 0:
|
||||
# 从 DB 查找名称和 ID
|
||||
try:
|
||||
sql = "SELECT showName, searchId FROM searchdata WHERE name = %s OR showName = %s"
|
||||
sqldata = await self._db.fetch_one(sql, (Name, Name))
|
||||
if sqldata:
|
||||
showName = sqldata["showName"]
|
||||
searchId = sqldata["searchId"]
|
||||
else:
|
||||
raise ValueError("DB did not find data.")
|
||||
except Exception as e:
|
||||
logger.error(f"获取外观信息错误: {e}")
|
||||
return_data["msg"] = "未找到该外观信息"
|
||||
return return_data
|
||||
|
||||
if not showName or not searchId:
|
||||
return_data["msg"] = "无法确定外观名称或ID"
|
||||
return return_data
|
||||
|
||||
return_data["data"]["showName"] = showName
|
||||
return_data["data"]["searchId"] = searchId
|
||||
|
||||
# 3. 获取外观详细数据 (aijx3_GoodsDetail)
|
||||
params_detail = {"goodsName": showName}
|
||||
detail_data: Optional[Dict[str, Any]] = await self._base_request("aijx3_GoodsDetail", "POST", params=params_detail)
|
||||
|
||||
if not detail_data:
|
||||
return_data["msg"] = "获取外观详细数据失败"
|
||||
return return_data
|
||||
|
||||
# 提取外观详细数据
|
||||
try:
|
||||
imgs = detail_data.get("imgs", [])
|
||||
return_data["data"].update({
|
||||
"goodsDesc": detail_data.get("goodsDesc", "无描述"),
|
||||
"publishTime": detail_data.get("publishTime", "无价格"),
|
||||
"priceNum": detail_data.get("priceNum", 0),
|
||||
"goodsId": detail_data.get("goodsId", "无数据"),
|
||||
"imgs": imgs[0] if imgs and isinstance(imgs, list) else "",
|
||||
"goodsAlias": detail_data.get("goodsAlias", "无别名"),
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error(f"提取外观详细数据失败: {e}")
|
||||
return_data["msg"] = "提取外观详细数据失败"
|
||||
return return_data
|
||||
|
||||
# 4. 查询万宝楼数据（公示和在售）
|
||||
wbl_data = await self._get_wbl_data(showName)
|
||||
if wbl_data:
|
||||
return_data["data"]["wblgs"] = wbl_data["wblgs"]
|
||||
return_data["data"]["wblzs"] = wbl_data["wblzs"]
|
||||
return_data["msg"] = "获取万宝楼数据完成"
|
||||
else:
|
||||
logger.warning("万宝楼数据获取失败")
|
||||
|
||||
# 5. 加载模板
|
||||
try:
|
||||
return_data["temp"] = load_template("wujia.html")
|
||||
return_data["code"] = 200
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"加载模板失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
return return_data
|
||||
|
||||
return return_data
|
||||
|
||||
|
||||
async def _get_wbl_data(self,showName):
|
||||
"""获取万宝楼数据（公示和在售）"""
|
||||
try:
|
||||
#在配置文件中获取接口配置
|
||||
#api_config = self._api_config["aijx3_wblwg"]
|
||||
api_config = self._api_config["wbl_waiguan"]
|
||||
#更新参数
|
||||
api_config["params"]["filter[role_appearance]"] = showName
|
||||
# 获取公示数据
|
||||
api_config["params"]["filter[state]"] = "1"
|
||||
datawblgs = await self._api.get(api_config["url"], api_config["params"], "data")
|
||||
# 获取在售数据
|
||||
api_config["params"]["filter[state]"] = "2"
|
||||
datawblzs = await self._api.get(api_config["url"], api_config["params"], "data")
|
||||
|
||||
return {
|
||||
"wblgs": await self._process_wbl_records(datawblgs.get("list", [])),
|
||||
"wblzs": await self._process_wbl_records(datawblzs.get("list", []))
|
||||
}
|
||||
except Exception as e:
|
||||
logger.error(f"获取万宝楼数据出错: {e}")
|
||||
return None
|
||||
|
||||
|
||||
async def _process_wbl_records(self,records):
|
||||
"""处理万宝楼记录数据"""
|
||||
processed = []
|
||||
for record in records:
|
||||
try:
|
||||
# 转换时间戳为可读格式
|
||||
timestamp = record.get("remaining_time", 0)
|
||||
days = timestamp // 86400 # 每天有 86400 秒
|
||||
hours = (timestamp % 86400) // 3600 # 每小时有 3600 秒
|
||||
minutes = (timestamp % 3600) // 60 # 每分钟有 60 秒
|
||||
result = ""
|
||||
if days > 0:
|
||||
result = (f"{days}天")
|
||||
if hours > 0:
|
||||
result += (f"{hours}时")
|
||||
if minutes > 0:
|
||||
result += (f"{minutes}分钟")
|
||||
single_unit_price = record.get("single_unit_price", 0)
|
||||
processed.append({
|
||||
"priceNum": "{:.2f}".format( single_unit_price / 100),
|
||||
"belongQf2": record.get("server_name", "无数据"),
|
||||
"replyTime": result,
|
||||
"discountRate": record.get("discountRate", 0.0),
|
||||
})
|
||||
except Exception as e:
|
||||
logger.error(f"处理万宝楼记录出错: {e}")
|
||||
continue
|
||||
|
||||
return processed
|
||||
|
||||
|
||||
async def jiaoyihang(self, Name: str , server: str) -> Dict[str, Any]:
|
||||
"""区服交易行"""
|
||||
return_data = self._init_return_data()
|
||||
|
||||
# 1. 查找物品 ID 列表 (多页)
|
||||
api_config_search = self._api_config["jx3box_item"]
|
||||
search_params = api_config_search.get("params", {}).copy()
|
||||
search_params["keyword"] = Name
|
||||
|
||||
# 使用 all_pages 获取所有数据
|
||||
all_items: List[Dict[str, Any]] = await self._api.all_pages(
|
||||
"GET",
|
||||
api_config_search["url"],
|
||||
params_data=search_params, # 对应到 APIClient.all_pages
|
||||
out_key="data",
|
||||
list_key="data"
|
||||
)
|
||||
|
||||
if not all_items:
|
||||
return_data["msg"] = "未找到该物品"
|
||||
return return_data
|
||||
|
||||
# 2. 提取 ID、IconID 和 Name
|
||||
fields = ["IconID", "Name", "id"]
|
||||
# 安全提取字段，并确保 id 存在且是数字
|
||||
result = [extract_fields([item], fields)[0] for item in all_items if item.get("id") is not None]
|
||||
|
||||
lists_id = [str(item.get("id")) for item in result if item.get("id") is not None]
|
||||
strlists_id = ",".join(lists_id)
|
||||
|
||||
if not strlists_id:
|
||||
return_data["msg"] = "未找到有效的物品ID"
|
||||
return return_data
|
||||
|
||||
# 3. 查询物品价格 (jx3box_itemprice)
|
||||
api_config_price = self._api_config["jx3box_itemprice"]
|
||||
price_params = api_config_price.get("params", {}).copy()
|
||||
price_params["itemIds"] = strlists_id
|
||||
price_params["server"] = server
|
||||
|
||||
price_data: Optional[Dict[str, Dict[str, Any]]] = await self._api.get(
|
||||
api_config_price["url"], params=price_params, out_key="data"
|
||||
)
|
||||
|
||||
if not price_data:
|
||||
return_data["msg"] = "未找到在售物品"
|
||||
return return_data
|
||||
|
||||
# 4. 合并数据和格式化
|
||||
fieldsjyh = ["ItemId", "SampleSize", "LowestPrice", "AvgPrice", "Date"]
|
||||
resultjyh = [{f: v[f] for f in fieldsjyh} for v in price_data.values()]
|
||||
#合并表格数据
|
||||
# 先建立一个字典映射，加快查找
|
||||
map_result = {item["id"]: {"IconID": item["IconID"], "Name": item["Name"]} for item in result}
|
||||
# 遍历 resultjyh，合并字段
|
||||
for item in resultjyh:
|
||||
if item["ItemId"] in map_result:
|
||||
item.update(map_result[item["ItemId"]])
|
||||
#处理数据
|
||||
for item in resultjyh:
|
||||
item["LowestPrice"] = gold_to_string(item["LowestPrice"])
|
||||
item["AvgPrice"] = gold_to_string(item["AvgPrice"])
|
||||
item["IconID"] = f"https://icon.jx3box.com/icon/{item['IconID']}.png"
|
||||
|
||||
# 5. 模板渲染
|
||||
try:
|
||||
return_data["temp"] = load_template("jiaoyihang.html")
|
||||
return_data["data"] = {
|
||||
"items": resultjyh,
|
||||
"server": server,
|
||||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
}
|
||||
return_data["code"] = 200
|
||||
except FileNotFoundError as e:
|
||||
logger.error(f"加载模板失败: {e}")
|
||||
return_data["msg"] = "系统错误：模板文件不存在"
|
||||
except Exception as e:
|
||||
logger.error(f"模板渲染数据准备失败: {e}")
|
||||
return_data["msg"] = "系统错误：数据处理失败"
|
||||
|
||||
return return_data
|
||||
+192
@@ -0,0 +1,192 @@
|
||||
# core/request.py
|
||||
import json
|
||||
import aiohttp
|
||||
from typing import Optional, Dict, Any, Union, List
|
||||
|
||||
from aiohttp import ClientTimeout, ClientSession
|
||||
from astrbot.api import logger
|
||||
|
||||
class APIClient:
|
||||
"""
|
||||
API客户端类
|
||||
|
||||
优化说明：
|
||||
1. 复用 aiohttp.ClientSession 以提高性能。
|
||||
2. 增加类型提示 (Type Hints)。
|
||||
3. 支持异步上下文管理器 (Async Context Manager)。
|
||||
"""
|
||||
|
||||
def __init__(self, base_timeout: int = 10, ssl_verify: bool = False):
|
||||
self.base_timeout = base_timeout
|
||||
self.ssl_verify = ssl_verify
|
||||
self._session: Optional[ClientSession] = None
|
||||
|
||||
async def get_session(self) -> ClientSession:
|
||||
"""获取或创建单例 Session"""
|
||||
if self._session is None or self._session.closed:
|
||||
timeout = ClientTimeout(total=self.base_timeout)
|
||||
self._session = ClientSession(timeout=timeout)
|
||||
return self._session
|
||||
|
||||
async def close(self):
|
||||
"""关闭 Session"""
|
||||
if self._session and not self._session.closed:
|
||||
await self._session.close()
|
||||
|
||||
async def __aenter__(self):
|
||||
await self.get_session()
|
||||
return self
|
||||
|
||||
async def __aexit__(self, exc_type, exc_val, exc_tb):
|
||||
await self.close()
|
||||
|
||||
async def _request(self, method: str, url: str, params: Optional[Dict] = None, json_data: Optional[Dict] = None) -> Any:
|
||||
"""
|
||||
统一的内部请求处理方法
|
||||
"""
|
||||
session = await self.get_session()
|
||||
method = method.upper()
|
||||
|
||||
# 记录日志
|
||||
logger.debug(f"发起 {method} 请求: {url}")
|
||||
if params: logger.debug(f"Query参数: {params}")
|
||||
if json_data: logger.debug(f"Body数据: {json_data}")
|
||||
|
||||
try:
|
||||
# aiohttp 会自动处理 json=json_data 时的 Content-Type
|
||||
async with session.request(
|
||||
method=method,
|
||||
url=url,
|
||||
params=params,
|
||||
json=json_data,
|
||||
ssl=self.ssl_verify
|
||||
) as response:
|
||||
return await self._handle_response(response)
|
||||
|
||||
except aiohttp.ClientError as e:
|
||||
logger.error(f"网络请求出错 ({method} {url}): {e}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.error(f"未知错误 ({method} {url}): {e}")
|
||||
return None
|
||||
|
||||
async def _handle_response(self, response: aiohttp.ClientResponse) -> Any:
|
||||
"""处理响应：自动识别二进制或JSON"""
|
||||
try:
|
||||
logger.debug(f"响应状态: {response.status}")
|
||||
response.raise_for_status() # 如果是 4xx/5xx 直接抛出异常
|
||||
|
||||
content_type = response.headers.get('Content-Type', '').lower()
|
||||
|
||||
# 处理二进制流 (图片、文件等)
|
||||
if 'image' in content_type or 'octet-stream' in content_type:
|
||||
return await response.read()
|
||||
|
||||
# 处理 JSON
|
||||
# 优先尝试标准 json 解析
|
||||
try:
|
||||
data = await response.json()
|
||||
except Exception:
|
||||
# 容错：有些 API 返回 text/html 但内容是 JSON
|
||||
text = await response.text()
|
||||
try:
|
||||
data = json.loads(text)
|
||||
except json.JSONDecodeError:
|
||||
logger.error(f"无法解析响应为 JSON。原始内容: {text[:100]}...")
|
||||
return None
|
||||
|
||||
logger.debug(f"响应数据: {data}")
|
||||
return self._validate_api_payload(data)
|
||||
|
||||
except aiohttp.ClientError as e:
|
||||
logger.error(f"HTTP响应错误: {e}")
|
||||
return None
|
||||
|
||||
def _validate_api_payload(self, data: Any) -> Any:
|
||||
"""校验业务层面的 JSON 数据结构"""
|
||||
if not data:
|
||||
logger.error("API返回空数据")
|
||||
return None
|
||||
|
||||
# 如果返回的是 JSON 字符串而非对象，再次解析
|
||||
if isinstance(data, str):
|
||||
try:
|
||||
data = json.loads(data)
|
||||
except json.JSONDecodeError:
|
||||
return None
|
||||
|
||||
if isinstance(data, dict) and 'code' in data:
|
||||
# 兼容多种成功状态码：200, "0", 0, 1
|
||||
code = data.get('code')
|
||||
if code not in [200, "0", 0, 1]:
|
||||
msg = data.get('msg') or data.get('message', '未知错误')
|
||||
logger.error(f"API业务报错: code={code}, msg={msg}")
|
||||
return None
|
||||
|
||||
return data
|
||||
|
||||
async def get(self, url: str, params: Optional[Dict] = None, out_key: Optional[str] = None) -> Any:
|
||||
"""GET 请求封装"""
|
||||
data = await self._request('GET', url, params=params)
|
||||
return self._extract_data(data, out_key)
|
||||
|
||||
async def post(self, url: str, data: Optional[Dict] = None, out_key: Optional[str] = None) -> Any:
|
||||
"""POST 请求封装 (默认发送 JSON)"""
|
||||
data = await self._request('POST', url, json_data=data)
|
||||
return self._extract_data(data, out_key)
|
||||
|
||||
def _extract_data(self, data: Any, key: Optional[str]) -> Any:
|
||||
"""辅助方法：从结果中提取指定字段"""
|
||||
if data is None:
|
||||
return None
|
||||
if isinstance(data, bytes):
|
||||
return data
|
||||
if key and isinstance(data, dict):
|
||||
return data.get(key, {})
|
||||
return data
|
||||
|
||||
async def all_pages(
|
||||
self,
|
||||
method: str,
|
||||
url: str,
|
||||
params_data: Optional[Dict] = None,
|
||||
out_key: str = "",
|
||||
list_key: str = "list",
|
||||
max_pages: int = 10
|
||||
) -> List[Any]:
|
||||
"""
|
||||
分页获取所有数据
|
||||
:param method: GET 或 POST
|
||||
:param list_key: 列表数据在 JSON 中的字段名，如 'data' 或 'list'
|
||||
"""
|
||||
all_data = []
|
||||
current_page = 1
|
||||
params = params_data.copy() if params_data else {}
|
||||
|
||||
while True:
|
||||
params["page"] = str(current_page)
|
||||
|
||||
if method.upper() == "POST":
|
||||
data = await self.post(url, data=params, out_key=out_key)
|
||||
else:
|
||||
data = await self.get(url, params=params, out_key=out_key)
|
||||
|
||||
# 终止条件判断
|
||||
if not data or isinstance(data, bytes):
|
||||
break
|
||||
|
||||
# 如果 data 是列表本身（有些API直接返回列表）
|
||||
page_items = data if isinstance(data, list) else data.get(list_key)
|
||||
|
||||
if not page_items:
|
||||
break
|
||||
|
||||
all_data.extend(page_items)
|
||||
|
||||
if current_page >= max_pages:
|
||||
break
|
||||
|
||||
current_page += 1
|
||||
logger.info(f"已获取第 {current_page} 页数据")
|
||||
|
||||
return all_data
|
||||
@@ -8,8 +8,8 @@ from astrbot.api.star import Context, Star, register, StarTools
|
||||
from astrbot.api import logger
|
||||
from astrbot.api import AstrBotConfig
|
||||
|
||||
from .core.AsyncMySQL import AsyncMySQL
|
||||
from .core.JX3Function import JX3Function
|
||||
from .core.async_mysql import AsyncMySQL
|
||||
from .core.jx3_service import JX3Service
|
||||
from .core.WZRYFunction import WZRYFunction
|
||||
|
||||
|
||||
@@ -52,7 +52,7 @@ class Jx3ApiPlugin(Star):
|
||||
}
|
||||
#创建类实例
|
||||
self.db = AsyncMySQL(db_config)
|
||||
self.jx3fun = JX3Function(self.api_config,self.db)
|
||||
self.jx3fun = JX3Service(self.api_config,self.db)
|
||||
self.wzry = WZRYFunction(self.api_config,self.db)
|
||||
# 周期函数调用
|
||||
|
||||
|
||||
Reference in New Issue
Block a user