425 lines
14 KiB
Python
425 lines
14 KiB
Python
from datetime import datetime
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import astrbot.api.message_components as Comp
|
||
from astrbot.api import logger
|
||
|
||
from .base import DATA_PROCESSING_ERRORS, BaseDomainService
|
||
|
||
|
||
class CharacterService(BaseDomainService):
|
||
async def zhenyan(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_zhenyan", 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 DATA_PROCESSING_ERRORS 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.new_result()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"name": name}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[List[Dict[str, Any]]] = await self.request(
|
||
"jx3_qixue", 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 DATA_PROCESSING_ERRORS as e:
|
||
logger.error(f"处理奇穴数据失败: {e}")
|
||
return_data["msg"] = "处理奇穴数据失败"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
if not await self.attach_template(return_data, "qixue.html"):
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
return return_data
|
||
|
||
async def jineng(self, name: str) -> Dict[str, Any]:
|
||
"""技能"""
|
||
return_data = self.new_result()
|
||
|
||
params = {"name": name}
|
||
|
||
data: Optional[List[Dict[str, Any]]] = await self.request(
|
||
"jx3_jineng", 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 DATA_PROCESSING_ERRORS as e:
|
||
logger.error(f"处理技能数据失败: {e}")
|
||
return_data["msg"] = "处理技能数据失败"
|
||
return return_data
|
||
|
||
if not await self.attach_template(return_data, "jineng.html"):
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
async def jueshe(self, name: str, server: str) -> Dict[str, Any]:
|
||
"""角色"""
|
||
params = {"server": server, "name": name}
|
||
data: Optional[Dict[str, Any]] = await self.request(
|
||
"jx3_jueshexinxi", params=params
|
||
)
|
||
if not isinstance(data, dict) or not data:
|
||
return self.failure("获取接口信息失败")
|
||
|
||
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('globalId', '无')}\n"
|
||
f"职业：{data.get('forceName', '无')}·{data.get('bodyName', '无')}\n"
|
||
f"帮会：{data.get('tongName', '无')}\n"
|
||
f"阵营：{data.get('campName', '无')}\n"
|
||
)
|
||
return self.success(result_msg)
|
||
|
||
async def jueshemingpian(self, server: str, name: str) -> Dict[str, Any]:
|
||
"""单张名片"""
|
||
return_data = self.new_result()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "name": name}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self.request(
|
||
"jx3_jieshemingpian", 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 DATA_PROCESSING_ERRORS 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.new_result()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "name": name}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self.request(
|
||
"jx3_jingnai", 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 DATA_PROCESSING_ERRORS as e:
|
||
logger.error(f"处理精耐数据失败: {e}")
|
||
return_data["msg"] = "处理精耐数据失败"
|
||
return return_data
|
||
|
||
# 4. 加载模板
|
||
if not await self.attach_template(return_data, "jingnai.html"):
|
||
return return_data
|
||
|
||
return_data["code"] = 200
|
||
|
||
return return_data
|
||
|
||
async def shuoyoumingpian(self, server: str, name: str) -> Dict[str, Any]:
|
||
"""所有名片"""
|
||
return_data = self.new_result()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "name": name}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[list[Dict[str, Any]]] = await self.request(
|
||
"jx3_allmingpian", 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 DATA_PROCESSING_ERRORS 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.new_result()
|
||
|
||
# 1. 构造请求参数
|
||
params = {"server": server, "body": body, "force": force}
|
||
|
||
# 2. 调用基础请求
|
||
data: Optional[Dict[str, Any]] = await self.request(
|
||
"jx3_shuijimingpian", 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
|