qsc20001102/astrbot_plugin_jx3
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ok

Browse Source
This commit is contained in:
qsc
2026-02-15 16:16:46 +08:00
parent 0630492d95
commit 22b62d6d45
5 changed files with 138 additions and 147 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/bilei_data.py
+1 -1
View File
@@ -12,7 +12,7 @@ from astrbot.api import AstrBotConfig
from .sqlite import AsyncSQLiteDB
from .fun_basic import load_template
class BiLeidata:
def __init__(self,sqlite:AsyncSQLiteDB):
def __init__(self, sqlite:AsyncSQLiteDB):
# 引用sqlite
self._sql_db = sqlite
core/jx3_data.py
+34 -67
View File
@@ -15,12 +15,14 @@ from .sqlite import AsyncSQLiteDB
from .fun_basic import load_template,gold_to_string,week_to_num,compare_date_str
class JX3Service:
def __init__(self, api_config, config:AstrBotConfig):
def __init__(self, api_config, config:AstrBotConfig, sqlite:AsyncSQLiteDB):
self._api = APIClient()
# 引用API配置文件
self._api_config = api_config
# 引用插件配置文件
self._config = config
# 引用sqlite
self._sql_db = sqlite
# 获取配置中的 Token
self.token = self._config.get("jx3api_token", "")
@@ -48,7 +50,8 @@ class JX3Service:
return {
"code": 0,
"msg": "功能函数未执行",
"data": {}
"data": {},
"temp": ""
}
@@ -1132,8 +1135,8 @@ class JX3Service:
data_new = data[0]
data_old = data[1]
result_msg = f"{server}\n"
result_msg += f"上次[扶摇九天]开启时间：\n{datetime.fromtimestamp(data_old['time']).strftime('%Y-%m-%d %H:%M:%S')}\n"
result_msg += f"下次[扶摇九天]开启时间：\n{datetime.fromtimestamp(data_new['time']).strftime('%Y-%m-%d %H:%M:%S')}"
result_msg += f"上次[扶摇九天]开启时间：{datetime.fromtimestamp(data_old['time']).strftime('%Y-%m-%d %H:%M:%S')}\n"
result_msg += f"本次[扶摇九天]开启时间：{datetime.fromtimestamp(data_new['time']).strftime('%Y-%m-%d %H:%M:%S')}"
return_data["data"] = result_msg
except Exception as e:
logger.error(f"处理返回数据失败: {e}")
@@ -1168,8 +1171,7 @@ class JX3Service:
result_msg += f"黑戈壁：\n{_data['黑戈壁'][0]}\n"
result_msg += f"阴山大草原：\n{_data['阴山大草原'][0]}\n"
result_msg += f"鲲鹏岛：\n{_data['鲲鹏岛'][0]}\n"
result_msg += f"的卢:\n{_data['龙泉府 / 进图（21:10）'][0]}\n"
result_msg += f"赤兔:\n{data['data']['note']}"
result_msg += f"{data['data']['note']}"
return_data["data"] = result_msg
except Exception as e:
logger.error(f"处理返回数据失败: {e}")
@@ -1314,10 +1316,24 @@ class JX3Service:
return return_data
async def hong1(self, kungfu: str) -> Dict[str, Any]:
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)
)
kungfu = result.get("name",None)
if kungfu == None:
return_data["msg"] = "未找到该心法"
return return_data
logger.debug(f"查询到数据：{kungfu}")
# 1. 构造请求参数
params = {"kungfu": kungfu}
@@ -1341,7 +1357,7 @@ class JX3Service:
n += 1
return_data["msg"] = msg
return_data["data"]["pid"] = pid_list
return_data["data"]["list"] = pid_list
return_data["data"]["num"] = n
except Exception as e:
@@ -1392,69 +1408,20 @@ class JX3Service:
"""配装"""
return_data = self._init_return_data()
name_map = {
"冰心":10081,
"云裳心经":10080,
"奶秀":10080,
"花间游":10021,
"离经易道":10028,
"奶花":10028,
"毒经":10175,
"补天诀":10176,
"奶毒":10176,
"莫问":10447,
"相知":10448,
"奶歌":10448,
"无方":10627,
"灵素":10626,
"奶药":10626,
"傲血战意":10026,
"铁牢律":10062,
"策T":10062,
"易筋经":10003,
"洗髓经":10002,
"秃T":10002,
"焚影圣诀":10242,
"明尊琉璃体":10243,
"喵T":10243,
"分山劲":10390,
"铁骨衣":10389,
"苍T":10389,
"紫霞功":10014,
"气纯":10014,
"太虚剑意":10015,
"剑纯":10015,
"天罗诡道":10225,
"惊羽诀":10224,
"问水诀":10144,
"藏剑":10144,
"笑尘诀":10268,
"丐帮":10268,
"北傲诀":10464,
"霸刀":10464,
"凌海诀":10533,
"蓬莱":10533,
"隐龙诀":10585,
"凌雪":10585,
"太玄经":10615,
"衍天":10615,
"孤锋诀":10698,
"刀宗":10698,
"山海心诀":10756,
"万灵":10698,
"周天功":10786,
"段式":10786,
"幽罗引":10821,
"无相楼":10821
# 数据库查询数据
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)
)
}
mount = name_map.get(name,None)
mount = result.get("pzid",None)
if mount == None:
return_data["msg"] = "未找到该心法配装"
return_data["msg"] = "未找到该心法"
return return_data
logger.debug(mount)
logger.debug(f"查询到数据：{mount}")
# 1. 构造请求参数
params = {"mount": mount, "tags": tags}
core/sqlite.py
+2
View File
@@ -1,3 +1,5 @@
# pyright: reportOptionalMemberAccess=false
import aiosqlite
from typing import Any, Dict, List, Optional, Tuple
data/sqlite.db
BIN
View File
Binary file not shown.
main.py
+101 -79
View File
@@ -36,16 +36,19 @@ class Jx3ApiPlugin(Star):
# 本地数据存储路径
self.local_data_dir = StarTools.get_data_dir("astrbot_plugin_jx3")
# 插件数据存储路径
self.data_file_path = Path(__file__).parent / "data"
# SQLite本地路径
self.sqlite_path = Path(self.local_data_dir) /"sqlite.db"
logger.info(f"SQLite数据文件路径：{self.sqlite_path}")
# 插件自带数据文件路径
self.data_file_path = Path(__file__).parent / "data"
logger.info(f"本地数据文件路径：{self.sqlite_path}")
# SQLite插件路径
self.sqlite_plugin = Path(self.data_file_path) /"sqlite.db"
logger.info(f"插件数据文件路径：{self.sqlite_plugin}")
# 读取API配置文件
self.api_file_path = Path(__file__).parent / "data" / "api_config.json"
self.api_file_path = self.data_file_path / "api_config.json"
with open(self.api_file_path, 'r', encoding='utf-8') as f:
self.api_config = json.load(f)
@@ -100,12 +103,14 @@ class Jx3ApiPlugin(Star):
user TEXT
)
""")
self.sql_db_plugin = AsyncSQLiteDB(self.sqlite_plugin) # pyright: ignore[reportArgumentType]
await self.sql_db_plugin.connect()
# 避雷功能 实例化
self.bilei = BiLeidata(self.sql_db)
# 剑三功能 实例化
self.jx3fun = JX3Service(self.api_config, self.conf)
self.jx3fun = JX3Service(self.api_config, self.conf, self.sql_db_plugin)
# 后台推送 实例化
self.at = AsyncTask(self.context, self.conf, self.jx3fun)
@@ -136,6 +141,11 @@ class Jx3ApiPlugin(Star):
if self.sql_db:
await self.sql_db.close()
self.sql_db = None
if self.sql_db_plugin:
await self.sql_db_plugin.close()
self.sql_db_plugin = None
logger.info("jx3api插件已卸载/停用")
@@ -345,6 +355,85 @@ class Jx3ApiPlugin(Star):
await event.send(event.plain_result("猪脑过载，请稍后再试"))
async def handler_plain_image_msg(self, event: AstrMessageEvent, action1, action2):
"""两轮会话消息发送通用，先文本列表等反馈序号在发送图片"""
# 会话触发
try:
# 获取一轮数据
data = await action1()
if data["code"] == 200:
# 发送一轮消息
await event.send(event.plain_result(data["msg"]))
# 获取触发用户ID
user_id = event.get_sender_id()
# 二轮会话流程
@session_waiter(timeout=30)
async def macro_select_waiter(controller: SessionController,new_event: AstrMessageEvent):
# 跳过非触发用户消息
if new_event.get_sender_id() != user_id:
return
# 获取用户消息
msg = new_event.get_message_str().strip()
# 判断消息是否为数字
if not msg.isdigit():
await new_event.send(
MessageChain().message("输入异常，结束会话")
)
controller.stop()
return
# 判断数字是否在有效值内
num = int(msg)
if num < 1 or num > data["data"]["num"]:
await new_event.send(
MessageChain().message("无效序号，结束会话")
)
controller.stop()
return
# 获取二轮数据
try:
data1 = await action2(data["data"]["list"][num])
if data1["code"] != 200:
await new_event.send(
MessageChain().message("获取详细数据失败")
)
controller.stop()
return
# 消息拼接发送
chain = MessageChain()
msg_text = data1["data"]
chain.message(msg_text)
if data1["temp"] != "":
url = await self.html_render(data1["temp"], {}, options={})
chain.url_image(url)
await new_event.send(chain)
except Exception as e:
logger.error(f"功能函数执行错误: {e}")
await new_event.send(
MessageChain().message("猪脑过载，请稍后再试")
)
controller.stop()
# 二轮会话激活
try:
await macro_select_waiter(event)
except TimeoutError:
await event.send(event.plain_result("选择超时，已结束会话"))
except Exception:
logger.error("宏选择发生异常", exc_info=True)
else:
await event.send(event.plain_result(f"未搜索到相关内容"))
return
except Exception as e:
logger.error(f"功能函数执行错误: {e}")
await event.send(event.plain_result("猪脑过载，请稍后再试"))
async def jx3_helps(self, event: AstrMessageEvent):
"""剑三 功能"""
@@ -426,76 +515,9 @@ class Jx3ApiPlugin(Star):
return await self.T2I_image_msg(event, lambda: self.jx3fun.qiyugonglue(name))
async def jx3_hong(self, event: AstrMessageEvent, kungfu: str = "易筋经"):
async def jx3_hong(self, event: AstrMessageEvent,name: str = "易筋经"):
"""剑三 宏 心法"""
# 获取宏列表
try:
data = await self.jx3fun.hong1(kungfu)
if data["code"] == 200:
await event.send(event.plain_result(data["msg"]))
# 获取用户ID
user_id = event.get_sender_id()
# 等等用户回复
@session_waiter(timeout=30)
async def macro_select_waiter(controller: SessionController,new_event: AstrMessageEvent):
if new_event.get_sender_id() != user_id:
return
msg = new_event.get_message_str().strip()
if not msg.isdigit():
await new_event.send(
MessageChain().message("输入异常，结束会话")
)
controller.stop()
return
num = int(msg)
if num < 1 or num > data["data"]["num"]:
await new_event.send(
MessageChain().message("无效序号，结束会话")
)
controller.stop()
return
try:
data1 = await self.jx3fun.hong2(data["data"]["pid"][num])
if data1["code"] != 200:
await new_event.send(
MessageChain().message("获取详细宏数据失败")
)
controller.stop()
return
chain = MessageChain()
if data1["temp"] != "":
url = await self.html_render(data1["temp"], {}, options={})
chain.url_image(url)
msg_text = data1["data"]
chain.message(msg_text)
await new_event.send(chain)
except Exception as e:
logger.error(f"功能函数执行错误: {e}")
await new_event.send(
MessageChain().message("猪脑过载，请稍后再试")
)
controller.stop()
try:
await macro_select_waiter(event)
except TimeoutError:
await event.send(event.plain_result("选择宏超时，已结束会话"))
except Exception:
logger.error("宏选择发生异常", exc_info=True)
else:
await event.send(event.plain_result(f"未搜索到与【{kungfu}】相关的宏"))
return
except Exception as e:
logger.error(f"功能函数执行错误: {e}")
await event.send(event.plain_result("猪脑过载，请稍后再试"))
return await self.handler_plain_image_msg(event, lambda: self.jx3fun.hong1(name), self.jx3fun.hong2)
async def jx3_peizhuang(self, event: AstrMessageEvent,name: str = "易筋经", tags: str = ""):
@@ -606,24 +628,24 @@ class Jx3ApiPlugin(Star):
async def jx3_kaifhujiank(self, event: AstrMessageEvent):
"""剑三 开服监控"""
return_msg = await self.at.get_task_info("kfjk")
yield event.plain_result(return_msg)
await event.send(event.plain_result(return_msg))
async def jx3_xinwenzhixun(self, event: AstrMessageEvent):
"""剑三 新闻推送"""
return_msg = await self.at.get_task_info("xwzx")
yield event.plain_result(return_msg)
await event.send(event.plain_result(return_msg))
async def jx3_shuamamsg(self, event: AstrMessageEvent):
"""剑三 刷马推送"""
return_msg = await self.at.get_task_info("smxx")
yield event.plain_result(return_msg)
await event.send(event.plain_result(return_msg))
async def jx3_chitusg(self, event: AstrMessageEvent):
"""剑三 赤兔推送"""
return_msg = await self.at.get_task_info("ctxx")
yield event.plain_result(return_msg)
await event.send(event.plain_result(return_msg))