diff --git a/CHANGELOG.md b/CHANGELOG.md
index a409e97..96fc915 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,9 @@
## 更新记录
+### version: 3.0:
+
+大部分功能重写,还有部分未写完。
+
### version: 2.7:
重构 **交易行 物品名称 [服务器]** 指令,改为免令牌图片查询。
diff --git a/README.md b/README.md
index 73b3ff5..5ca57ba 100644
--- a/README.md
+++ b/README.md
@@ -60,7 +60,29 @@
资历榜单:资历排行 [服务器] [门派]
技能信息:技能 心法
奇穴信息:奇穴 心法
-
+【游戏社区】
+角色聊天;聊天 服务器 角色 [条数] [页数]
+统战歪歪:统战 [服务器]
+家园鲜花:花价 服务器 [名称] [地图]
+家园装饰:装饰 名称
+器物图谱:器物 地图名称
+师徒列表:拜师 服务器 [关键词] | 收徒 服务器 [关键词]
+维护公告:维护 [数量]
+新闻资讯:新闻 [数量]
+团队招募:招募 服务器 [副本] | 团长 服务器 [名称] | 团牌 服务器 [内容]
+骚话语录:答案之书 | 舔狗语录 | 疯狂星期四 | 彩虹屁 | 毒鸡汤 | 朋友圈 | 喝什么 | 吃什么 | 骚话 | 渣男语录
+【百度贴吧】
+外观记录:贴吧物价 名称 [服务器] [数量]
+百度贴吧:818 [服务器] [数量]
+【系统工具】
+科举答案:科举 题目 [条数]
+大区状态:区服 | 开服 服务器
+技改记录:技改
+秘境方位:解密
+掉落统计:掉落 物品 [服务器] [数量]
+【魔盒数据】
+资历查询:资历 角色 服务器
+魔盒宏:交易行 物品名称 服务器
## 安装与依赖
diff --git a/core/jx3api_data.py b/core/jx3api_data.py
index b9d2617..85d44ef 100644
--- a/core/jx3api_data.py
+++ b/core/jx3api_data.py
@@ -16,29 +16,7 @@ from .fun_basic import load_template,gold_to_parts,week_to_num,compare_date_str,
-ACHIEVEMENT_CHOICES = [
- (0, None, "资历总览"),
- (1, "1", "杂闻总览"),
- (2, "2", "武学总览"),
- (3, "3", "修为总览"),
- (4, "4", "装备总览"),
- (5, "5", "技艺总览"),
- (6, "6", "阅读总览"),
- (7, "7", "任务总览"),
- (8, "8", "足迹总览"),
- (9, "9", "战斗总览"),
- (10, "10", "声望总览"),
- (11, "11", "秘境总览"),
- (12, "12", "帮会总览"),
- (13, "13", "阵营总览"),
- (14, "15", "节日总览"),
- (15, "16", "活动总览"),
- (16, "17", "风雨江湖路总览"),
- (17, "40", "家园总览"),
- (18, "41", "剑侠录总览"),
-]
-ACHIEVEMENT_CHOICE_MAP = {index: (menu_id, title) for index, menu_id, title in ACHIEVEMENT_CHOICES}
class JX3APIService:
def __init__(self, config: AstrBotConfig, sqlite: AsyncSQLiteDB, cache_sqlite: Optional[AsyncSQLiteDB] = None):
@@ -1263,407 +1241,45 @@ class JX3APIService:
)
+ async def tongzhanyy(self, server: str) -> Dict[str, Any]:
+ """统战歪歪"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ lines = ["【统战歪歪】"]
+ for group in data:
+ group_server = group.get("server", "")
+ channels = group.get("data", [])
+
+ if len(lines) > 1:
+ lines.append("")
+ lines.append(f"服务器:{group_server}")
+
+ for item in channels:
+ if not isinstance(item, dict):
+ continue
+
+ short_id = item.get("esid") or item.get("asid", "")
+ lines.extend([
+ f"阵营:{item.get('campName', '')}",
+ f"频道ID:{item.get('sid', '')}",
+ f"短位ID:{short_id}",
+ f"在线人数:{item.get('users', '')}",
+ f"频道名:{item.get('snick', '')}",
+ "",
+ ])
+
+ return_data["data"] = "\n".join(lines)
+
+ return await self._request_api(
+ path="/duowan/statistics",
+ params= {"server": server},
+ processor=processor,
+ template=""
+ )
-
-
-
-
-
-
-
- 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:
+ async def xiaoyao(self, name:str) -> Dict[str, Any]:
+ """小吃小药"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
result = {}
for item in data:
@@ -1685,256 +1301,18 @@ class JX3APIService:
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"] = "处理接口返回信息时出错"
+ return await self._request_api(
+ path="/food/list",
+ params= {"name": name},
+ processor=processor,
+ template="xiaoyao.html"
+ )
- # 加载模板
- 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 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 jiemi(self) -> Dict[str, Any]:
- """解密"""
- return_data = self._init_return_data()
-
- params = {"token": self.token}
-
- data: Optional[Dict[str, Any]] = await self._base_request(
- "jx3_jiemi", "GET", params=params
- )
-
- if not data or not isinstance(data, dict):
- return_data["msg"] = "未查询到解密信息"
- return return_data
-
- try:
- curr = data.get("curr", {})
- next_data = data.get("next", {})
- if not isinstance(curr, dict):
- curr = {}
- if not isinstance(next_data, dict):
- next_data = {}
-
- return_data["data"] = "\n".join([
- "解密",
- f"当前时间:{data.get('time', '')}",
- f"当前节点:{curr.get('node', '')}",
- f"当前结果:{curr.get('data', '')}",
- f"下轮节点:{next_data.get('node', '')}",
- f"下轮结果:{next_data.get('data', '')}",
- f"剩余时间:{data.get('cdtn', '')}",
- ])
- except Exception 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 tongzhanyy(self, server: str) -> Dict[str, Any]:
- """统战歪歪"""
- return_data = self._init_return_data()
-
- params = {
- "server": server,
- "token": self.token,
- }
-
- data: Optional[List[Dict[str, Any]]] = await self._base_request(
- "jx3_tongzhanyy", "GET", params=params
- )
-
- if not data or not isinstance(data, list):
- return_data["msg"] = "未查询到统战歪歪信息"
- return return_data
-
- try:
- lines = ["统战歪歪"]
- has_channel = False
-
- for group in data:
- if not isinstance(group, dict):
- continue
-
- group_server = group.get("server", "")
- channels = group.get("data", [])
- if not isinstance(channels, list) or not channels:
- continue
-
- if len(lines) > 1:
- lines.append("")
- lines.append(f"服务器:{group_server}")
-
- for item in channels:
- if not isinstance(item, dict):
- continue
-
- has_channel = True
- short_id = item.get("esid") or item.get("asid", "")
- lines.extend([
- f"阵营:{item.get('campName', '')}",
- f"频道ID:{item.get('sid', '')}",
- f"短位ID:{short_id}",
- f"在线人数:{item.get('users', '')}",
- f"频道名:{item.get('snick', '')}",
- "",
- ])
-
- if not has_channel:
- return_data["msg"] = "未查询到统战歪歪信息"
- return return_data
-
- return_data["data"] = "\n".join(lines).rstrip()
- except Exception 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]:
+ async def pianzhi(self, server:str, 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:
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
records = data["records"]
if not records:
@@ -1956,583 +1334,383 @@ class JX3APIService:
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
+
+ return await self._request_api(
+ path="/fraud/detail",
+ params= {"server": server, "uid": uid, "token": self.token},
+ processor=processor,
+ template=""
+ )
+ async def huajia(self,server: str, name: str, map: str) -> Dict[str, Any]:
+ """家园鲜花"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ return_data["data"]["data"] = data
+ return_data["data"]["server"] = server
+
+ return await self._request_api(
+ path="/home/flower",
+ params= {"server": server, "name": name, "map": map},
+ processor=processor,
+ template="huajia.html"
+ )
+ async def zhuangshi(self,name: str) -> Dict[str, Any]:
+ """家园装饰"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ return_data["data"]["data"] = data
+
+ return await self._request_api(
+ path="/home/furniture",
+ params= { "name": name},
+ processor=processor,
+ template="zhuangshi.html"
+ )
+ async def qiwu(self,name: str) -> Dict[str, Any]:
+ """器物图谱"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ return_data["data"]["data"] = data
+ return_data["data"]["name"] = name
+ return await self._request_api(
+ path="/home/travel",
+ params= { "name": name},
+ processor=processor,
+ template="qiwu.html"
+ )
+
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- 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]:
+ async def shitu(self, label: int, server: str, keyword: str, limit:int) -> 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", [])
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ items = data
if not items:
return_data["msg"] = "未查询到师徒招募信息"
return return_data
- title = "收徒信息" if type_value == 1 else "拜师信息"
+ title = "收徒信息" if label == 1 else "拜师信息"
return_data["data"] = {
"items": items,
- "server": data.get("server", server),
+ "server": server,
"keyword": keyword,
- "type_value": type_value,
+ "type_value": label,
"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
+ return await self._request_api(
+ path="/mentor/search",
+ params= {"label": label, "server": server, "keyword": keyword, "limit": limit, "token": self.token},
+ processor=processor,
+ template="shitu.html"
+ )
+ async def weihu(self, limit:int) -> Dict[str, Any]:
+ """维护公告"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ result = data[0]
+ return_data["status"] = result.get('id')
+
+ result_msg = "维护推送\n"
+ # 仅展示前1条,避免消息过长
+ for i, item in enumerate(data[:limit], 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
+
+ return await self._request_api(
+ path="/news/announce",
+ params= {"limit": limit},
+ processor=processor,
+ template=""
+ )
+ async def xinwen(self, limit:int) -> Dict[str, Any]:
+ """新闻资讯"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ result = data[0]
+ return_data["status"] = result.get('catid')
- async def _load_achievement_cache(self, key: str) -> tuple[Optional[Any], bool]:
- """读取资历基础数据缓存,返回数据和是否已过期"""
- try:
- row = await self._cache_db.select_one("achievement_cache", "key=?", (key,))
- except Exception as e:
- logger.error(f"读取资历缓存失败: {e}")
- return None, True
+ result_msg = "新闻资讯推送\n"
+ # 仅展示前1条,避免消息过长
+ for i, item in enumerate(data[:limit], 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
- if not row:
- return None, True
-
- try:
- payload = json.loads(row.get("content", "{}"))
- updated_at = datetime.strptime(row.get("updated_at", ""), "%Y-%m-%d %H:%M:%S")
- expired = datetime.now() - updated_at > timedelta(days=30)
- return payload, expired
- except Exception as e:
- logger.error(f"解析资历缓存失败: {e}")
- return None, True
+ return await self._request_api(
+ path="/news/records",
+ params= {"limit": limit},
+ processor=processor,
+ template=""
+ )
- async def _save_achievement_cache(self, key: str, payload: Any):
- """写入资历基础数据缓存"""
- try:
- await self._cache_db.execute(
- """
- INSERT INTO achievement_cache (key, content, updated_at)
- VALUES (?, ?, ?)
- ON CONFLICT(key) DO UPDATE SET
- content=excluded.content,
- updated_at=excluded.updated_at
- """,
- (
- key,
- json.dumps(payload, ensure_ascii=False),
- datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
- ),
+ async def tuanduizhaomu(self, server: str, label:int, keyword: str, limit:int) -> Dict[str, Any]:
+ """团队招募"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ for item in data:
+ item["createTime"] = format_time(item["createTime"])
+ item["maxMemberCount"] = f"{item['currentMemberCount']}/{item['maxMemberCount']}"
+ return_data["data"]["list"] = data
+
+ return await self._request_api(
+ path="/recruit/search",
+ params= {"server": server, "label": label, "keyword": keyword, "limit": limit, "token": self.token},
+ processor=processor,
+ template="tuanduizhaomu.html"
+ )
+
+
+ async def daanzhishu(self) -> Dict[str, Any]:
+ """答案之书"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ return_data["data"] = (
+ f"答案:{data.get('answer', '')}\n"
+ f"鼓励:{data.get('hearten', '')}\n"
)
- except Exception as e:
- logger.error(f"写入资历缓存失败: {e}")
+
+ return await self._request_api(
+ path="/saohua/answer",
+ params= {},
+ processor=processor,
+ template=""
+ )
- async def _get_achievement_base_data(self, cache_key: str, config_key: str) -> Optional[Dict[str, Any]]:
- """获取资历菜单或点数数据,优先使用未过期缓存"""
- cached, expired = await self._load_achievement_cache(cache_key)
- if cached and not expired:
- return cached
+ async def tiangou(self) -> Dict[str, Any]:
+ """舔狗日志"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ return_data["data"] = data.get('text')
- data: Optional[Dict[str, Any]] = await self._base_request(config_key, "GET")
- if data and isinstance(data, dict):
- await self._save_achievement_cache(cache_key, data)
- return data
-
- if cached:
- logger.warning(f"资历基础数据接口失败,使用旧缓存: {cache_key}")
- return cached
-
- return None
+ return await self._request_api(
+ path="/saohua/content",
+ params= {},
+ processor=processor,
+ template=""
+ )
- async def _get_trade_item_groups(self) -> Optional[List[Dict[str, Any]]]:
- """获取交易行物品库,优先使用未过期缓存"""
- cache_key = "trade_item_groups"
- cached, expired = await self._load_achievement_cache(cache_key)
- if isinstance(cached, list) and not expired:
- return cached
+ async def fengleiyulu(self, name:str) -> Dict[str, Any]:
+ """分类语录"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ return_data["data"] = data.get('content')
- data = await self._base_request("jx3box_trade_items", "GET")
- if isinstance(data, list) and data:
- await self._save_achievement_cache(cache_key, data)
- return data
-
- if isinstance(cached, list) and cached:
- logger.warning("交易行物品库接口失败,使用旧缓存")
- return cached
-
- return None
+ return await self._request_api(
+ path="/saohua/context",
+ params= { "name": name, "token": self.token},
+ processor=processor,
+ template=""
+ )
- def _flatten_trade_items(self, groups: List[Dict[str, Any]]) -> list[Dict[str, Any]]:
- """从交易行物品分组中提取可查询物品"""
- items = []
- for group in groups:
- if not isinstance(group, dict):
- continue
- for item in group.get("items", []) or []:
- if not isinstance(item, dict):
- continue
- item_id = item.get("item_id")
- label = item.get("label")
- if not item_id or not label:
- continue
- items.append(
- {
- "item_id": str(item_id),
- "label": str(label),
- "icon": str(item.get("icon") or ""),
- }
- )
- return items
+ async def heshengme(self) -> Dict[str, Any]:
+ """喝什么"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ return_data["data"] = (
+ f"{', '.join(data or [])}\n"
+ )
+
+ return await self._request_api(
+ path="/saohua/drink",
+ params= {},
+ processor=processor,
+ template=""
+ )
- def _match_trade_items(self, items: list[Dict[str, Any]], keyword: str, limit: int = 50) -> list[Dict[str, Any]]:
- """按物品名模糊匹配交易行物品"""
- keyword = (keyword or "").strip()
- if not keyword:
- return []
+ async def chishengme(self) -> Dict[str, Any]:
+ """吃什么"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ return_data["data"] = (
+ f"{', '.join(data or [])}\n"
+ )
- matched = []
- seen = set()
- for item in items:
- label = item.get("label", "")
- item_id = item.get("item_id", "")
- if keyword not in label or item_id in seen:
- continue
- seen.add(item_id)
- if label == keyword:
- rank = 0
- elif label.startswith(keyword):
- rank = 1
+ return await self._request_api(
+ path="/saohua/eat",
+ params= {},
+ processor=processor,
+ template=""
+ )
+
+
+ async def shaohua(self) -> Dict[str, Any]:
+ """随机骚话"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ return_data["data"] = data.get('text')
+
+ return await self._request_api(
+ path="/saohua/random",
+ params= {},
+ processor=processor,
+ template=""
+ )
+
+
+ async def zhananyulu(self) -> Dict[str, Any]:
+ """渣男语录"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ return_data["data"] = data.get('text')
+
+ return await self._request_api(
+ path="/saohua/zhanan",
+ params= {},
+ processor=processor,
+ template=""
+ )
+
+
+ async def keju(self,subject: str, limit: int) -> Dict[str, Any]:
+ """科举"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ 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
+
+ return await self._request_api(
+ path="/exam/search",
+ params= {"subject": subject, "limit": limit},
+ processor=processor,
+ template=""
+ )
+
+
+ async def zhuangtai(self,server:str) -> Dict[str, Any]:
+ """区服状态"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ 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
+
+ return await self._request_api(
+ path="/server/status/check",
+ params= {"server": server},
+ processor=processor,
+ template="qufuzhuangtai.html"
+ )
+
+
+ async def kaifu(self, server: str) -> Dict[str, Any]:
+ """开服状态查询"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ if not data.get("status") == "维护" :
+ status = 1
else:
- rank = 2
- matched.append((rank, len(label), label, item))
+ status = 0
+
+ if status == 1:
+ status_str = f"{server}服务器已开服,快冲,快冲!"
+ status_bool = True
+ else:
+ status_str = f"{server}服务器当前维护中,等会再来吧!"
+ status_bool = False
- matched.sort(key=lambda row: (row[0], row[1], row[2]))
- return [row[3] for row in matched[:limit]]
+ return_data["status"] = status_bool
+ return_data["data"] = status_str
+
+ return await self._request_api(
+ path="/server/status/check",
+ params= {"server": server},
+ processor=processor,
+ template="qufuzhuangtai.html"
+ )
- def _flatten_achievement_ids(self, values: Any) -> list[int]:
- """展开菜单中的单个资历 ID 和数组资历 ID"""
- result = []
+ async def jigai(self) -> Dict[str, Any]:
+ """技改记录"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ 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
- if isinstance(values, list):
- for item in values:
- result.extend(self._flatten_achievement_ids(item))
- return result
-
- try:
- result.append(int(values))
- except (TypeError, ValueError):
- pass
-
- return result
+ return await self._request_api(
+ path="/skill/rework",
+ params= {},
+ processor=processor,
+ template=""
+ )
- def _category_achievement_ids(self, category: Dict[str, Any]) -> list[int]:
- """提取分类及其子分类包含的资历 ID"""
- ids = self._flatten_achievement_ids(category.get("achievements", []))
- for child in category.get("children", []) or []:
- if isinstance(child, dict):
- ids.extend(self._flatten_achievement_ids(child.get("achievements", [])))
- return ids
+ async def jiemi(self) -> Dict[str, Any]:
+ """解密"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ return_data["data"] = "\n".join([
+ "解密",
+ f"当前时间:{data.get('nowTime', '')}",
+ f"当前节点:{data.get('nowNode', '')}",
+ f"当前结果:{data.get('nowResult', '')}",
+ f"下轮节点:{data.get('nextNode', '')}",
+ f"下轮结果:{data.get('nextResult', '')}",
+ f"剩余时间:{data.get('intervalTime', '')}",
+ ])
+
+ return await self._request_api(
+ path="/mech/decrypt",
+ params= {"token": self.token},
+ processor=processor,
+ template=""
+ )
- def _build_achievement_progress(
- self,
- name: str,
- achievement_ids: list[int],
- point_map: Dict[str, Any],
- completed_ids: set[int],
- ) -> Dict[str, Any]:
- """按资历点数计算完成进度"""
- unique_ids = set(achievement_ids)
- total_points = 0
- completed_points = 0
- completed_count = 0
-
- for achievement_id in unique_ids:
- try:
- point = int(point_map.get(str(achievement_id), 0) or 0)
- except (TypeError, ValueError):
- point = 0
-
- total_points += point
- if achievement_id in completed_ids:
- completed_count += 1
- completed_points += point
-
- percent = round(completed_points / total_points * 100, 2) if total_points else 0
-
- return {
- "name": name,
- "total_points": total_points,
- "completed_points": completed_points,
- "percent": percent,
- "percent_text": f"{percent:.2f}%",
- "achievement_count": len(unique_ids),
- "completed_count": completed_count,
- }
-
-
- async def zili(self, name: str, server: str, choice: int) -> Dict[str, Any]:
- """角色资历"""
- return_data = self._init_return_data()
-
- if choice not in ACHIEVEMENT_CHOICE_MAP:
- return_data["msg"] = "无效序号,结束会话"
- return return_data
-
- role_params = {"server": server, "name": name, "token": self.token}
- role_data: Optional[Dict[str, Any]] = await self._base_request(
- "jx3_jueshexinxi", "GET", params=role_params
- )
- if not role_data or not isinstance(role_data, dict):
- return_data["msg"] = "未查询到角色"
- return return_data
-
- global_id = role_data.get("globalId")
- if not global_id:
- return_data["msg"] = "无法获取角色全区 ID"
- return return_data
-
- achievement_data = await self._api.get(
- "https://next2.jx3box.com/api/next2/user-achievements",
- params={"jx3id": global_id},
- out_key="data",
- )
- if not achievement_data or not isinstance(achievement_data, dict):
- return_data["msg"] = "未查询到资历数据"
- return return_data
-
- completed_ids = set()
- achievements = achievement_data.get("achievements") or ""
- if isinstance(achievements, str):
- for item in achievements.split(","):
- try:
- completed_ids.add(int(item.strip()))
- except ValueError:
- continue
- elif isinstance(achievements, list):
- completed_ids = set(self._flatten_achievement_ids(achievements))
-
- menu_payload = await self._get_achievement_base_data(
- "achievement_menus",
- "jx3box_achievement_menus",
- )
- point_payload = await self._get_achievement_base_data(
- "achievement_points",
- "jx3box_achievement_points",
- )
- if not menu_payload or not point_payload:
- return_data["msg"] = "基础资历数据获取失败"
- return return_data
-
- menus = menu_payload.get("menus", {})
- points = point_payload.get("points", {})
- if not isinstance(menus, dict) or not isinstance(points, dict):
- return_data["msg"] = "基础资历数据格式异常"
- return return_data
-
- selected_menu_id, selected_title = ACHIEVEMENT_CHOICE_MAP[choice]
- items = []
-
- if choice == 0:
- total_ids = []
- for _, menu_id, title in ACHIEVEMENT_CHOICES[1:]:
- category = menus.get(str(menu_id), {})
- if not isinstance(category, dict):
- continue
- category_ids = self._category_achievement_ids(category)
- total_ids.extend(category_ids)
- items.append(
- self._build_achievement_progress(title, category_ids, points, completed_ids)
- )
- summary = self._build_achievement_progress(selected_title, total_ids, points, completed_ids)
- else:
- category = menus.get(str(selected_menu_id), {})
- if not isinstance(category, dict) or not category:
- return_data["msg"] = "未找到该资历分类"
- return return_data
-
- category_ids = self._category_achievement_ids(category)
- summary = self._build_achievement_progress(selected_title, category_ids, points, completed_ids)
- for child in category.get("children", []) or []:
- if not isinstance(child, dict):
- continue
- child_ids = self._flatten_achievement_ids(child.get("achievements", []))
- items.append(
- self._build_achievement_progress(child.get("name", "未命名"), child_ids, points, completed_ids)
- )
-
- try:
- return_data["temp"] = await load_template("zili.html")
- except FileNotFoundError as e:
- logger.error(f"加载模板失败: {e}")
- return_data["msg"] = "系统错误:模板文件不存在"
- return return_data
-
- return_data["data"] = {
- "title": selected_title,
- "summary": summary,
- "items": items,
- "role_name": role_data.get("roleName", name),
- "server": role_data.get("serverName", server),
- "zone": role_data.get("zoneName", ""),
- "force_name": role_data.get("forceName", "无"),
- "camp_name": role_data.get("campName", "无"),
- "tong_name": role_data.get("tongName", "无"),
- "updated_at": achievement_data.get("updated_at", ""),
- "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
- }
- return_data["code"] = 200
-
- return return_data
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- async def jiaoyihang(self, name: str , server: str) -> Dict[str, Any]:
- """区服交易行"""
- return_data = self._init_return_data()
-
- item_groups = await self._get_trade_item_groups()
- if not item_groups:
- return_data["msg"] = "交易行基础物品数据获取失败"
- return return_data
-
- trade_items = self._flatten_trade_items(item_groups)
- matched_items = self._match_trade_items(trade_items, name, 50)
- if not matched_items:
- return_data["msg"] = "未找到匹配的交易行物品"
- return return_data
-
- item_map = {item["item_id"]: item for item in matched_items}
- params = {
- "item_ids": list(item_map.keys()),
- "server": server,
- "aggregate_type": "hourly",
- }
- price_data: Optional[List[Dict[str, Any]]] = await self._base_request(
- "jx3_jiaoyihang", "POST", params=params, out_key=""
- )
-
- if not price_data or not isinstance(price_data, list):
- return_data["msg"] = "未查询到交易行价格数据"
- return return_data
-
- try:
- result = []
- for price_item in price_data:
- if not isinstance(price_item, dict):
- continue
-
- item_id = str(price_item.get("item_id") or "")
- base_item = item_map.get(item_id)
- if not base_item:
- continue
-
- timestamp = price_item.get("timestamp")
- try:
- created = datetime.fromtimestamp(int(timestamp)).strftime("%Y-%m-%d %H:%M:%S")
- except (TypeError, ValueError, OSError):
- created = ""
-
- result.append(
- {
- "item_id": item_id,
- "name": base_item.get("label", ""),
- "icon": f"https://icon.jx3box.com/icon/{base_item.get('icon', '')}.png",
- "server": price_item.get("server", server),
- "price": price_item.get("price", 0),
- "price_parts": gold_to_parts(price_item.get("price", 0)),
- "sample": price_item.get("sample", 0),
- "created": created,
- }
- )
-
- if not result:
- return_data["msg"] = "未查询到交易行价格数据"
- return return_data
+ async def diaoluo(self, name: str, server: str, limit: int ) -> Dict[str, Any]:
+ """物品掉落记录"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ for item in data:
+ item["time"] = format_time(item.get("time"))
return_data["data"] = {
- "search_name": name,
+ "items": data,
+ "name": name,
+ "limit": limit,
"server": server,
- "matched_count": len(matched_items),
- "result_count": len(result),
- "list": result,
- "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+ "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
- # 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
+ return await self._request_api(
+ path="/reward/statistics",
+ params= {"server": server, "name": name, "limit": limit, "token": self.token},
+ processor=processor,
+ template="diaoluo.html"
+ )
- async def tiebawujia(self, name: str, limit: int = 5, server: str = "") -> Dict[str, Any]:
+
+
+ async def tiebawujia(self, name: str, server: str, limit: int ) -> 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 > 50:
- return_data["msg"] = "贴吧物价记录数量必须在 1-50 之间"
- return return_data
-
- params = {
- "server": server,
- "name": name,
- "limit": limit,
- "token": self.token,
- }
-
- data: Optional[List[Dict[str, Any]]] = await self._base_request(
- "jx3_tiebawujia", "GET", params=params
- )
-
- if not data or not isinstance(data, list):
- return_data["msg"] = "未查询到贴吧物价记录"
- return return_data
-
- try:
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
lines = [
f"贴吧物价:{name}",
f"服务器:{server}",
@@ -2566,83 +1744,38 @@ class JX3APIService:
return return_data
return_data["data"] = "\n".join(lines).rstrip()
- except Exception as e:
- logger.error(f"处理贴吧物价数据失败: {e}")
- return_data["msg"] = "处理贴吧物价数据失败"
- return return_data
- return_data["code"] = 200
-
- return return_data
+ return await self._request_api(
+ path="/tieba/item/records",
+ params= {"server": server,"name": name,"limit": limit,"token": self.token,},
+ processor=processor,
+ template=""
+ )
- 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]:
+ async def bagua(self, tags: str, server:str, limit:str ) -> Dict[str, Any]:
"""八卦"""
+ async def processor(data: Any, return_data: Dict[str, Any]) -> None:
+ if not data:
+ result_msg = f"未找到相关 {tags} 记录。\n"
+ result_msg += f"可选范围:818 616 鬼网三 鬼网3 树洞 记录 教程 街拍 故事 避雷 吐槽 提问"
+ else:
+ result_msg = f"类型:【{tags}】\n\n"
+
+ for item in data:
+ result_msg += f"{item['title']}\n"
+ result_msg += f"服务器:{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
+
+ return await self._request_api(
+ path="/tieba/random",
+ params= {"server": server,"tags": tags,"limit": limit,"token": self.token,},
+ processor=processor,
+ template=""
+ )
return_data = self._init_return_data()
# 1. 构造请求参数
@@ -2681,97 +1814,7 @@ class JX3APIService:
- 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
diff --git a/core/jx3box_data.py b/core/jx3box_data.py
index 65fc52b..4211632 100644
--- a/core/jx3box_data.py
+++ b/core/jx3box_data.py
@@ -14,6 +14,29 @@ from .request import APIClient
from .sqlite import AsyncSQLiteDB
from .fun_basic import load_template,gold_to_parts,week_to_num,compare_date_str,format_time,format_remaining
+ACHIEVEMENT_CHOICES = [
+ (0, None, "资历总览"),
+ (1, "1", "杂闻总览"),
+ (2, "2", "武学总览"),
+ (3, "3", "修为总览"),
+ (4, "4", "装备总览"),
+ (5, "5", "技艺总览"),
+ (6, "6", "阅读总览"),
+ (7, "7", "任务总览"),
+ (8, "8", "足迹总览"),
+ (9, "9", "战斗总览"),
+ (10, "10", "声望总览"),
+ (11, "11", "秘境总览"),
+ (12, "12", "帮会总览"),
+ (13, "13", "阵营总览"),
+ (14, "15", "节日总览"),
+ (15, "16", "活动总览"),
+ (16, "17", "风雨江湖路总览"),
+ (17, "40", "家园总览"),
+ (18, "41", "剑侠录总览"),
+]
+
+ACHIEVEMENT_CHOICE_MAP = {index: (menu_id, title) for index, menu_id, title in ACHIEVEMENT_CHOICES}
class JX3BOXService:
def __init__(self, config: AstrBotConfig, sqlite: AsyncSQLiteDB, cache_sqlite: Optional[AsyncSQLiteDB] = None):
@@ -25,7 +48,7 @@ class JX3BOXService:
self._sql_db = sqlite
self._cache_db = cache_sqlite or sqlite
-
+ self.token = self._config.get("jx3api_token", "")
async def close(self):
"""释放底层 APIClient 资源"""
@@ -143,4 +166,502 @@ class JX3BOXService:
return_data["code"] = 200
- return return_data
\ No newline at end of file
+ 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. 调用基础请求
+ api_url = "https://cms.jx3box.com/api/cms/posts"
+ data: Optional[Dict[str, Any]] = await self._api.get(api_url, params=params, out_key="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 isinstance(data, dict):
+ return_data["msg"] = "获取宏数据异常"
+ return return_data
+
+ try:
+ # 文章正文
+ return_data["temp"] = data.get("post_content", "")
+
+ # 安全获取宏数据,避免 KeyError
+ macro_list = data.get("post_meta", {}).get("data", [])
+
+ msg = ""
+
+ for m in macro_list:
+ msg += f"【宏名称】\n{m.get('name', '')}\n"
+ msg += f"【使用说明】\n{m.get('desc', '')}\n"
+ msg += f"【宏脚本】\n{m.get('macro', '')}\n\n"
+
+ # 没有宏数据时返回文章信息
+ if not msg:
+ msg = (
+ f"标题:{data.get('post_title', '')}\n"
+ f"作者:{data.get('author', '')}\n"
+ "该帖子没有宏数据"
+ )
+
+ return_data["data"] = msg
+ return_data["code"] = 200
+
+ except Exception:
+ logger.exception("处理返回数据失败")
+ return_data["msg"] = "处理返回数据失败"
+
+ return return_data
+
+
+ async def _load_achievement_cache(self, key: str) -> tuple[Optional[Any], bool]:
+ """读取资历基础数据缓存,返回数据和是否已过期"""
+ try:
+ row = await self._cache_db.select_one("achievement_cache", "key=?", (key,))
+ except Exception as e:
+ logger.error(f"读取资历缓存失败: {e}")
+ return None, True
+
+ if not row:
+ return None, True
+
+ try:
+ payload = json.loads(row.get("content", "{}"))
+ updated_at = datetime.strptime(row.get("updated_at", ""), "%Y-%m-%d %H:%M:%S")
+ expired = datetime.now() - updated_at > timedelta(days=30)
+ return payload, expired
+ except Exception as e:
+ logger.error(f"解析资历缓存失败: {e}")
+ return None, True
+
+
+ async def _save_achievement_cache(self, key: str, payload: Any):
+ """写入资历基础数据缓存"""
+ try:
+ await self._cache_db.execute(
+ """
+ INSERT INTO achievement_cache (key, content, updated_at)
+ VALUES (?, ?, ?)
+ ON CONFLICT(key) DO UPDATE SET
+ content=excluded.content,
+ updated_at=excluded.updated_at
+ """,
+ (
+ key,
+ json.dumps(payload, ensure_ascii=False),
+ datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+ ),
+ )
+ except Exception as e:
+ logger.error(f"写入资历缓存失败: {e}")
+
+
+ async def _get_achievement_base_data(self, cache_key: str, config_key: str) -> Optional[Dict[str, Any]]:
+ """获取资历菜单或点数数据,优先使用未过期缓存"""
+ cached, expired = await self._load_achievement_cache(cache_key)
+ if cached and not expired:
+ return cached
+
+ data: Optional[Dict[str, Any]] = await self._base_request(config_key, "GET")
+ if data and isinstance(data, dict):
+ await self._save_achievement_cache(cache_key, data)
+ return data
+
+ if cached:
+ logger.warning(f"资历基础数据接口失败,使用旧缓存: {cache_key}")
+ return cached
+
+ return None
+
+
+ async def _get_trade_item_groups(self) -> Optional[List[Dict[str, Any]]]:
+ """获取交易行物品库,优先使用未过期缓存"""
+ cache_key = "trade_item_groups"
+ cached, expired = await self._load_achievement_cache(cache_key)
+ if isinstance(cached, list) and not expired:
+ return cached
+
+
+ api_url = "https://cms.jx3box.com/api/cms/pvx/item/group"
+ data = await self._api.get(api_url, params={"client":"std"}, out_key="data")
+ if isinstance(data, list) and data:
+ await self._save_achievement_cache(cache_key, data)
+ return data
+
+ if isinstance(cached, list) and cached:
+ logger.warning("交易行物品库接口失败,使用旧缓存")
+ return cached
+
+ return None
+
+
+ def _flatten_trade_items(self, groups: List[Dict[str, Any]]) -> list[Dict[str, Any]]:
+ """从交易行物品分组中提取可查询物品"""
+ items = []
+ for group in groups:
+ if not isinstance(group, dict):
+ continue
+ for item in group.get("items", []) or []:
+ if not isinstance(item, dict):
+ continue
+ item_id = item.get("item_id")
+ label = item.get("label")
+ if not item_id or not label:
+ continue
+ items.append(
+ {
+ "item_id": str(item_id),
+ "label": str(label),
+ "icon": str(item.get("icon") or ""),
+ }
+ )
+ return items
+
+
+ def _match_trade_items(self, items: list[Dict[str, Any]], keyword: str, limit: int = 50) -> list[Dict[str, Any]]:
+ """按物品名模糊匹配交易行物品"""
+ keyword = (keyword or "").strip()
+ if not keyword:
+ return []
+
+ matched = []
+ seen = set()
+ for item in items:
+ label = item.get("label", "")
+ item_id = item.get("item_id", "")
+ if keyword not in label or item_id in seen:
+ continue
+ seen.add(item_id)
+ if label == keyword:
+ rank = 0
+ elif label.startswith(keyword):
+ rank = 1
+ else:
+ rank = 2
+ matched.append((rank, len(label), label, item))
+
+ matched.sort(key=lambda row: (row[0], row[1], row[2]))
+ return [row[3] for row in matched[:limit]]
+
+
+ def _flatten_achievement_ids(self, values: Any) -> list[int]:
+ """展开菜单中的单个资历 ID 和数组资历 ID"""
+ result = []
+
+ if isinstance(values, list):
+ for item in values:
+ result.extend(self._flatten_achievement_ids(item))
+ return result
+
+ try:
+ result.append(int(values))
+ except (TypeError, ValueError):
+ pass
+
+ return result
+
+
+ def _category_achievement_ids(self, category: Dict[str, Any]) -> list[int]:
+ """提取分类及其子分类包含的资历 ID"""
+ ids = self._flatten_achievement_ids(category.get("achievements", []))
+ for child in category.get("children", []) or []:
+ if isinstance(child, dict):
+ ids.extend(self._flatten_achievement_ids(child.get("achievements", [])))
+ return ids
+
+
+ def _build_achievement_progress(
+ self,
+ name: str,
+ achievement_ids: list[int],
+ point_map: Dict[str, Any],
+ completed_ids: set[int],
+ ) -> Dict[str, Any]:
+ """按资历点数计算完成进度"""
+ unique_ids = set(achievement_ids)
+ total_points = 0
+ completed_points = 0
+ completed_count = 0
+
+ for achievement_id in unique_ids:
+ try:
+ point = int(point_map.get(str(achievement_id), 0) or 0)
+ except (TypeError, ValueError):
+ point = 0
+
+ total_points += point
+ if achievement_id in completed_ids:
+ completed_count += 1
+ completed_points += point
+
+ percent = round(completed_points / total_points * 100, 2) if total_points else 0
+
+ return {
+ "name": name,
+ "total_points": total_points,
+ "completed_points": completed_points,
+ "percent": percent,
+ "percent_text": f"{percent:.2f}%",
+ "achievement_count": len(unique_ids),
+ "completed_count": completed_count,
+ }
+
+
+ async def zili(self, name: str, server: str, choice: int) -> Dict[str, Any]:
+ """角色资历"""
+ return_data = self._init_return_data()
+
+ if choice not in ACHIEVEMENT_CHOICE_MAP:
+ return_data["msg"] = "无效序号,结束会话"
+ return return_data
+
+ role_params = {"server": server, "name": name, "token": self.token}
+ api_url = "https://www.jx3api.com/role/detail"
+ role_data: Optional[Dict[str, Any]] = await self._api.get(api_url, params=role_params, out_key="data")
+
+ if not role_data or not isinstance(role_data, dict):
+ return_data["msg"] = "未查询到角色"
+ return return_data
+
+ global_id = role_data.get("globalId")
+ if not global_id:
+ return_data["msg"] = "无法获取角色全区 ID"
+ return return_data
+
+ achievement_data = await self._api.get(
+ "https://next2.jx3box.com/api/next2/user-achievements",
+ params={"jx3id": global_id},
+ out_key="data",
+ )
+ if not achievement_data or not isinstance(achievement_data, dict):
+ return_data["msg"] = "未查询到资历数据"
+ return return_data
+
+ completed_ids = set()
+ achievements = achievement_data.get("achievements") or ""
+ if isinstance(achievements, str):
+ for item in achievements.split(","):
+ try:
+ completed_ids.add(int(item.strip()))
+ except ValueError:
+ continue
+ elif isinstance(achievements, list):
+ completed_ids = set(self._flatten_achievement_ids(achievements))
+
+ menu_payload = await self._get_achievement_base_data(
+ "achievement_menus",
+ "jx3box_achievement_menus",
+ )
+ point_payload = await self._get_achievement_base_data(
+ "achievement_points",
+ "jx3box_achievement_points",
+ )
+ if not menu_payload or not point_payload:
+ return_data["msg"] = "基础资历数据获取失败"
+ return return_data
+
+ menus = menu_payload.get("menus", {})
+ points = point_payload.get("points", {})
+ if not isinstance(menus, dict) or not isinstance(points, dict):
+ return_data["msg"] = "基础资历数据格式异常"
+ return return_data
+
+ selected_menu_id, selected_title = ACHIEVEMENT_CHOICE_MAP[choice]
+ items = []
+
+ if choice == 0:
+ total_ids = []
+ for _, menu_id, title in ACHIEVEMENT_CHOICES[1:]:
+ category = menus.get(str(menu_id), {})
+ if not isinstance(category, dict):
+ continue
+ category_ids = self._category_achievement_ids(category)
+ total_ids.extend(category_ids)
+ items.append(
+ self._build_achievement_progress(title, category_ids, points, completed_ids)
+ )
+ summary = self._build_achievement_progress(selected_title, total_ids, points, completed_ids)
+ else:
+ category = menus.get(str(selected_menu_id), {})
+ if not isinstance(category, dict) or not category:
+ return_data["msg"] = "未找到该资历分类"
+ return return_data
+
+ category_ids = self._category_achievement_ids(category)
+ summary = self._build_achievement_progress(selected_title, category_ids, points, completed_ids)
+ for child in category.get("children", []) or []:
+ if not isinstance(child, dict):
+ continue
+ child_ids = self._flatten_achievement_ids(child.get("achievements", []))
+ items.append(
+ self._build_achievement_progress(child.get("name", "未命名"), child_ids, points, completed_ids)
+ )
+
+ try:
+ return_data["temp"] = await load_template("zili.html")
+ except FileNotFoundError as e:
+ logger.error(f"加载模板失败: {e}")
+ return_data["msg"] = "系统错误:模板文件不存在"
+ return return_data
+
+ return_data["data"] = {
+ "title": selected_title,
+ "summary": summary,
+ "items": items,
+ "role_name": role_data.get("roleName", name),
+ "server": role_data.get("serverName", server),
+ "zone": role_data.get("zoneName", ""),
+ "force_name": role_data.get("forceName", "无"),
+ "camp_name": role_data.get("campName", "无"),
+ "tong_name": role_data.get("tongName", "无"),
+ "updated_at": achievement_data.get("updated_at", ""),
+ "update_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
+ }
+ return_data["code"] = 200
+
+ return return_data
+
+
+ async def jiaoyihang(self, name: str , server: str) -> Dict[str, Any]:
+ """区服交易行"""
+ return_data = self._init_return_data()
+
+ item_groups = await self._get_trade_item_groups()
+ if not item_groups:
+ return_data["msg"] = "交易行基础物品数据获取失败"
+ return return_data
+
+ trade_items = self._flatten_trade_items(item_groups)
+ matched_items = self._match_trade_items(trade_items, name, 50)
+ if not matched_items:
+ return_data["msg"] = "未找到匹配的交易行物品"
+ return return_data
+
+ item_map = {item["item_id"]: item for item in matched_items}
+ params = {
+ "item_ids": list(item_map.keys()),
+ "server": server,
+ "aggregate_type": "hourly",
+ }
+ api_url = "https://next2.jx3box.com/api/auction/"
+ price_data: Optional[List[Dict[str, Any]]] = await self._api.post(api_url, data=params, out_key="")
+
+
+ if not price_data or not isinstance(price_data, list):
+ return_data["msg"] = "未查询到交易行价格数据"
+ return return_data
+
+ try:
+ result = []
+ for price_item in price_data:
+ if not isinstance(price_item, dict):
+ continue
+
+ item_id = str(price_item.get("item_id") or "")
+ base_item = item_map.get(item_id)
+ if not base_item:
+ continue
+
+ timestamp = price_item.get("timestamp")
+ try:
+ created = datetime.fromtimestamp(int(timestamp)).strftime("%Y-%m-%d %H:%M:%S")
+ except (TypeError, ValueError, OSError):
+ created = ""
+
+ result.append(
+ {
+ "item_id": item_id,
+ "name": base_item.get("label", ""),
+ "icon": f"https://icon.jx3box.com/icon/{base_item.get('icon', '')}.png",
+ "server": price_item.get("server", server),
+ "price": price_item.get("price", 0),
+ "price_parts": gold_to_parts(price_item.get("price", 0)),
+ "sample": price_item.get("sample", 0),
+ "created": created,
+ }
+ )
+
+ if not result:
+ return_data["msg"] = "未查询到交易行价格数据"
+ return return_data
+
+ return_data["data"] = {
+ "search_name": name,
+ "server": server,
+ "matched_count": len(matched_items),
+ "result_count": len(result),
+ "list": result,
+ "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
+
+ # 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
\ No newline at end of file
diff --git a/core/message.py b/core/message.py
index 96676f5..6f87ac8 100644
--- a/core/message.py
+++ b/core/message.py
@@ -246,7 +246,7 @@ class MessageBuilder:
return
try:
- data = await self.jx3api.zili(name, server, choice)
+ data = await self.jx3box.zili(name, server, choice)
if data["code"] != 200:
await new_event.send(MessageChain().message(data.get("msg", "获取资历数据失败")))
controller.stop()
@@ -532,244 +532,160 @@ class MessageBuilder:
return await self.T2I_image_msg(event, lambda: self.jx3api.qixue(name,0))
async def liaotian(self, event: AstrMessageEvent, server:str, name: str, limit:int = 20, page:int = 1):
- """ 奇穴 心法"""
+ """ 聊天 服务器 角色 条数 页数"""
return await self.T2I_image_msg(event, lambda: self.jx3api.juesheliaotian(server,name,limit,page))
+ async def tongzhanyy(self, event: AstrMessageEvent, server: str = ""):
+ """ 统战 服务器"""
+ return await self.plain_msg(event, lambda: self.jx3api.tongzhanyy(server))
+ async def xiaoyao(self, event: AstrMessageEvent, name:str = ""):
+ """ 小药 心法"""
+ return await self.T2I_image_msg(event, lambda: self.jx3api.xiaoyao(name))
- async def keju(self, event: AstrMessageEvent,subject: str, limit: int = 5):
- """ 科举"""
- return await self.plain_msg(event, lambda: self.jx3api.keju(subject,limit))
-
-
- async def huajia(self, event: AstrMessageEvent, name: str= "", server: str = "", map: str= ""):
- """ 花价 名称 服务器 地图"""
- return await self.T2I_image_msg(event, lambda: self.jx3api.huajia(self.serverdefault(server),name,map))
+ async def pianzhi(self, event: AstrMessageEvent, uid: str, server:str = ""):
+ """ 骗子 uid 服务器"""
+ return await self.plain_msg(event, lambda: self.jx3api.pianzhi(server,uid))
+ async def huajia(self, event: AstrMessageEvent, server: str, name: str= "" , map: str= ""):
+ """ 花价 服务器 名称 地图"""
+ return await self.T2I_image_msg(event, lambda: self.jx3api.huajia(server,name,map))
async def zhuangshi(self, event: AstrMessageEvent, name: str):
""" 装饰 名称"""
return await self.T2I_image_msg(event, lambda: self.jx3api.zhuangshi(name))
-
async def qiwu(self, event: AstrMessageEvent, name: str):
""" 器物 地图名称"""
return await self.T2I_image_msg(event, lambda: self.jx3api.qiwu(name))
+ async def baishi(self, event: AstrMessageEvent, server: str, keyword: str = ""):
+ """ 拜师 服务器 关键词 """
+ return await self.T2I_image_msg(event, lambda: self.jx3api.shitu(2, server, keyword, 50))
- async def xinwen(self, event: AstrMessageEvent,num:int = 5):
- """ 新闻"""
- return await self.plain_msg(event, lambda: self.jx3api.xinwen(num))
+ async def shoutu(self, event: AstrMessageEvent, server: str, keyword: str = ""):
+ """ 收徒 服务器 关键词 """
+ return await self.T2I_image_msg(event, lambda: self.jx3api.shitu(1, server, keyword, 50))
+ async def weihu(self, event: AstrMessageEvent,limit:int = 5):
+ """ 维护 数量"""
+ return await self.plain_msg(event, lambda: self.jx3api.weihu(limit))
- async def weihu(self, event: AstrMessageEvent,num:int = 5):
- """ 维护"""
- return await self.plain_msg(event, lambda: self.jx3api.weihu(num))
-
+ async def xinwen(self, event: AstrMessageEvent,limit:int = 5):
+ """ 新闻 数量"""
+ return await self.plain_msg(event, lambda: self.jx3api.xinwen(limit))
- async def qufu(self, event: AstrMessageEvent,name: str = "双梦镇"):
- """ 区服"""
- return await self.plain_msg(event, lambda: self.jx3api.qufu(name))
+ async def tuanduizhaomu(self, event: AstrMessageEvent,server: str, keyword: str = ""):
+ """ 招募 服务器 副本"""
+ return await self.T2I_image_msg(event, lambda: self.jx3api.tuanduizhaomu( server,1,keyword,50))
+ async def tuanzhang(self, event: AstrMessageEvent,server: str, keyword: str = ""):
+ """ 团长 服务器 名字"""
+ return await self.T2I_image_msg(event, lambda: self.jx3api.tuanduizhaomu( server,2,keyword,50))
- async def kaifu(self, event: AstrMessageEvent,server: str = ""):
- """ 开服 服务器"""
- return await self.plain_msg(event, lambda: self.jx3api.kaifu(self.serverdefault(server)))
+ async def tuanpai(self, event: AstrMessageEvent,server: str, keyword: str = ""):
+ """ 团牌 服务器 内容"""
+ return await self.T2I_image_msg(event, lambda: self.jx3api.tuanduizhaomu( server,3,keyword,50))
+ async def daanzhishu(self, event: AstrMessageEvent):
+ """ 答案之书"""
+ return await self.plain_msg(event, self.jx3api.daanzhishu)
+
+ async def tiangou(self, event: AstrMessageEvent):
+ """ 舔狗语录"""
+ return await self.plain_msg(event, self.jx3api.tiangou)
+
+ async def fkxq4(self, event: AstrMessageEvent):
+ """ 疯狂星期四"""
+ return await self.plain_msg(event, lambda: self.jx3api.fengleiyulu("疯狂星期四"))
+
+ async def caihongpi(self, event: AstrMessageEvent):
+ """ 彩虹屁"""
+ return await self.plain_msg(event, lambda: self.jx3api.fengleiyulu("彩虹屁"))
+
+ async def dujitang(self, event: AstrMessageEvent):
+ """ 毒鸡汤"""
+ return await self.plain_msg(event, lambda: self.jx3api.fengleiyulu("毒鸡汤"))
+
+ async def pengyouquan(self, event: AstrMessageEvent):
+ """ 朋友圈"""
+ return await self.plain_msg(event, lambda: self.jx3api.fengleiyulu("朋友圈"))
+
+ async def heshengme(self, event: AstrMessageEvent,):
+ """ 喝什么"""
+ return await self.plain_msg(event, self.jx3api.heshengme)
+
+ async def chishengme(self, event: AstrMessageEvent,):
+ """ 吃什么"""
+ return await self.plain_msg(event, self.jx3api.chishengme)
+
+ async def shaohua(self, event: AstrMessageEvent,):
+ """ 骚话"""
+ return await self.plain_msg(event, self.jx3api.shaohua)
+
+ async def zhananyulu(self, event: AstrMessageEvent,):
+ """ 渣男语录"""
+ return await self.plain_msg(event, self.jx3api.zhananyulu)
+
+ async def keju(self, event: AstrMessageEvent,subject: str, limit: int = 5):
+ """ 科举 题目 条数"""
+ return await self.plain_msg(event, lambda: self.jx3api.keju(subject,limit))
async def zhuangtai(self, event: AstrMessageEvent):
- """ 状态"""
- return await self.T2I_image_msg(event, self.jx3api.zhuangtai)
+ """ 区服"""
+ return await self.T2I_image_msg(event, lambda: self.jx3api.zhuangtai(""))
+ async def kaifu(self, event: AstrMessageEvent,server: str):
+ """ 开服 服务器"""
+ return await self.plain_msg(event, lambda: self.jx3api.kaifu(server))
async def jigai(self, event: AstrMessageEvent,):
""" 技改"""
return await self.plain_msg(event, self.jx3api.jigai)
+ async def jiemi(self, event: AstrMessageEvent):
+ """ 解密"""
+ return await self.plain_msg(event, self.jx3api.jiemi)
- async def xiaoyao(self, event: AstrMessageEvent):
- """ 小药"""
- return await self.T2I_image_msg(event, self.jx3api.xiaoyao)
+ async def tiebawujia(self, event: AstrMessageEvent, name: str, server: str = "", limit: str = "5", ):
+ """ 贴吧物价 名称 服务器 数量 """
+ return await self.plain_msg(event, lambda: self.jx3api.tiebawujia(name, server, limit))
+ async def bagua(self, event: AstrMessageEvent, server:str = "", limit:str = 10):
+ """ 818 """
+ return await self.plain_msg(event, lambda: self.jx3api.bagua(818,server,limit))
+ async def diaoluo(self, event: AstrMessageEvent, name: str, server: str = "", limit: str = "20",):
+ """ 掉落 物品 服务器 数量 """
+ return await self.T2I_image_msg(event, lambda: self.jx3api.diaoluo(name, server, limit))
-
-
-
-
-
-
-
-
-
-
- async def zilipaihang(self, event: AstrMessageEvent):
- """ 资历排行 """
- return await self.T2I_image_msg(event, lambda: self.jx3api.zilipaixing("", ""))
-
-
-
-
- async def shaohua(self, event: AstrMessageEvent,):
- """ 骚话"""
- return await self.plain_msg(event, self.jx3api.shaohua)
+ async def hong(self, event: AstrMessageEvent,name: str = "易筋经"):
+ """ 宏 心法"""
+ return await self.handler_plain_image_msg(event, lambda: self.jx3box.hong1(name), self.jx3box.hong2)
async def zili(self, event: AstrMessageEvent, name: str, server: str = ""):
""" 资历 角色名称 服务器"""
return await self.handler_zili_msg(event, name, self.serverdefault(server))
-
-
- async def jiemi(self, event: AstrMessageEvent):
- """ 解密"""
- return await self.plain_msg(event, self.jx3api.jiemi)
-
-
-
-
-
-
-
-
- async def fuyaojjiutian(self, event: AstrMessageEvent,server: str = ""):
- """ 扶摇九天 服务器"""
- return await self.plain_msg(event, lambda: self.jx3api.fuyaojjiutian( self.serverdefault(server)))
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- async def tongzhanyy(self, event: AstrMessageEvent, server: str = ""):
- """ 统战歪歪 服务器"""
- return await self.plain_msg(event, lambda: self.jx3api.tongzhanyy(self.serverdefault(server)))
-
-
-
-
-
-
-
-
- async def pianzhi(self, event: AstrMessageEvent,qq: str):
- """ 骗子 QQ"""
- return await self.plain_msg(event, lambda: self.jx3api.pianzhi(qq))
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
- async def tuanduizhaomu(self, event: AstrMessageEvent,keyword: str = "25人普通会战弓月城", server: str = ""):
- """ 招募 副本 服务器"""
- return await self.T2I_image_msg(event, lambda: self.jx3api.tuanduizhaomu( self.serverdefault(server),keyword))
-
-
- async def baishi(self, event: AstrMessageEvent,keyword: str = "", server: str = ""):
- """ 拜师 搜索关键词 服务器"""
- return await self.T2I_image_msg(event, lambda: self.jx3api.shitu(2, keyword, self.serverdefault(server)))
-
-
- async def shoutu(self, event: AstrMessageEvent,keyword: str = "", server: str = ""):
- """ 收徒 搜索关键词 服务器"""
- return await self.T2I_image_msg(event, lambda: self.jx3api.shitu(1, keyword, self.serverdefault(server)))
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
-
async def jiaoyihang(self, event: AstrMessageEvent,Name: str = "守缺式",server: str = ""):
""" 交易行 物品名称 服务器"""
- return await self.T2I_image_msg(event, lambda: self.jx3api.jiaoyihang(Name, self.serverdefault(server)))
-
-
- async def tiebawujia(self, event: AstrMessageEvent, name: str = "狐金", limit_or_server: str = "5", server: str = ""):
- """ 贴吧物价 物品名称 数量 服务器"""
- limit = 5
- target_server = server
-
- if str(limit_or_server).isdigit():
- limit = int(limit_or_server)
- else:
- target_server = limit_or_server
-
- return await self.plain_msg(event, lambda: self.jx3api.tiebawujia(name, limit, self.serverdefault(target_server)))
-
-
- async def diaoluo(self, event: AstrMessageEvent, name: str = "玄晶", limit_or_server: str = "20", server: str = ""):
- """ 掉落 物品名称 数量 服务器"""
- limit = 20
- target_server = server
-
- if str(limit_or_server).isdigit():
- limit = int(limit_or_server)
- else:
- target_server = limit_or_server
-
- return await self.T2I_image_msg(event, lambda: self.jx3api.diaoluo(name, limit, self.serverdefault(target_server)))
-
-
- async def bagua(self, event: AstrMessageEvent,name: str = "818"):
- """ 八卦 类型"""
- return await self.plain_msg(event, lambda: self.jx3api.bagua(name))
+ return await self.T2I_image_msg(event, lambda: self.jx3box.jiaoyihang(Name, self.serverdefault(server)))
+
+
+
+
+
+
+
+
+
- async def hong(self, event: AstrMessageEvent,name: str = "易筋经"):
- """ 宏 心法"""
- return await self.handler_plain_image_msg(event, lambda: self.jx3api.hong1(name), self.jx3api.hong2)
diff --git a/main.py b/main.py
index ba9c458..2a609d6 100644
--- a/main.py
+++ b/main.py
@@ -19,7 +19,7 @@ from .core.fun_basic import load_as_base64
@register("astrbot_plugin_jx3",
"fxdyz",
"通过调用剑网三API接口获取游戏数据,处理发送。",
- "2.8",
+ "3.0",
"https://github.com/qsc20001102/astrbot_plugin_jx3"
)
class Jx3ApiPlugin(Star):
@@ -270,66 +270,44 @@ class Jx3ApiPlugin(Star):
"技能": self. jx3cmd.jineng,
"奇穴": self. jx3cmd.qixue,
"聊天": self. jx3cmd.liaotian,
-
-
-
-
-
- "科举": self. jx3cmd.keju,
+ "统战": self. jx3cmd.tongzhanyy,
+ "小药": self. jx3cmd.xiaoyao,
+ "骗子": self. jx3cmd.pianzhi,
"花价": self. jx3cmd.huajia,
"装饰": self. jx3cmd.zhuangshi,
"器物": self. jx3cmd.qiwu,
- "公告": self. jx3cmd.xinwen,
- "维护": self. jx3cmd.weihu,
- "区服": self. jx3cmd.qufu,
- "开服": self. jx3cmd.kaifu,
- "状态": self. jx3cmd.zhuangtai,
- "技改": self. jx3cmd.jigai,
- "小药": self. jx3cmd.xiaoyao,
-
-
- "全服资历排行": self. jx3cmd.zilipaihang,
-
- "骚话": self. jx3cmd.shaohua,
- "资历": self. jx3cmd.zili,
- "解密": self. jx3cmd.jiemi,
-
-
- "宏": self. jx3cmd.hong,
-
-
- "扶摇": self. jx3cmd.fuyaojjiutian,
-
-
-
-
-
- "统战": self. jx3cmd.tongzhanyy,
-
-
- "骗子": self. jx3cmd.pianzhi,
-
-
-
-
-
-
- "招募": self. jx3cmd.tuanduizhaomu,
"拜师": self. jx3cmd.baishi,
"收徒": self. jx3cmd.shoutu,
-
-
-
+ "维护": self. jx3cmd.weihu,
+ "新闻": self. jx3cmd.xinwen,
+ "招募": self. jx3cmd.tuanduizhaomu,
+ "团长": self. jx3cmd.tuanzhang,
+ "团牌": self. jx3cmd.tuanpai,
+ "答案之书": self. jx3cmd.daanzhishu,
+ "舔狗语录": self. jx3cmd.tiangou,
+ "疯狂星期四": self. jx3cmd.fkxq4,
+ "彩虹屁": self. jx3cmd.caihongpi,
+ "毒鸡汤": self. jx3cmd.dujitang,
+ "朋友圈": self. jx3cmd.pengyouquan,
+ "喝什么": self. jx3cmd.heshengme,
+ "吃什么": self. jx3cmd.chishengme,
+ "骚话": self. jx3cmd.shaohua,
+ "渣男语录": self. jx3cmd.zhananyulu,
+ "科举": self. jx3cmd.keju,
+ "区服": self. jx3cmd.zhuangtai,
+ "开服": self. jx3cmd.kaifu,
+ "技改": self. jx3cmd.jigai,
+ "解密": self. jx3cmd.jiemi,
+ "资历": self. jx3cmd.zili,
+ "宏": self. jx3cmd.hong,
-
-
-
-
+
- "八卦": self. jx3cmd.bagua,
+ "818": self. jx3cmd.bagua,
"交易行": self. jx3cmd.jiaoyihang,
"贴吧物价": self. jx3cmd.tiebawujia,
"掉落": self. jx3cmd.diaoluo,
+
"开服推送": self. jx3cmd.kaifhujiank,
"新闻推送": self. jx3cmd.xinwenzhixun,
"刷马推送": self. jx3cmd.shuamamsg,
diff --git a/metadata.yaml b/metadata.yaml
index a3c6332..0911404 100644
--- a/metadata.yaml
+++ b/metadata.yaml
@@ -1,7 +1,7 @@
name: astrbot_plugin_jx3
display_name: 剑网三游戏数据查询工具
desc: 通过接口调用剑网三API接口获取游戏数据,处理发送。
-version: v2.8
+version: v3.0
author: 飞翔大野猪
repo: https://github.com/qsc20001102/astrbot_plugin_jx3
astrbot_version: ">=4.11.0"
diff --git a/templates/diaoluo.html b/templates/diaoluo.html
index 117a319..2820fef 100644
--- a/templates/diaoluo.html
+++ b/templates/diaoluo.html
@@ -117,10 +117,10 @@
{% for item in items %}
- | {{ item.item_name }} |
+ {{ item.itemName }} |
{{ item.server }} |
- {{ item.map_name }} |
- {{ item.role_name }} |
+ {{ item.mapName }} |
+ {{ item.roleName }} |
{{ item.time }} |
{% endfor %}
diff --git a/templates/helps.html b/templates/helps.html
index 18288a4..9dec8ad 100644
--- a/templates/helps.html
+++ b/templates/helps.html
@@ -1,1403 +1,757 @@
-
-
-剑三机器人功能指令
-
+ .section.wide {
+ grid-column: span 2;
+ }
+ }
+
+
-
-
- ASTRBOT · JX3 COMMAND GUIDE
- 剑三机器人功能指令
- 按使用场景重新分组,保留每条指令的名称、使用示例与说明。卡片底色表示指令是否需要令牌,免令牌与需令牌功能一眼区分。
-
-
88条指令
-
13个功能分组
+
+
+
+
剑网三综合查询插件
+
+
+ 指令与参数之间使用空格分隔
+ 带有 [ ] 的参数为可选参数
-
+
-
-
- 底色说明
- 免令牌
- 需令牌
-
-
- [] 为可选参数;未填写服务器时使用默认服务器。
-
-
+
-
-
-
-
-
-
日
-
日常功能
+
+ 活动情报
+
+
+
+
活动日历
+
+ 日常 [延后天数]
+ 日常预测
+
-
-
-
-
-
日常活动
-
免令牌
+
+
关隘首领
+
+ 关隘
-
日常 [服务器] [天数]
-
查询指定日期的日常活动。
-
+
-
-
-
日常预测
-
免令牌
+
+
赤兔
+
+ 赤兔
+ 本周赤兔
-
月历
-
查看当前日期前后 15 天日常安排。
-
+
-
-
-
名望事件
-
免令牌
+
+
阵营事件
+
+ 阵营奉献 [阵营]
-
名望 [穹野卫]
-
查询指定名望势力事件。
-
+
-
-
-
诛恶
-
需令牌
+
+
烟花记录
+
+ 烟花 区服 角色
-
诛恶
-
查询诛恶事件服务器、地图和时间记录。
-
+
+
+
-
-
-
-
-
-
官
-
官方功能
+
+
+
+ 名剑相关
+
+
+
+
名剑战绩
+
+ 战绩 服务器 角色 [模式]
+
-
-
-
-
-
新闻资讯
-
免令牌
+
+
名剑统计
+
+ 名剑统计 [模式]
-
公告 [数量]
-
获取最新官方公告,数量默认为 5 条。
-
-
-
-
- 维护 [数量]
- 获取最新官方维护公告,数量默认为 5 条。
-
-
-
-
- 技改
- 获取最近一次官方职业技改内容。
-
-
-
-
- 区服 [服务器]
- 查询服务器基础信息。
-
-
-
-
- 开服 [服务器]
- 查询指定服务器开服状态。
-
-
-
-
- 状态
- 查询所有服务器当前状态。
-
+
-
-
-
-
-
-
角
-
角色功能
+
+
+
+ 物价交易
+
+
+
+
阵营拍卖
+
+ 阵营拍卖 服务器 [物品] [数量]
+
-
-
-
-
-
角色信息
-
需令牌
+
+
金价行情
+
+ 金价 服务器
-
角色 名称 [服务器]
-
查询角色基础信息。
-
+
-
-
-
角色资历
-
需令牌
+
+
物品价格
+
+ 物价 外观名称 [服务器]
-
资历 角色名 [服务器]
-
按分类查询角色资历点数完成进度。
-
+
-
-
-
角色名片
-
需令牌
+
+
成本计算
+
+ 成本 服务器 物品名称
-
名片 角色名 [服务器]
-
查询角色当前名片。
-
+
-
-
-
所有名片
-
需令牌
+
+
编号搜索
+
+ 看号 万宝楼编号
-
所有名片 角色名 [服务器]
-
查询角色已拥有或可展示的全部名片。
-
-
-
-
- 奇遇 角色名 [服务器]
- 查询指定角色已触发奇遇。
-
-
-
-
- 未做奇遇 角色名 [服务器]
- 查询指定角色尚未完成的奇遇。
-
-
-
-
- 战绩 角色名 [服务器] [模式]
- 查询角色名剑大会战绩,支持 22 / 33 模式。
-
-
-
-
- 精耐 角色名 [服务器]
- 查询角色百战精力、耐力和技能列表。
-
+
-
-
-
-
-
-
职
-
职业功能
+
+
+
+ 阵营战场
+
+
+
-
-
-
-
-
小药推荐
-
免令牌
+
+
诛恶事件
+
+ 诛恶 服务器
-
小药
-
查询指定心法常用小药、宴席与辅助搭配。
-
-
-
-
- 宏 心法
- 查询指定心法推荐宏设置。
-
-
-
-
- 配装 心法 [筛选]
- 查询指定心法配装,支持 PvP / PvE 筛选。
-
-
-
-
- 阵眼 心法
- 查询指定心法阵眼效果。
-
-
-
-
- 奇穴 心法
- 查询指定心法奇穴详情。
-
-
-
-
- 技能 心法
- 查询指定心法技能详情。
-
+
-
-
-
-
-
-
乐
-
休闲功能
+
+
+
+ 排行榜单
+
+
+
+
排行统计
+
+
+ 名士五十强 服务器
+ 老江湖五十强 服务器
+ 兵甲藏家五十强 服务器
+ 名师五十强 服务器
+ 阵营英雄五十强 服务器
+ 薪火相传五十强 服务器
+ 庐园广记一百强 服务器
+ 浩气神兵宝甲五十强 服务器
+ 恶人神兵宝甲五十强 服务器
+ 浩气爱心帮会五十强 服务器
+ 恶人爱心帮会五十强 服务器
+ 赛季恶人五十强 服务器
+ 赛季浩气五十强 服务器
+ 上周恶人五十强 服务器
+ 上周浩气五十强 服务器
+ 本周恶人五十强 服务器
+ 本周浩气五十强 服务器
+
-
-
-
-
-
- 烟花 角色名 [服务器]
- 查询指定角色烟花记录。
-
-
-
-
- 拜师 [搜索关键词] [服务器]
- 查询徒弟找师傅信息。
-
-
-
-
- 收徒 [搜索关键词] [服务器]
- 查询师傅找徒弟信息。
-
-
-
-
- 随机名片 职业 体型 [服务器]
- 按职业、体型、服务器随机抽取名片。
-
-
-
-
- 攻略 奇遇名称
- 查询指定奇遇的触发条件、奖励与流程攻略。
-
-
-
-
- 奇遇统计 奇遇名称 [服务器]
- 统计指定奇遇触发次数
-
-
-
-
- 近期奇遇 [服务器]
- 查询指定服务器最近触发的奇遇记录
-
-
-
-
- 奇遇汇总 [天数] [服务器]
- 统计指定服务器最近一段时间的奇遇触发汇总
-
-
-
-
-
-
-
营
-
阵营功能
+
+
+
+ 角色名片
+
+
+
-
-
-
-
-
阵营沙盘
-
免令牌
+
+
随机名片
+
+ 随机秀 服务器 [门派] [体型]
-
沙盘 [服务器]
-
查询指定服务器阵营沙盘与据点状态。
-
-
-
-
- 拍卖 物品名称 [服务器]
- 查询阵营拍卖物品及服务器记录。
-
-
-
-
- 阵营事件
- 查询跨服阵营事件记录。
-
-
-
-
- 关隘首领
- 查询各分线关隘首领状态与剩余时间。
-
-
-
-
- 帮战记录 [服务器]
- 查询指定服务器帮战记录。
-
-
-
-
- 统战 [服务器]
- 查询指定服务器阵营统战 YY 信息。
-
+
-
-
-
-
-
-
本
-
副本功能
+
+
+
+ 百战与角色
+
+
+
-
-
-
-
-
百战首领
-
需令牌
+
+
成就查询
+
+ 成就 服务器 角色 成就
-
百战
-
查询本期百战异闻录首领与关卡信息。
-
+
-
-
-
副本招募
-
需令牌
+
+
角色详情
+
+ 角色 服务器 名称
-
招募 副本名称 [服务器]
-
查询指定服务器副本招募信息。
-
-
-
-
- 掉落 物品名称 [数量] [服务器]
- 查询团队秘境贵重物品掉落记录
-
-
-
-
- 解密
- 查询当前机关解密结果与下轮时间。
-
+
-
-
-
-
-
-
活
-
活动功能
+
+
+
+ 奇遇相关
+
+
+
+
奇遇记录
+
+ 奇遇 服务器 角色
+ 查询 服务器 角色
+
-
-
-
-
-
扶摇九天
-
需令牌
+
+
奇遇汇总
+
+ 汇总 服务器 [天数]
-
扶摇 [服务器]
-
查询指定服务器扶摇九天活动记录。
-
+
-
-
-
本日赤兔
-
需令牌
+
+
近期奇遇
+
+ 近期 服务器 [数量]
-
本日赤兔
-
查询本日赤兔时间、区服和地图记录。
-
+
-
-
-
本周赤兔
-
需令牌
+
+
奇遇统计
+
+ 统计 奇遇 [服务器] [数量]
-
本周赤兔
-
查询本周赤兔时间、区服和地图记录。
-
+
-
-
-
马场刷新
-
需令牌
+
+
奇遇攻略
+
+ 攻略 奇遇
-
刷马 [服务器]
-
查询指定服务器马场刷新记录。
-
-
-
-
- 的卢 [服务器]
- 查询指定服务器的卢马刷新与捕获记录。
-
+
-
-
-
-
-
-
市
-
交易功能
+
+
+
+ 心法配装
+
+
+
-
-
-
-
-
金价查询
-
需令牌
+
+
资历榜单
+
+ 资历排行 [服务器] [门派]
-
金价 [服务器]
-
查询指定服务器金价行情。
-
+
-
-
-
外观物价
-
需令牌
+
+
技能信息
+
+ 技能 心法
-
物价 外观名称 [服务器]
-
查询指定外观在服务器内的物价行情。
-
+
-
-
-
交易行
-
免令牌
+
+
奇穴信息
+
+ 奇穴 心法
-
交易行 物品名称 [服务器]
-
查询指定物品交易行价格与挂单信息。
-
-
-
-
- 贴吧物价 物品名称 [记录数量] [服务器]
- 查询指定物品的贴吧交易记录。
-
+
-
-
-
-
-
-
杂
-
杂项功能
+
+
+
+ 游戏社区
+
+
+
+
角色聊天
+
+ 聊天 服务器 角色 [条数] [页数]
+
-
-
-
-
-
科举答题
-
免令牌
+
+
家园功能
+
+ 花价 服务器 [名称] [地图]
+ 装饰 名称
+ 器物 地图名称
-
科举 内容
-
按题目关键字查询科举答案。
-
+
-
-
-
家园鲜花
-
免令牌
+
+
师徒列表
+
+ 拜师 服务器 [关键词]
+ 收徒 服务器 [关键词]
-
花价 名称 [服务器] [家园地图]
-
查询家园鲜花价格,支持服务器与家园地图筛选。
-
+
-
-
-
家园装饰
-
免令牌
+
+
资讯公告
+
+ 维护 [数量]
+ 新闻 [数量]
-
装饰 名称
-
查询家园装饰物名称、来源与相关信息。
-
+
-
-
-
器物图谱
-
免令牌
+
+
团队招募
+
+ 招募 服务器 [副本]
+ 团长 服务器 [名称]
+ 团牌 服务器 [内容]
-
器物 地图名称
-
查询指定地图可收集的器物图谱。
-
+
-
-
-
八卦记录
-
需令牌
+
+
骚话语录
+
+ 答案之书
+ 舔狗语录
+ 疯狂星期四
+ 彩虹屁
+ 毒鸡汤
+ 朋友圈
+ 喝什么
+ 吃什么
+ 骚话
+ 渣男语录
-
八卦 类型
-
查询 818 / 616 / 避雷等八卦记录。
-
-
-
-
- 骗子 QQ号
- 按 QQ 号查询骗子记录。
-
-
-
-
- 骚话
- 随机返回一句趣味骚话,用于群内互动。
-
+
-
-
-
-
-
-
榜
-
排行功能
+
+
+
+ 百度贴吧
+
+
+
+
外观记录
+
+ 贴吧物价 名称 [服务器] [数量]
+
-
-
-
-
-
- 名剑统计 [模式]
- 统计指定模式名剑大会数据。
-
-
-
-
- 名剑排行 [数量] [模式]
- 查询名剑大会排名,支持 22 / 33 模式。
-
-
-
-
- 资历排行 职业 [服务器]
- 查询指定职业资历排行榜。
-
-
-
-
- 试炼排行 心法 [服务器]
- 查询指定心法木桩试炼排行榜。
-
-
-
-
- 名士五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 老江湖五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 兵甲藏家五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 名师五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 阵营英雄五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 薪火相传五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 庐园广记一百强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 浩气神兵宝甲五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 恶人神兵宝甲五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 浩气爱心帮会五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 恶人爱心帮会五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 赛季恶人五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 赛季浩气五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 上周恶人五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 上周浩气五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 本周恶人五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
- 本周浩气五十强 [服务器]
- 查询风云录排行榜数据。
-
-
-
-
-
-
-
推
-
推送功能
+
+
+
+ 系统工具
+
+
+
-
-
-
-
-
开服推送
-
免令牌
+
+
技改记录
+
+ 技改
-
开服推送
-
订阅或管理服务器开服状态推送。
-
+
-
-
-
新闻推送
-
免令牌
+
+
秘境方位
+
+ 解密
-
新闻推送
-
订阅或管理官方新闻公告推送。
-
+
-
-
-
刷马推送
-
免令牌
+
+
掉落统计
+
+ 掉落 物品 [服务器] [数量]
-
刷马推送
-
订阅或管理刷马刷新推送。
-
-
-
-
- 赤兔推送
- 订阅或管理赤兔马刷新推送。
-
+
-
-
-
-
-
-
避
-
避雷功能
+
+
+
+ 魔盒数据
+
+
+
-
-
-
-
-
- 避雷查看
- 查看当前群内避雷记录列表。
-
-
-
-
- 避雷添加 名称 备注
- 新增一条避雷记录。
-
-
-
-
- 避雷删除 ID
- 按 ID 删除避雷记录。
-
-
-
-
- 避雷修改 ID 名称 备注
- 按 ID 修改避雷名称与备注。
-
-
-
-
- 避雷查询 名称
- 按名称关键字查询避雷记录。
-
-
-
-
+
-
-
+
+
+
+
+
-
+