864 lines
31 KiB
Python
864 lines
31 KiB
Python
|
||
from datetime import datetime
|
||
from typing import Dict, Any, Optional, List, Union
|
||
|
||
from astrbot.api import logger
|
||
from astrbot.api import AstrBotConfig
|
||
|
||
from .request import APIClient
|
||
from .aiosqlite import AsyncSQLite
|
||
from .function_basic import load_template,flatten_field,extract_fields,gold_to_string
|
||
|
||
class JX3Service:
|
||
def __init__(self, api_config, config:AstrBotConfig):
|
||
self._api = APIClient()
|
||
# 获取API配置文件
|
||
self._api_config = api_config
|
||
# 获取插件配置文件
|
||
self._config = config
|
||
# 获取配置中的 Token
|
||
self.token = self._config.get("jx3api_token", "")
|
||
if self.token == "":
|
||
logger.info("获取配置token失败，请正确填写token,否则部分功能无法正常使用")
|
||
else:
|
||
logger.info(f"获取配置token成功。{self.token}")
|
||
# 获取配置中的 ticket
|
||
self.ticket = self._config.get("jx3api_ticket", "")
|
||
if self.ticket == "":
|
||
logger.info("获取配置ticket失败，请正确填写ticket,否则部分功能无法正常使用")
|
||
else:
|
||
logger.info(f"获取配置ticket成功。{self.ticket}")
|
||
|
||
|
||
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 helps(self) -> Dict[str, Any]:
|
||
"""帮助"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = load_template("helps.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
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', [])
|
||
luck_msg = f"[宠物福缘]：\n{', '.join(luck)}\n"
|
||
card = data.get('card', [])
|
||
card_msg = f"[家园声望·加倍道具]：\n{', '.join(card)}\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 richangyuche(self) -> Dict[str, Any]:
|
||
"""日常预测"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = { "num": 30}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_richangyuche", "GET", params=params
|
||
)
|
||
logger.info(f"richang 接口返回数据: {data}")
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
# 格式化字符串，利用字典的 get 方法提供默认值
|
||
|
||
return_data["code"] = 0
|
||
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 zhuangtai(self) -> Dict[str, Any]:
|
||
"""区服状态"""
|
||
return_data = self._init_return_data()
|
||
|
||
|
||
data: Optional[Dict[str, Any]] = await self._base_request("jx3_zhuangtai", "GET")
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
server_wj = []
|
||
server_dx = []
|
||
server_sx = []
|
||
|
||
for itme in data:
|
||
if itme['zone'] == "无界区":
|
||
server_wj.append(itme)
|
||
elif itme['zone'] == "电信区":
|
||
server_dx.append(itme)
|
||
elif itme['zone'] == "双线区":
|
||
server_sx.append(itme)
|
||
|
||
return_data["data"]["server_wj"] = server_wj
|
||
return_data["data"]["server_dx"] = server_dx
|
||
return_data["data"]["server_sx"] = server_sx
|
||
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = load_template("qufuzhuangtai.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
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 xinwei(self) -> Dict[str, Any]:
|
||
"""新闻资讯"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 提取字段可能返回列表
|
||
data: Optional[List[Dict[str, Any]]] = await self._base_request("jx3_xinweng", "GET")
|
||
|
||
if not data or not isinstance(data, list):
|
||
return_data["msg"] = "获取接口信息失败或数据格式错误"
|
||
return return_data
|
||
|
||
try:
|
||
#
|
||
result = data[0]
|
||
return_data["status"] = result.get('id')
|
||
|
||
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('date', '未知时间')}\n"
|
||
result_msg += f"链接：{item.get('url', '无链接')}\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, limit:str) -> Dict[str, Any]:
|
||
"""区服金价"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 获取配置中的 Token
|
||
token = self._config.get("jx3api_token", "")
|
||
if token == "":
|
||
return_data["msg"] = "系统未配置API访问Token"
|
||
return return_data
|
||
|
||
params = {"server": server, "limit": limit, "token": token}
|
||
data_list: Optional[List[Dict[str, Any]]] = await self._base_request("jx3_jinjia", "GET", params=params)
|
||
|
||
if not data_list or not isinstance(data_list, list):
|
||
return_data["msg"] = "获取接口信息失败或数据格式错误"
|
||
return return_data
|
||
# 加载模板
|
||
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"] = data_list
|
||
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 wujia(self, Name: str, server:str) -> Dict[str, Any]:
|
||
"""物价查询"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 2. 确定外观名称和 ID
|
||
|
||
params_search = {"name": Name,"token": self.token, "server": server}
|
||
search_data: Optional[Dict[str, Any]] = await self._base_request("jx3_wujia", "GET", params=params_search)
|
||
|
||
if not search_data:
|
||
return_data["msg"] = "未找到改外观"
|
||
return return_data
|
||
|
||
return_data["data"] = search_data
|
||
|
||
# 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 jiaoyihang(self, name: str , server: str) -> Dict[str, Any]:
|
||
"""区服交易行"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "name": name,"token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_jiaoyihang", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "未找到该物品"
|
||
return return_data
|
||
|
||
# 2. 数据处理
|
||
result = []
|
||
|
||
try:
|
||
for item in data:
|
||
inner_list = item.get("data", [])
|
||
first = inner_list[0] if inner_list else {}
|
||
new_item = {
|
||
"name": item.get("name"),
|
||
"icon": f"https://icon.jx3box.com/icon/{item.get('icon')}.png",
|
||
"sever": first.get("server"),
|
||
"count": len(inner_list),
|
||
"unit_price": gold_to_string(first.get("unit_price")),
|
||
"created": datetime.fromtimestamp(first.get("created")).strftime("%Y-%m-%d %H:%M:%S"),
|
||
}
|
||
result.append(new_item)
|
||
except Exception as e:
|
||
logger.error(f"处理交易行数据失败: {e}")
|
||
return_data["msg"] = "处理交易行数据失败"
|
||
|
||
return_data["data"]["list"] = result
|
||
|
||
# 5. 模板渲染
|
||
try:
|
||
return_data["temp"] = load_template("jiaoyihang.html")
|
||
return_data["code"] = 200
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
|
||
return return_data
|
||
|
||
|
||
async def jueshemingpian(self, server: str, name:str ) -> Dict[str, Any]:
|
||
"""角色名片"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "name": name,"token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_jieshemingpian", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "未找到该角色"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据 (直接提取图片 URL)
|
||
return_data["data"] = data
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def shuijimingpian(self, force: str, body:str, server:str) -> Dict[str, Any]:
|
||
"""随机名片"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "body": body, "force":force, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_shuijimingpian", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据 (直接提取图片 URL)
|
||
return_data["data"] = data
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def yanhuachaxun(self, server: str, name:str ) -> Dict[str, Any]:
|
||
"""烟花查询"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "name": name,"token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_yanhuachaxun", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据 (直接提取图片 URL)
|
||
# 格式化时间
|
||
for item in data:
|
||
timestamp = item.get("time")
|
||
if timestamp and isinstance(timestamp, (int, float)):
|
||
# 修复时间戳：原代码显示这里是毫秒级，除以 1000
|
||
item["time"] = datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S")
|
||
else:
|
||
item["time"] = "未知时间" # 确保即使 time 字段缺失也不会报错
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = load_template("yanhuan.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["data"]["list"] = data
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def dilujilu(self, server: str) -> Dict[str, Any]:
|
||
"""的卢记录"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_dilujilu", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
for item in data:
|
||
item["refresh_time"] = datetime.fromtimestamp(item["refresh_time"]).strftime("%Y-%m-%d %H:%M:%S")
|
||
item["capture_time"] = datetime.fromtimestamp(item["capture_time"]).strftime("%Y-%m-%d %H:%M:%S")
|
||
item["auction_time"] = datetime.fromtimestamp(item["auction_time"]).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = load_template("dilujilu.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["data"]["list"] = data
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def tuanduizhaomu(self, server: str, keyword: str) -> Dict[str, Any]:
|
||
"""团队招募"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "keyword": keyword, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_tuanduizhaomu", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
for item in data["data"]:
|
||
item["createTime"] = datetime.fromtimestamp(item["createTime"]).strftime("%Y-%m-%d %H:%M:%S")
|
||
item["number"] = f"{item['number']}/{item['maxNumber']}"
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = load_template("tuanduizhaomu.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["data"]["list"] = data["data"]
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def zhanji(self, name: str, server:str, mode:str) -> Dict[str, Any]:
|
||
"""战绩+名片"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "name":name, "mode":mode, "token": self.token, "ticket": self.ticket}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_zhanji", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "查询角色战绩失败"
|
||
return return_data
|
||
logger.info("战绩获取完成")
|
||
|
||
# 角色名片获取
|
||
datamp = await self.jueshemingpian(server,name)
|
||
if datamp["code"] == 200:
|
||
data["showAvatar"] = datamp['data']['showAvatar']
|
||
logger.info("名片获取完成")
|
||
else:
|
||
data["showAvatar"] = ""
|
||
logger.info("名片获取失败")
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = load_template("zhanji.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["data"] = data
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def juesheqiyu(self, name: str, server: str) -> Dict[str, Any]:
|
||
"""角色奇遇"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "name": name, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_qiyu", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "未找到改角色奇遇信息"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
return_data["data"]["ptqy"] = []
|
||
return_data["data"]["jsqy"] = []
|
||
return_data["data"]["cwqy"] = []
|
||
|
||
for item in data:
|
||
item["time"] = datetime.fromtimestamp(item["time"]).strftime("%Y-%m-%d %H:%M:%S")
|
||
if item["level"] == 1:
|
||
return_data["data"]["ptqy"].append(item)
|
||
if item["level"] == 2:
|
||
return_data["data"]["jsqy"].append(item)
|
||
if item["level"] == 3:
|
||
return_data["data"]["cwqy"].append(item)
|
||
except Exception as e:
|
||
logger.error(f"处理返回数据失败: {e}")
|
||
return_data["msg"] = "处理返回数据失败"
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = load_template("juesheqiyu.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def zhengyingpaimai(self, server: str, name: str) -> Dict[str, Any]:
|
||
"""阵营拍卖"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "name": name, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_zhengyingpaimai", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
for item in data:
|
||
item["time"] = datetime.fromtimestamp(item["time"]).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = load_template("zhengyingpaimai.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["data"]["list"] = data
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def fuyaojjiutian(self, server: str) -> Dict[str, Any]:
|
||
"""扶摇九天"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_fuyaojiutian", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
data_new = data[0]
|
||
data_old = data[1]
|
||
result_msg = f"{server}\n"
|
||
result_msg += f"上次[扶摇九天]开启时间：\n{datetime.fromtimestamp(data_old['time']).strftime('%Y-%m-%d %H:%M:%S')}\n"
|
||
result_msg += f"下次[扶摇九天]开启时间：\n{datetime.fromtimestamp(data_new['time']).strftime('%Y-%m-%d %H:%M:%S')}"
|
||
except Exception as e:
|
||
logger.error(f"处理返回数据失败: {e}")
|
||
return_data["msg"] = "处理返回数据失败"
|
||
|
||
return_data["data"] = result_msg
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def shuma(self, server: str) -> Dict[str, Any]:
|
||
"""刷马"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_shuama", "GET", params=params, out_key=""
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
_data = data["data"]["data"]
|
||
result_msg = f"{server}\n"
|
||
result_msg += f"黑戈壁：\n{_data['黑戈壁'][0]}\n"
|
||
result_msg += f"阴山大草原：\n{_data['阴山大草原'][0]}\n"
|
||
result_msg += f"鲲鹏岛：\n{_data['鲲鹏岛'][0]}\n"
|
||
result_msg += f"的卢:\n{_data['龙泉府 / 进图（21:10）'][0]}\n"
|
||
result_msg += f"赤兔:\n{data['data']['note']}"
|
||
except Exception as e:
|
||
logger.error(f"处理返回数据失败: {e}")
|
||
return_data["msg"] = "处理返回数据失败"
|
||
|
||
return_data["data"] = result_msg
|
||
return_data["code"] = 200
|
||
|
||
return return_data |