413 lines
13 KiB
Python
413 lines
13 KiB
Python
from datetime import datetime
|
|
from typing import Any, Dict, List, Optional, Union
|
|
|
|
from astrbot.api import logger
|
|
|
|
from .base import DATA_PROCESSING_ERRORS, BaseDomainService
|
|
|
|
|
|
class InformationService(BaseDomainService):
|
|
async def keju(self, subject: str, limit: int) -> Dict[str, Any]:
|
|
"""科举"""
|
|
return_data = self.new_result()
|
|
|
|
# 1. 构造请求参数
|
|
params = {"subject": subject, "limit": limit}
|
|
|
|
# 2. 调用基础请求
|
|
data: Optional[List[Dict[str, Any]]] = await self.request(
|
|
"jx3_keju", 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 DATA_PROCESSING_ERRORS 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.new_result()
|
|
|
|
# 1. 构造请求参数
|
|
params = {"server": server, "name": name, "map": map}
|
|
|
|
# 2. 调用基础请求
|
|
data: Optional[Dict[str, Any]] = await self.request("jx3_huajia", params=params)
|
|
if not data:
|
|
return_data["msg"] = "未查询到相关内容"
|
|
return return_data
|
|
|
|
# 3. 处理返回数据
|
|
try:
|
|
return_data["data"]["data"] = data
|
|
return_data["data"]["server"] = server
|
|
|
|
except DATA_PROCESSING_ERRORS as e:
|
|
logger.error(f"数据处理时出错: {e}")
|
|
return_data["msg"] = "处理接口返回信息时出错"
|
|
return return_data
|
|
|
|
# 加载模板
|
|
if not await self.attach_template(return_data, "huajia.html"):
|
|
return return_data
|
|
|
|
return_data["code"] = 200
|
|
|
|
return return_data
|
|
|
|
async def zhuangshi(self, name: str) -> Dict[str, Any]:
|
|
"""装饰"""
|
|
return_data = self.new_result()
|
|
|
|
# 1. 构造请求参数
|
|
params = {"name": name}
|
|
|
|
# 2. 调用基础请求
|
|
data: Optional[Dict[str, Any]] = await self.request(
|
|
"jx3_zhuangshi", params=params
|
|
)
|
|
if not data:
|
|
return_data["msg"] = "未查询到相关内容"
|
|
return return_data
|
|
|
|
# 3. 处理返回数据
|
|
try:
|
|
return_data["data"]["data"] = data
|
|
|
|
except DATA_PROCESSING_ERRORS as e:
|
|
logger.error(f"数据处理时出错: {e}")
|
|
return_data["msg"] = "处理接口返回信息时出错"
|
|
return return_data
|
|
|
|
# 加载模板
|
|
if not await self.attach_template(return_data, "zhuangshi.html"):
|
|
return return_data
|
|
|
|
return_data["code"] = 200
|
|
|
|
return return_data
|
|
|
|
async def qiwu(self, name: str) -> Dict[str, Any]:
|
|
"""器物"""
|
|
return_data = self.new_result()
|
|
|
|
# 1. 构造请求参数
|
|
params = {"name": name}
|
|
|
|
# 2. 调用基础请求
|
|
data: Optional[Dict[str, Any]] = await self.request("jx3_qiwu", params=params)
|
|
if not data:
|
|
return_data["msg"] = "未查询到相关内容"
|
|
return return_data
|
|
|
|
# 3. 处理返回数据
|
|
try:
|
|
return_data["data"]["data"] = data
|
|
return_data["data"]["name"] = name
|
|
|
|
except DATA_PROCESSING_ERRORS as e:
|
|
logger.error(f"数据处理时出错: {e}")
|
|
return_data["msg"] = "处理接口返回信息时出错"
|
|
return return_data
|
|
|
|
# 加载模板
|
|
if not await self.attach_template(return_data, "qiwu.html"):
|
|
return return_data
|
|
|
|
return_data["code"] = 200
|
|
|
|
return return_data
|
|
|
|
async def xinwen(self, num: int) -> Dict[str, Any]:
|
|
"""新闻资讯"""
|
|
return_data = self.new_result()
|
|
|
|
params = {"limit": num}
|
|
# 提取字段可能返回列表
|
|
data: Optional[List[Dict[str, Any]]] = await self.request(
|
|
"jx3_xinweng", 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 DATA_PROCESSING_ERRORS 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.new_result()
|
|
|
|
params = {"limit": num}
|
|
# 提取字段可能返回列表
|
|
data: Optional[List[Dict[str, Any]]] = await self.request(
|
|
"jx3_weihu", 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 DATA_PROCESSING_ERRORS 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.new_result()
|
|
|
|
# 1. 构造请求参数
|
|
params = {"name": name}
|
|
|
|
# 2. 调用基础请求
|
|
data: Optional[Dict[str, Any]] = await self.request("jx3_qufu", 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 DATA_PROCESSING_ERRORS 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.new_result()
|
|
|
|
# 1. 构造请求参数
|
|
params = {"server": server, "type": "1"}
|
|
|
|
# 2. 调用基础请求
|
|
data: Optional[Dict[str, Union[int, str]]] = await self.request(
|
|
"jx3_zhuangtai", 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 DATA_PROCESSING_ERRORS 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.new_result()
|
|
|
|
# 1. 构造请求参数
|
|
params = {"server": "", "type": "2"}
|
|
|
|
data: Optional[List[Dict[str, Any]]] = await self.request(
|
|
"jx3_zhuangtai", 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 DATA_PROCESSING_ERRORS as e:
|
|
logger.error(f"数据处理时出错: {e}")
|
|
return_data["msg"] = "处理接口返回信息时出错"
|
|
# 加载模板
|
|
if not await self.attach_template(return_data, "qufuzhuangtai.html"):
|
|
return return_data
|
|
|
|
return_data["code"] = 200
|
|
|
|
return return_data
|
|
|
|
async def jigai(self) -> Dict[str, Any]:
|
|
"""技改记录"""
|
|
return_data = self.new_result()
|
|
|
|
# 提取字段可能返回列表
|
|
data: Optional[List[Dict[str, Any]]] = await self.request("jx3_jigai")
|
|
|
|
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 DATA_PROCESSING_ERRORS 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.new_result()
|
|
|
|
data: Optional[List[Dict[str, Any]]] = await self.request("jx3_xiaoyao")
|
|
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 DATA_PROCESSING_ERRORS as e:
|
|
logger.error(f"数据处理时出错: {e}")
|
|
return_data["msg"] = "处理接口返回信息时出错"
|
|
|
|
# 加载模板
|
|
if not await self.attach_template(return_data, "xiaoyao.html"):
|
|
return return_data
|
|
|
|
return_data["code"] = 200
|
|
|
|
return return_data
|