2532 lines
86 KiB
Python
2532 lines
86 KiB
Python
import json
|
||
from datetime import datetime
|
||
from typing import Dict, Any, Optional, List, Union
|
||
from collections import defaultdict
|
||
|
||
from astrbot.api import logger
|
||
from astrbot.api import AstrBotConfig
|
||
import astrbot.api.message_components as Comp
|
||
|
||
from .request import APIClient
|
||
from .sqlite import AsyncSQLiteDB
|
||
from .fun_basic import load_template,gold_to_string,week_to_num,compare_date_str
|
||
|
||
ROLE_RANK_NAMES = {
|
||
"名士五十强",
|
||
"老江湖五十强",
|
||
"兵甲藏家五十强",
|
||
"名师五十强",
|
||
"阵营英雄五十强",
|
||
"薪火相传五十强",
|
||
"庐园广记一百强",
|
||
}
|
||
|
||
TONG_RANK_NAMES = {
|
||
"浩气神兵宝甲五十强",
|
||
"恶人神兵宝甲五十强",
|
||
"浩气爱心帮会五十强",
|
||
"恶人爱心帮会五十强",
|
||
"赛季恶人五十强",
|
||
"赛季浩气五十强",
|
||
"上周恶人五十强",
|
||
"上周浩气五十强",
|
||
"本周恶人五十强",
|
||
"本周浩气五十强",
|
||
}
|
||
|
||
class JX3Service:
|
||
def __init__(self, api_data_path: str, config: AstrBotConfig, sqlite: AsyncSQLiteDB):
|
||
# 加载API配置文件
|
||
try:
|
||
with open(api_data_path, 'r', encoding='utf-8') as f:
|
||
self._api_config = json.load(f)
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载API配置文件失败: {e}")
|
||
self._api_config = {}
|
||
# 实例化 API Client
|
||
self._api: APIClient = APIClient()
|
||
# 引用插件配置文件
|
||
self._config = config
|
||
# 引用sqlite
|
||
self._sql_db = sqlite
|
||
|
||
# 获取配置中的 Token
|
||
self.token = self._config.get("jx3api_token", "")
|
||
if self.token == "":
|
||
logger.warning("获取配置token失败，请正确填写token,否则部分功能无法正常使用")
|
||
else:
|
||
logger.debug(f"获取配置token成功。{self.token}")
|
||
# 获取配置中的 ticket
|
||
self.ticket = self._config.get("jx3api_ticket", "")
|
||
if self.ticket == "":
|
||
logger.warning("获取配置ticket失败，请正确填写ticket,否则部分功能无法正常使用")
|
||
else:
|
||
logger.debug(f"获取配置ticket成功。{self.ticket}")
|
||
|
||
|
||
async def close(self):
|
||
"""释放底层 APIClient 资源"""
|
||
if self._api:
|
||
await self._api.close()
|
||
|
||
|
||
def _init_return_data(self) -> Dict[str, Any]:
|
||
"""初始化标准的返回数据结构"""
|
||
return {
|
||
"code": 0,
|
||
"msg": "功能函数未执行",
|
||
"data": {},
|
||
"temp": "",
|
||
"icons": {}
|
||
}
|
||
|
||
|
||
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:
|
||
if not self._api:
|
||
logger.error("API client is not initialized")
|
||
return None
|
||
|
||
api_config = self._api_config.get(config_key)
|
||
if not api_config:
|
||
logger.error(f"配置文件中未找到 key: {config_key}")
|
||
return None
|
||
|
||
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"] = await 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\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[1] or '无'}\n"
|
||
|
||
return_data["data"] = result_msg + luck_msg + card_msg + team_msg
|
||
return_data["code"] = 200
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
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
|
||
)
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
items = []
|
||
num_richang = week_to_num(data["data"][0]["week"]) or 0
|
||
# 空白数据
|
||
for _ in range(num_richang):
|
||
items.append({
|
||
"en": False,
|
||
"compare": "",
|
||
"date": "",
|
||
"war": "",
|
||
"battle": ""
|
||
})
|
||
|
||
# 真实数据
|
||
for m in data["data"]:
|
||
items.append({
|
||
"en": True,
|
||
"compare": compare_date_str(m.get("date", "")),
|
||
"date": m.get("date", ""),
|
||
"war": m.get("war", ""),
|
||
"battle": m.get("battle", "")
|
||
})
|
||
|
||
return_data["data"]["items"] = items
|
||
return_data["data"]["today"] = data["today"]
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("richangyuche.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def xingxiashijian(self,name: str) -> Dict[str, Any]:
|
||
"""行侠事件"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = { "name": name}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_xingxiashijian", "GET", params=params
|
||
)
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
return_data["data"]["items"] = data
|
||
return_data["data"]["name"] = name
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("xingxiashijian.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def keju(self,subject: str, limit: int) -> Dict[str, Any]:
|
||
"""科举"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"subject": subject, "limit": limit}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_keju", "GET", params=params
|
||
)
|
||
if not data:
|
||
return_data["msg"] = "未查询到相关题目"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
# 格式化字符串，利用字典的 get 方法提供默认值
|
||
result_msg = ""
|
||
for m in data:
|
||
result_msg += f"{m['id']}.{m['question']}\n"
|
||
result_msg += f"答案：{m['answer']}\n\n"
|
||
|
||
return_data["data"] = result_msg
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def huajia(self,server: str, name: str, map: str) -> Dict[str, Any]:
|
||
"""花价"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "name": name, "map": map}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_huajia", "GET", params=params
|
||
)
|
||
if not data:
|
||
return_data["msg"] = "未查询到相关内容"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
return_data["data"]["data"] = data
|
||
return_data["data"]["server"] = server
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("huajia.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def zhuangshi(self,name: str) -> Dict[str, Any]:
|
||
"""装饰"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = { "name": name}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_zhuangshi", "GET", params=params
|
||
)
|
||
if not data:
|
||
return_data["msg"] = "未查询到相关内容"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
return_data["data"]["data"] = data
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("zhuangshi.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def qiwu(self,name: str) -> Dict[str, Any]:
|
||
"""器物"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = { "name": name}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_qiwu", "GET", params=params
|
||
)
|
||
if not data:
|
||
return_data["msg"] = "未查询到相关内容"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
return_data["data"]["data"] = data
|
||
return_data["data"]["name"] = name
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("qiwu.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def xinwen(self, num:int) -> Dict[str, Any]:
|
||
"""新闻资讯"""
|
||
return_data = self._init_return_data()
|
||
|
||
params = {"limit": num}
|
||
# 提取字段可能返回列表
|
||
data: Optional[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_xinweng", "GET", params=params)
|
||
|
||
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[:num], 1):
|
||
result_msg += f"{i}. 【{item.get('type', '无类型')}】\n"
|
||
result_msg += f"标题：{item.get('title', '未知时间')}\n"
|
||
result_msg += f"时间：{item.get('date', '未知时间')}\n"
|
||
result_msg += f"链接：{item.get('url', '无链接')}\n"
|
||
|
||
return_data["data"] = result_msg
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def weihu(self, num:int) -> Dict[str, Any]:
|
||
"""维护"""
|
||
return_data = self._init_return_data()
|
||
|
||
params = {"limit": num}
|
||
# 提取字段可能返回列表
|
||
data: Optional[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_weihu", "GET", params=params)
|
||
|
||
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[:num], 1):
|
||
result_msg += f"{i}. 【{item.get('type', '无类型')}】\n"
|
||
result_msg += f"标题：{item.get('title', '未知时间')}\n"
|
||
result_msg += f"时间：{item.get('date', '未知时间')}\n"
|
||
result_msg += f"链接：{item.get('url', '无链接')}\n"
|
||
|
||
return_data["data"] = result_msg
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
async def qufu(self,name: str) -> Dict[str, Any]:
|
||
"""区服信息"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"name": name}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_qufu", "GET", params=params
|
||
)
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
# 格式化字符串，利用字典的 get 方法提供默认值
|
||
result_msg = f"主服：{data.get('zone', '无')}-{data.get('name', '无')}\n"
|
||
|
||
slave = data.get('slave', [])
|
||
slave_msg = f"区服：{', '.join(slave)}\n"
|
||
|
||
alias = data.get('alias', [])
|
||
alias_msg = f"别名：{', '.join(alias)}\n"
|
||
|
||
return_data["data"] = result_msg + slave_msg + alias_msg
|
||
return_data["code"] = 200
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def kaifu(self, server: str) -> Dict[str, Any]:
|
||
"""开服状态查询"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server,"type": "1"}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Union[int, str]]] = await self._base_request(
|
||
"jx3_zhuangtai", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
status = data.get("status", 0)
|
||
lasttime = data.get("lasttime", 0)
|
||
shuttime = data.get("shuttime", 0)
|
||
|
||
lasttime_t = datetime.fromtimestamp(float(lasttime)).strftime("%Y-%m-%d %H:%M:%S")
|
||
shuttime_t = datetime.fromtimestamp(float(shuttime)).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
if status == 1:
|
||
status_str = f"{server}服务器已开服，快冲，快冲！\n开服时间：{lasttime_t}"
|
||
status_bool = True
|
||
else:
|
||
status_str = f"{server}服务器当前维护中，等会再来吧！\n维护时间：{shuttime_t}"
|
||
status_bool = False
|
||
|
||
return_data["status"] = status_bool
|
||
return_data["data"] = status_str
|
||
|
||
except Exception as e:
|
||
logger.error(f"kaifu 数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def zhuangtai(self) -> Dict[str, Any]:
|
||
"""区服状态"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": "","type": "2"}
|
||
|
||
data: Optional[List[Dict[str, Any]]] = await self._base_request("jx3_zhuangtai", "GET", params=params)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
# 处理返回数据
|
||
try:
|
||
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
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = await 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"
|
||
|
||
for i, item in enumerate(data[:3], 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
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def xiaoyao(self) -> Dict[str, Any]:
|
||
"""小药"""
|
||
return_data = self._init_return_data()
|
||
|
||
data: Optional[List[Dict[str, Any]]] = await self._base_request("jx3_xiaoyao", "GET")
|
||
logger.debug(data)
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 处理返回数据
|
||
try:
|
||
result = {}
|
||
|
||
for item in data:
|
||
k = item["kungfu"]
|
||
color = item["color"]
|
||
cls = item["class"]
|
||
name = item["name"]
|
||
|
||
if k not in result:
|
||
result[k] = {
|
||
"kungfu": k,
|
||
"purple": {},
|
||
"blue": {}
|
||
}
|
||
|
||
if color == "紫":
|
||
result[k]["purple"][cls] = name
|
||
else:
|
||
result[k]["blue"][cls] = name
|
||
|
||
return_data["data"]["items"] = list(result.values())
|
||
logger.debug(return_data["data"])
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("xiaoyao.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def zhenyan(self, name: str) -> Dict[str, Any]:
|
||
"""阵眼"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"name": name, "ticket": self.ticket, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_zhenyan", "GET", params=params
|
||
)
|
||
|
||
if not data or not isinstance(data, dict):
|
||
return_data["msg"] = "未查询到该心法阵眼信息"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
items = data.get("data", [])
|
||
if not isinstance(items, list) or not items:
|
||
return_data["msg"] = "未查询到该心法阵眼信息"
|
||
return return_data
|
||
|
||
result_msg = f"{data.get('name', name)}-{data.get('skillName', '')}\n"
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
result_msg += f"{item.get('name', '')}：{item.get('desc', '')}\n"
|
||
|
||
return_data["data"] = result_msg.rstrip()
|
||
except Exception as e:
|
||
logger.error(f"处理阵眼数据失败: {e}")
|
||
return_data["msg"] = "处理阵眼数据失败"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def qixue(self, name: str) -> Dict[str, Any]:
|
||
"""奇穴"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"name": name, "ticket": self.ticket, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_qixue", "GET", params=params
|
||
)
|
||
|
||
if not data or not isinstance(data, list):
|
||
return_data["msg"] = "未查询到该心法奇穴信息"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
level_titles = {
|
||
1: "主奇穴",
|
||
2: "第一重",
|
||
3: "第二重",
|
||
4: "第三重",
|
||
5: "第四重",
|
||
6: "第五重",
|
||
7: "第六重",
|
||
8: "混池",
|
||
}
|
||
|
||
groups = []
|
||
for group in data:
|
||
if not isinstance(group, dict):
|
||
continue
|
||
|
||
level = group.get("level")
|
||
try:
|
||
level_key = int(level)
|
||
except (TypeError, ValueError):
|
||
level_key = 0
|
||
|
||
items = group.get("data", [])
|
||
if not isinstance(items, list) or not items:
|
||
continue
|
||
|
||
parsed_items = []
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
|
||
class_value = item.get("class")
|
||
is_active = class_value == 1 or str(class_value) == "1"
|
||
parsed_items.append({
|
||
"name": item.get("name", ""),
|
||
"icon": item.get("icon", ""),
|
||
"desc": item.get("desc", ""),
|
||
"class_text": "主动" if is_active else "被动",
|
||
"interval": item.get("interval", "") if is_active else "",
|
||
})
|
||
|
||
if parsed_items:
|
||
groups.append({
|
||
"level": level_key,
|
||
"title": level_titles.get(level_key, f"第{level_key}组"),
|
||
"talents": parsed_items,
|
||
})
|
||
|
||
if not groups:
|
||
return_data["msg"] = "未查询到该心法奇穴信息"
|
||
return return_data
|
||
|
||
groups.sort(key=lambda item: item["level"])
|
||
return_data["data"] = {
|
||
"name": name,
|
||
"groups": groups,
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"处理奇穴数据失败: {e}")
|
||
return_data["msg"] = "处理奇穴数据失败"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("qixue.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
return return_data
|
||
|
||
|
||
async def jineng(self, name: str) -> Dict[str, Any]:
|
||
"""技能"""
|
||
return_data = self._init_return_data()
|
||
|
||
params = {"name": name, "ticket": self.ticket, "token": self.token}
|
||
|
||
data: Optional[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_jineng", "GET", params=params
|
||
)
|
||
|
||
if not data or not isinstance(data, list):
|
||
return_data["msg"] = "未查询到该心法技能信息"
|
||
return return_data
|
||
|
||
try:
|
||
groups = []
|
||
for group in data:
|
||
if not isinstance(group, dict):
|
||
continue
|
||
|
||
skills = group.get("data", [])
|
||
if not isinstance(skills, list) or not skills:
|
||
continue
|
||
|
||
parsed_skills = []
|
||
for skill in skills:
|
||
if not isinstance(skill, dict):
|
||
continue
|
||
|
||
parsed_skills.append({
|
||
"name": skill.get("name", ""),
|
||
"icon": skill.get("icon", ""),
|
||
"desc": skill.get("desc", ""),
|
||
"interval": skill.get("interval", ""),
|
||
"distance": skill.get("distance", ""),
|
||
"release_type": skill.get("releaseType", ""),
|
||
"weapon": skill.get("weapon", ""),
|
||
})
|
||
|
||
if parsed_skills:
|
||
groups.append({
|
||
"title": group.get("class", "其他技能"),
|
||
"skills": parsed_skills,
|
||
})
|
||
|
||
if not groups:
|
||
return_data["msg"] = "未查询到该心法技能信息"
|
||
return return_data
|
||
|
||
return_data["data"] = {
|
||
"name": name,
|
||
"groups": groups,
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"处理技能数据失败: {e}")
|
||
return_data["msg"] = "处理技能数据失败"
|
||
return return_data
|
||
|
||
try:
|
||
return_data["temp"] = await load_template("jineng.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def zilipaixing(self, school: str, server: str) -> Dict[str, Any]:
|
||
"""资历排行"""
|
||
return_data = self._init_return_data()
|
||
|
||
params = {
|
||
"server": server,
|
||
"school": school,
|
||
"ticket": self.ticket,
|
||
"token": self.token,
|
||
}
|
||
|
||
data: Optional[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_zilipaixing", "GET", params=params
|
||
)
|
||
|
||
if not data or not isinstance(data, list):
|
||
return_data["msg"] = "未查询到资历排行信息"
|
||
return return_data
|
||
|
||
try:
|
||
return_data["data"] = {
|
||
"items": data,
|
||
"school": school,
|
||
"server": server,
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"处理资历排行数据失败: {e}")
|
||
return_data["msg"] = "处理资历排行数据失败"
|
||
return return_data
|
||
|
||
try:
|
||
return_data["temp"] = await load_template("zilipaixing.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
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
|
||
else:
|
||
return_data["msg"] = "接口未返回文本"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
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
|
||
else:
|
||
return_data["msg"] = "接口未返回图片URL"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
|
||
async def baizhan(self) -> Dict[str, Any]:
|
||
"""百战首领"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = { "token": self.token}
|
||
|
||
data: Optional[Dict[str, Any]] = await self._base_request("jx3_baizhan", "GET", params=params)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
# 处理返回数据
|
||
|
||
try:
|
||
return_data["data"] = data
|
||
return_data["data"]["start"] = datetime.fromtimestamp(float(data["start"])).strftime("%Y-%m-%d %H:%M:%S")
|
||
return_data["data"]["end"] = datetime.fromtimestamp(float(data["end"])).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("baizhan.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_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[list[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_time = datetime.fromtimestamp(data_new['time']).strftime('%Y-%m-%d %H:%M:%S')
|
||
result_msg = f"{server}\n"
|
||
if data_new["status"] == 0:
|
||
result_msg += f"本轮扶摇九天尚未开启\n"
|
||
if data_new["status"] == 1:
|
||
result_msg += f"本轮扶摇九天正在进行\n"
|
||
if data_new["status"] == 2:
|
||
result_msg += f"本轮扶摇九天已经结束\n"
|
||
result_msg += f"开启时间：{data_time}\n"
|
||
return_data["data"] = result_msg
|
||
except Exception 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[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_zhengyingpaimai", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
for item in data:
|
||
item["time"] = datetime.fromtimestamp(item["time"]).strftime("%Y-%m-%d %H:%M:%S")
|
||
return_data["data"]["list"] = data
|
||
except Exception as e:
|
||
logger.error(f"处理返回数据失败: {e}")
|
||
return_data["msg"] = "处理返回数据失败"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("zhengyingpaimai.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_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[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_dilujilu", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
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")
|
||
return_data["data"]["list"] = data
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("dilujilu.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_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[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_yanhuachaxun", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
for item in data:
|
||
if isinstance(item, dict):
|
||
timestamp = item.get("time")
|
||
if timestamp and isinstance(timestamp, (int, float)):
|
||
item["time"] = datetime.fromtimestamp(timestamp).strftime("%Y-%m-%d %H:%M:%S")
|
||
else:
|
||
item["time"] = "未知时间"
|
||
|
||
return_data["data"]["list"] = data
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("yanhuan.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def pianzhi(self, uid: str) -> Dict[str, Any]:
|
||
"""骗子查询"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"uid": uid, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_pianzhi", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
records = data["records"]
|
||
|
||
if not records:
|
||
result_msg = "未找到该用户行骗记录，很棒！继续保持！"
|
||
else:
|
||
result_msg = ""
|
||
|
||
for record in records:
|
||
result_msg += f"区服：{record['server']} 标签：{record['tieba']}\n\n"
|
||
|
||
for item in record["data"]:
|
||
result_msg += f"标题：{item['title']}\n"
|
||
result_msg += f"地址：{item['url']}\n"
|
||
result_msg += f"ID：{item['tid']}\n"
|
||
result_msg += f"内容：{item['text']}\n"
|
||
result_msg += (
|
||
f"时间：{datetime.fromtimestamp(item['time']).strftime('%Y-%m-%d %H:%M:%S')}\n\n"
|
||
)
|
||
|
||
result_msg += "\n\n"
|
||
return_data["data"] = result_msg
|
||
except Exception as e:
|
||
logger.error(f"处理返回数据失败: {e}")
|
||
return_data["msg"] = "处理返回数据失败"
|
||
return return_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[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_juesheqiyu", "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"] = "处理返回数据失败"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await 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 weizuoqiyu(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[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_weizuoqiyu", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "未查询到未做奇遇信息"
|
||
return return_data
|
||
|
||
if not isinstance(data, list):
|
||
return_data["msg"] = "获取接口信息失败或数据格式错误"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
return_data["data"]["ptqy"] = []
|
||
return_data["data"]["jsqy"] = []
|
||
return_data["data"]["name"] = name
|
||
return_data["data"]["server"] = server
|
||
|
||
for item in data:
|
||
event_name = item.get("name") or item.get("event")
|
||
if not event_name:
|
||
continue
|
||
|
||
try:
|
||
level = int(item.get("level", 0))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
|
||
event_item = {
|
||
"event": event_name,
|
||
"time": "未触发"
|
||
}
|
||
|
||
if level == 1:
|
||
return_data["data"]["ptqy"].append(event_item)
|
||
elif level == 2:
|
||
return_data["data"]["jsqy"].append(event_item)
|
||
|
||
if not return_data["data"]["ptqy"] and not return_data["data"]["jsqy"]:
|
||
return_data["msg"] = "未查询到未做普通或绝世奇遇"
|
||
return return_data
|
||
except Exception as e:
|
||
logger.error(f"处理返回数据失败: {e}")
|
||
return_data["msg"] = "处理返回数据失败"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("weizuoqiyu.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def qiyutongji(self, name: str, server: str) -> Dict[str, Any]:
|
||
"""奇遇统计"""
|
||
return_data = self._init_return_data()
|
||
|
||
params = {"server": server, "name": name, "token": self.token}
|
||
|
||
data_list: Optional[List[Dict[str, Any]]] = await self._base_request("jx3_qiyutongji", "GET", params=params)
|
||
|
||
if not data_list or not isinstance(data_list, list):
|
||
return_data["msg"] = "获取接口信息失败或数据格式错误"
|
||
return return_data
|
||
|
||
# 格式化时间
|
||
try:
|
||
for item in data_list:
|
||
item["time"] = datetime.fromtimestamp(item["time"]).strftime("%Y-%m-%d %H:%M:%S")
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 准备模板渲染数据
|
||
try:
|
||
return_data["data"] = {
|
||
"items": data_list,
|
||
"server": server,
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"qiyuname": name
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"模板数据准备失败: {e}")
|
||
return_data["msg"] = "系统错误：模板渲染数据准备失败"
|
||
return return_data
|
||
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("qiyuliebiao.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def jinqiqiyu(self, server: str) -> Dict[str, Any]:
|
||
"""近期奇遇"""
|
||
return_data = self._init_return_data()
|
||
|
||
params = {"server": server, "token": self.token}
|
||
|
||
data_list: Optional[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_jinqiqiyu", "GET", params=params
|
||
)
|
||
|
||
if not data_list or not isinstance(data_list, list):
|
||
return_data["msg"] = "未查询到近期奇遇信息"
|
||
return return_data
|
||
|
||
# 格式化时间
|
||
try:
|
||
for item in data_list:
|
||
event_time = item.get("time")
|
||
if event_time:
|
||
try:
|
||
item["time"] = datetime.fromtimestamp(int(event_time)).strftime("%Y-%m-%d %H:%M:%S")
|
||
except (TypeError, ValueError, OSError):
|
||
item["time"] = str(event_time)
|
||
else:
|
||
item["time"] = ""
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 准备模板渲染数据
|
||
try:
|
||
return_data["data"] = {
|
||
"items": data_list,
|
||
"server": server,
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"模板数据准备失败: {e}")
|
||
return_data["msg"] = "系统错误：模板渲染数据准备失败"
|
||
return return_data
|
||
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("jinqiqiyu.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def qiyuhuizong(self, server: str, num: int = 7) -> Dict[str, Any]:
|
||
"""奇遇汇总"""
|
||
return_data = self._init_return_data()
|
||
|
||
if num < 1 or num > 30:
|
||
return_data["msg"] = "汇总天数需在 1-30 天之间"
|
||
return return_data
|
||
|
||
params = {"server": server, "num": num, "token": self.token}
|
||
|
||
data_list: Optional[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_qiyuhuizong", "GET", params=params
|
||
)
|
||
|
||
if not data_list or not isinstance(data_list, list):
|
||
return_data["msg"] = "未查询到奇遇汇总信息"
|
||
return return_data
|
||
|
||
# 整理最近一次触发信息
|
||
try:
|
||
for item in data_list:
|
||
latest = item.get("data")
|
||
if not isinstance(latest, dict):
|
||
latest = {}
|
||
|
||
item["latest_name"] = latest.get("name", "")
|
||
|
||
latest_time = latest.get("time")
|
||
if latest_time:
|
||
try:
|
||
item["latest_time"] = datetime.fromtimestamp(int(latest_time)).strftime("%Y-%m-%d %H:%M:%S")
|
||
except (TypeError, ValueError, OSError):
|
||
item["latest_time"] = str(latest_time)
|
||
else:
|
||
item["latest_time"] = ""
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 准备模板渲染数据
|
||
try:
|
||
return_data["data"] = {
|
||
"items": data_list,
|
||
"server": server,
|
||
"num": num,
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"模板数据准备失败: {e}")
|
||
return_data["msg"] = "系统错误：模板渲染数据准备失败"
|
||
return return_data
|
||
|
||
# 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("qiyuhuizong.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_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
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await 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 mingjianpaihang(self, limit: str, mode:str) -> Dict[str, Any]:
|
||
"""名剑排行"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"limit": limit, "mode":mode, "token": self.token, "ticket": self.ticket}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_mingjianpaihang", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口数据失败"
|
||
return return_data
|
||
|
||
return_data["data"]["lists"] = data
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("mingjianpaihang.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def mingjiantongji(self, mode: str = "33") -> Dict[str, Any]:
|
||
"""名剑统计"""
|
||
return_data = self._init_return_data()
|
||
|
||
mode = str(mode)
|
||
if mode not in {"22", "33", "55"}:
|
||
return_data["msg"] = "模式错误，可选值：22、33、55"
|
||
return return_data
|
||
|
||
# 1. 构造请求参数
|
||
params = {"mode": mode, "token": self.token, "ticket": self.ticket}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_mingjiantongji", "GET", params=params
|
||
)
|
||
|
||
if not data or not isinstance(data, list):
|
||
return_data["msg"] = "未查询到名剑统计信息"
|
||
return return_data
|
||
|
||
# 3. 准备模板渲染数据
|
||
try:
|
||
return_data["data"] = {
|
||
"items": data,
|
||
"mode": mode,
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"模板数据准备失败: {e}")
|
||
return_data["msg"] = "系统错误：模板渲染数据准备失败"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("mingjiantongji.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_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. 处理返回数据
|
||
try:
|
||
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']}"
|
||
return_data["data"]["list"] = data["data"]
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("tuanduizhaomu.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def shitu(self, type_value: int, keyword: str, server: str) -> Dict[str, Any]:
|
||
"""师徒招募"""
|
||
return_data = self._init_return_data()
|
||
|
||
if type_value not in {1, 2}:
|
||
return_data["msg"] = "师徒查询类型错误"
|
||
return return_data
|
||
|
||
# 1. 构造请求参数
|
||
params = {"type": type_value, "server": server, "keyword": keyword, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_shitu", "GET", params=params
|
||
)
|
||
|
||
if not data or not isinstance(data, dict):
|
||
return_data["msg"] = "未查询到师徒招募信息"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
items = data.get("data", [])
|
||
if not items:
|
||
return_data["msg"] = "未查询到师徒招募信息"
|
||
return return_data
|
||
|
||
title = "收徒信息" if type_value == 1 else "拜师信息"
|
||
return_data["data"] = {
|
||
"items": items,
|
||
"server": data.get("server", server),
|
||
"keyword": keyword,
|
||
"type_value": type_value,
|
||
"title": title,
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("shitu.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def rank_statistical(self, name: str, server: str) -> Dict[str, Any]:
|
||
"""各类风云录榜单"""
|
||
return_data = self._init_return_data()
|
||
|
||
if name in ROLE_RANK_NAMES:
|
||
template_name = "rank_role.html"
|
||
elif name in TONG_RANK_NAMES:
|
||
template_name = "rank_tong.html"
|
||
else:
|
||
return_data["msg"] = "榜单名称错误"
|
||
return return_data
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "name": name, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3_rank_statistical", "GET", params=params
|
||
)
|
||
|
||
if not data or not isinstance(data, dict):
|
||
return_data["msg"] = "未查询到榜单信息"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
items = data.get("data", [])
|
||
if not isinstance(items, list) or not items:
|
||
return_data["msg"] = "未查询到榜单信息"
|
||
return return_data
|
||
|
||
return_data["data"] = {
|
||
"items": items,
|
||
"server": data.get("server", server),
|
||
"rank_name": data.get("name", name),
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template(template_name)
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def jueshe(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_jueshexinxi", "GET", params=params
|
||
)
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
# 格式化字符串，利用字典的 get 方法提供默认值
|
||
result_msg = (
|
||
f"服务器：{data.get('zoneName', '无')}·{data.get('serverName', '无')}\n"
|
||
f"名称：{data.get('roleName', '无')}\n"
|
||
f"角色ID：{data.get('roleId', '无')}\n"
|
||
f"全区ID：{data.get('globalRoleId', '无')}\n"
|
||
f"职业：{data.get('forceName', '无')}·{data.get('bodyName', '无')}\n"
|
||
f"帮会：{data.get('tongName', '无')}\n"
|
||
f"阵营：{data.get('campName', '无')}\n"
|
||
)
|
||
|
||
return_data["data"] = result_msg
|
||
return_data["code"] = 200
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
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 or not isinstance(data, dict):
|
||
return_data["msg"] = "未找到该角色名片"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
url = data.get("showAvatar")
|
||
if not url:
|
||
return_data["msg"] = "未获取到名片图片"
|
||
return return_data
|
||
|
||
server_name = data.get("serverName", server)
|
||
role_name = data.get("roleName", name)
|
||
show_like = data.get("showLike", 0)
|
||
msg = f"{server_name}-{role_name} 点赞：{show_like}"
|
||
|
||
return_data["data"] = [
|
||
Comp.Plain(msg),
|
||
Comp.Image.fromURL(url),
|
||
]
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def jingnai(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_jingnai", "GET", params=params
|
||
)
|
||
|
||
if not data or not isinstance(data, dict):
|
||
return_data["msg"] = "未查询到角色精耐信息"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
items = data.get("skill_list", [])
|
||
if not isinstance(items, list) or not items:
|
||
return_data["msg"] = "未查询到技能信息"
|
||
return return_data
|
||
|
||
color_map = {
|
||
7: "黑色",
|
||
6: "紫色",
|
||
5: "红色",
|
||
4: "绿色",
|
||
3: "蓝色",
|
||
2: "黄色",
|
||
0: "无",
|
||
}
|
||
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
|
||
color = item.get("skill_color")
|
||
try:
|
||
color_key = int(color)
|
||
except (TypeError, ValueError):
|
||
color_key = None
|
||
item["skill_color_text"] = color_map.get(color_key, str(color) if color is not None else "")
|
||
|
||
update_time = data.get("update_time")
|
||
if update_time:
|
||
try:
|
||
update_time = datetime.fromtimestamp(int(update_time)).strftime("%Y-%m-%d %H:%M:%S")
|
||
except (TypeError, ValueError, OSError):
|
||
update_time = str(update_time)
|
||
else:
|
||
update_time = ""
|
||
|
||
return_data["data"] = {
|
||
"server": data.get("server", server),
|
||
"role_name": data.get("role_name", name),
|
||
"skill_energy": data.get("skill_energy", ""),
|
||
"skill_stamina": data.get("skill_stamina", ""),
|
||
"skill_count": data.get("skill_count", len(items)),
|
||
"items": items,
|
||
"update_time": update_time
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"处理精耐数据失败: {e}")
|
||
return_data["msg"] = "处理精耐数据失败"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
try:
|
||
return_data["temp"] = await load_template("jingnai.html")
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def shuoyoumingpian(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[list[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_allmingpian", "GET", params=params
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "未找到该角色"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
chain = []
|
||
for m in data:
|
||
status = "已启用" if m.get("showActive") else "未启用"
|
||
msg = f"第{m.get('showIndex')}张 {status}"
|
||
url = m.get("showAvatar")
|
||
|
||
if not url:
|
||
logger.warning(f"第{m.get('showIndex')}张名片缺少图片URL，已跳过")
|
||
continue
|
||
|
||
chain.extend([
|
||
Comp.Plain(msg),
|
||
Comp.Image.fromURL(url),
|
||
])
|
||
|
||
if not chain:
|
||
return_data["msg"] = "未获取到有效的名片数据"
|
||
return return_data
|
||
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
return_data["data"] = chain
|
||
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['showAvatar']
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def shuamamsg(self,server:str,type:str,subtype:str) -> Dict[str, Any]:
|
||
"""刷马消息"""
|
||
return_data = self._init_return_data()
|
||
|
||
params = {"server": server, "type": type, "subtype": subtype}
|
||
# 提取字段可能返回字典
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3box_shuamamsg", "GET", params=params)
|
||
|
||
if not data:
|
||
return_data["msg"] = "获取接口信息失败或数据格式错误"
|
||
return return_data
|
||
|
||
try:
|
||
#
|
||
new_msg = data.get("list",[])[0]
|
||
return_data["status"] = new_msg.get('id')
|
||
|
||
result_msg = f"{server}\n"
|
||
result_msg += f"{new_msg.get('content')}\n"
|
||
result_msg += f"{new_msg.get('created_at')}\n"
|
||
|
||
return_data["data"] = result_msg
|
||
except Exception as e:
|
||
logger.error(f"数据处理时出错: {e}")
|
||
return_data["msg"] = "处理接口返回信息时出错"
|
||
return return_data
|
||
|
||
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['data']['note']}"
|
||
return_data["data"] = result_msg
|
||
except Exception as e:
|
||
logger.error(f"处理返回数据失败: {e}")
|
||
return_data["msg"] = "处理返回数据失败"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def jinjia(self, server: str, limit:str) -> Dict[str, Any]:
|
||
"""区服金价"""
|
||
return_data = self._init_return_data()
|
||
|
||
|
||
params = {"server": server, "limit": limit, "token": self.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"] = await 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
|
||
|
||
except Exception as e:
|
||
logger.error(f"模板数据准备失败: {e}")
|
||
return_data["msg"] = "系统错误：模板渲染数据准备失败"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
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"] = await 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[List[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:
|
||
if isinstance(item, dict):
|
||
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)
|
||
return_data["data"]["list"] = result
|
||
except Exception as e:
|
||
logger.error(f"处理交易行数据失败: {e}")
|
||
return_data["msg"] = "处理交易行数据失败"
|
||
return return_data
|
||
|
||
# 5. 模板渲染
|
||
try:
|
||
return_data["temp"] = await load_template("jiaoyihang.html")
|
||
return_data["code"] = 200
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return return_data
|
||
|
||
|
||
async def diaoluo(self, name: str, limit: int = 20, server: str = "") -> Dict[str, Any]:
|
||
"""物品掉落记录"""
|
||
return_data = self._init_return_data()
|
||
|
||
try:
|
||
limit = int(limit)
|
||
except (TypeError, ValueError):
|
||
return_data["msg"] = "数量必须是数字"
|
||
return return_data
|
||
|
||
if limit < 1 or limit > 100:
|
||
return_data["msg"] = "掉落记录数量需在 1-100 之间"
|
||
return return_data
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "name": name, "limit": limit, "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[List[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_diaoluo", "GET", params=params
|
||
)
|
||
|
||
if not data or not isinstance(data, list):
|
||
return_data["msg"] = "未查询到掉落记录"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
for item in data:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
|
||
drop_time = item.get("time")
|
||
if drop_time:
|
||
try:
|
||
item["time"] = datetime.fromtimestamp(int(drop_time)).strftime("%Y-%m-%d %H:%M:%S")
|
||
except (TypeError, ValueError, OSError):
|
||
item["time"] = str(drop_time)
|
||
else:
|
||
item["time"] = ""
|
||
|
||
return_data["data"] = {
|
||
"items": data,
|
||
"name": name,
|
||
"limit": limit,
|
||
"server": server,
|
||
"update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
}
|
||
except Exception as e:
|
||
logger.error(f"处理掉落数据失败: {e}")
|
||
return_data["msg"] = "处理掉落数据失败"
|
||
return return_data
|
||
|
||
# 4. 模板渲染
|
||
try:
|
||
return_data["temp"] = await load_template("diaoluo.html")
|
||
return_data["code"] = 200
|
||
except FileNotFoundError as e:
|
||
logger.error(f"加载模板失败: {e}")
|
||
return_data["msg"] = "系统错误：模板文件不存在"
|
||
return return_data
|
||
|
||
return return_data
|
||
|
||
|
||
async def bagua(self, type: str) -> Dict[str, Any]:
|
||
"""八卦"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"class": type, "limit": "5", "token": self.token}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[list[Dict[str, Any]]] = await self._base_request(
|
||
"jx3_bagua", "GET", params=params
|
||
)
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
if not data:
|
||
result_msg = f"未找到相关 {type} 记录。\n"
|
||
result_msg += f"可选范围：818 616 鬼网三 鬼网3 树洞 记录 教程 街拍 故事 避雷 吐槽 提问"
|
||
else:
|
||
result_msg = f"类型：【{type}】\n\n"
|
||
|
||
for item in data:
|
||
result_msg += f"{item['title']}\n"
|
||
result_msg += f"分区：{item['zone']} 服务器：{item['server']}\n"
|
||
result_msg += f"所属吧：{item['name']}\n"
|
||
result_msg += f"链接：https://tieba.baidu.com/p/{item['url']}\n"
|
||
result_msg += f"日期：{item['date']}\n\n"
|
||
return_data["data"] = result_msg
|
||
except Exception as e:
|
||
logger.exception("处理返回数据失败")
|
||
return_data["msg"] = "处理返回数据失败"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def qiyugonglue(self, name: str) -> Dict[str, Any]:
|
||
"""奇遇攻略"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"name": name}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[list[Dict[str, Any]]] = await self._base_request(
|
||
"jx3box_qiyugonglue", "GET", params=params, out_key="list"
|
||
)
|
||
|
||
if not data:
|
||
return_data["msg"] = "未找到该奇遇"
|
||
return return_data
|
||
|
||
# 提取dwID
|
||
dwID = data[0]["dwID"]
|
||
url = f"https://node.jx3box.com/serendipity/{dwID}/achievement"
|
||
logger.debug(f"获取ID接口地址：{url}")
|
||
|
||
data1 = await self._api.get(url)
|
||
url1 = f"https://cms.jx3box.com/api/cms/wiki/post/type/achievement/source/{data1['achievement_id']}"
|
||
logger.debug(f"获取攻略接口地址：{url1}")
|
||
|
||
# 获取奇遇攻略
|
||
data2 = await self._api.get(url1, out_key="data")
|
||
if not data2:
|
||
return_data["msg"] = "获取攻略数据异常"
|
||
return return_data
|
||
|
||
# 4. 处理数据
|
||
try:
|
||
return_data["data"] = {}
|
||
content = data2["post"]["content"]
|
||
return_data["temp"] = content
|
||
except Exception as e:
|
||
logger.exception("处理返回数据失败")
|
||
return_data["msg"] = "处理返回数据失败"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def hong1(self, name: str) -> Dict[str, Any]:
|
||
"""宏 心法"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 数据库查询数据
|
||
result = await self._sql_db.select_one(
|
||
"kungfu",
|
||
"name=? OR name1=? OR name2=? OR name3=? OR name4=? OR name5=?",
|
||
(name, name, name, name, name, name)
|
||
)
|
||
|
||
if result is None:
|
||
return_data["msg"] = "未找到该心法"
|
||
return return_data
|
||
|
||
kungfu = result.get("name", None)
|
||
if kungfu is None:
|
||
return_data["msg"] = "未找到该心法"
|
||
return return_data
|
||
|
||
logger.debug(f"查询到数据：{kungfu}")
|
||
|
||
# 1. 构造请求参数
|
||
params = {"subtype": kungfu}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3box_hong", "GET", params=params
|
||
)
|
||
logger.debug(data)
|
||
if not data:
|
||
return_data["msg"] = "未找到该心法一键宏"
|
||
return return_data
|
||
|
||
# 提取ID
|
||
pid_list = [0]
|
||
msg = "按照热度排列\n"
|
||
n = 1
|
||
try:
|
||
for m in data.get('list',[]):
|
||
msg += f"{n}、{m['author']}\t{m['post_title']}\n"
|
||
pid_list.append(m["ID"])
|
||
n += 1
|
||
|
||
return_data["msg"] = msg
|
||
return_data["data"]["list"] = pid_list
|
||
return_data["data"]["num"] = n
|
||
|
||
except Exception as e:
|
||
logger.exception("处理返回数据失败")
|
||
return_data["msg"] = "处理返回数据失败"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def hong2(self, pid: str) -> Dict[str, Any]:
|
||
"""宏 心法"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 发起请求
|
||
url = f"https://cms.jx3box.com/api/cms/post/{pid}"
|
||
logger.debug(f"获取宏接口地址：{url}")
|
||
|
||
data = await self._api.get(url, out_key="data")
|
||
# 验证数据
|
||
if not data:
|
||
return_data["msg"] = "获取宏数据异常"
|
||
return return_data
|
||
|
||
# 4. 处理数据
|
||
try:
|
||
return_data["temp"] = data["post_content"]
|
||
msg = ""
|
||
for m in data["post_meta"]["data"]:
|
||
msg += f"【宏名称】\n{m['name']}\n"
|
||
msg += f"【使用说明】\n{m['desc']}\n"
|
||
msg += f"【宏脚本】\n{m['macro']}\n\n"
|
||
|
||
return_data["data"] = msg
|
||
|
||
except Exception as e:
|
||
logger.exception("处理返回数据失败")
|
||
return_data["msg"] = "处理返回数据失败"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|
||
async def peizhuang(self, name: str, tags: str) -> Dict[str, Any]:
|
||
"""配装"""
|
||
return_data = self._init_return_data()
|
||
|
||
# 数据库查询数据
|
||
result = await self._sql_db.select_one(
|
||
"kungfu",
|
||
"name=? OR name1=? OR name2=? OR name3=? OR name4=? OR name5=?",
|
||
(name, name, name, name, name, name)
|
||
)
|
||
logger.debug(result)
|
||
if result is None:
|
||
return_data["msg"] = "未找到该心法"
|
||
return return_data
|
||
|
||
mount = result.get("pzid", None)
|
||
if not mount:
|
||
return_data["msg"] = "未找到该心法"
|
||
return return_data
|
||
|
||
logger.debug(f"查询到数据：{mount}")
|
||
|
||
# 1. 构造请求参数
|
||
params = {"mount": mount, "tags": tags}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self._base_request(
|
||
"jx3box_peizhuang", "GET", params=params
|
||
)
|
||
logger.debug(data)
|
||
# 验证数据
|
||
if not data:
|
||
return_data["msg"] = "配装数据获取异常"
|
||
return return_data
|
||
|
||
# 3. 处理返回数据
|
||
try:
|
||
result_msg = f"{name}--配装\n"
|
||
for item in data["list"]:
|
||
result_msg += f"【{item['zlp']}】--{item['title']}\n"
|
||
result_msg += f"链接：https://www.jx3box.com/pz/view/{item['id']}\n\n"
|
||
|
||
return_data["data"] = result_msg
|
||
|
||
except Exception as e:
|
||
logger.exception("处理返回数据失败:",e)
|
||
return_data["msg"] = "处理返回数据失败"
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
|