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

fxdyz

Browse Source
This commit is contained in:
qsc
2026-09-04 22:38:16 +08:00
parent 3845950862
commit 9c9808adf7
30 changed files with 658 additions and 510 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/jx3api_data.py
+88 -26
View File
@@ -15,7 +15,62 @@ from .sqlite import AsyncSQLiteDB
from .fun_basic import load_template,gold_to_parts,week_to_num,compare_date_str,format_time,format_remaining
ROLE_RANK_NAMES = {
"名士五十强",
"老江湖五十强",
"兵甲藏家五十强",
"名师五十强",
"阵营英雄五十强",
"薪火相传五十强",
"庐园广记一百强",
}
TONG_RANK_NAMES0 = {
"赛季恶人五十强",
"赛季浩气五十强",
"本周恶人五十强",
"本周浩气五十强",
}
TONG_RANK_NAMES1 = {
"浩气神兵宝甲五十强",
"恶人神兵宝甲五十强",
"浩气爱心帮会五十强",
"恶人爱心帮会五十强",
}
TONG_RANK_NAMES2 = {
"上周恶人五十强",
"上周浩气五十强",
}
GUILD_RANK_OPTIONS = {
"1": "浩气神兵宝甲五十强",
"2": "恶人神兵宝甲五十强",
"3": "浩气爱心帮会五十强",
"4": "恶人爱心帮会五十强",
}
CAMP_RANK_OPTIONS = {
"1": "赛季恶人五十强",
"2": "赛季浩气五十强",
"3": "上周恶人五十强",
"4": "上周浩气五十强",
"5": "本周恶人五十强",
"6": "本周浩气五十强",
}
OTHER_RANK_OPTIONS = {
"1": "名士五十强",
"2": "老江湖五十强",
"3": "兵甲藏家五十强",
"4": "名师五十强",
"5": "阵营英雄五十强",
"6": "薪火相传五十强",
"7": "庐园广记一百强",
}
RANK_NAMES = frozenset().union(
ROLE_RANK_NAMES,
TONG_RANK_NAMES0,
TONG_RANK_NAMES1,
TONG_RANK_NAMES2,
)
class JX3APIService:
@@ -456,33 +511,36 @@ class JX3APIService:
)
async def banghui_rank_menu(self) -> Dict[str, str]:
"""帮会排行榜选择内容。"""
return GUILD_RANK_OPTIONS.copy()
async def zhenying_rank_menu(self) -> Dict[str, str]:
"""阵营排行榜选择内容。"""
return CAMP_RANK_OPTIONS.copy()
async def qita_rank_menu(self) -> Dict[str, str]:
"""其他排行榜选择内容。"""
return OTHER_RANK_OPTIONS.copy()
async def rank_statistical_select(self, server: str, selected: Dict[str, Any],) -> Dict[str, Any]:
"""排行榜次轮：根据单项选择数据查询对应榜单。"""
if not isinstance(selected, dict) or len(selected) != 1:
return_data = self._init_return_data()
return_data["msg"] = "排行榜选项数据格式异常"
return return_data
rank_name = str(next(iter(selected.values())) or "").strip()
if rank_name not in RANK_NAMES:
return_data = self._init_return_data()
return_data["msg"] = "无效排行榜选项"
return return_data
return await self.rank_statistical(rank_name, server)
async def rank_statistical(self, name: str, server: str) -> Dict[str, Any]:
"""排行榜单"""
ROLE_RANK_NAMES = {
"名士五十强",
"老江湖五十强",
"兵甲藏家五十强",
"名师五十强",
"阵营英雄五十强",
"薪火相传五十强",
"庐园广记一百强",
}
TONG_RANK_NAMES0 = {
"赛季恶人五十强",
"赛季浩气五十强",
"本周恶人五十强",
"本周浩气五十强",
}
TONG_RANK_NAMES1 = {
"浩气神兵宝甲五十强",
"恶人神兵宝甲五十强",
"浩气爱心帮会五十强",
"恶人爱心帮会五十强",
}
TONG_RANK_NAMES2 = {
"上周恶人五十强",
"上周浩气五十强",
}
if name in ROLE_RANK_NAMES:
template_name = "rank_role.html"
elif name in TONG_RANK_NAMES0:
@@ -494,7 +552,11 @@ class JX3APIService:
# 数据处理
async def processor(data: Any, return_data: Dict[str, Any]) -> None:
items = data.get("data", [])
items = data.get("data") or []
if isinstance(items, list):
items = items[:50]
else:
items = []
return_data["data"] = {
"items": items,
core/kungfu_alias.py
+15 -17
View File
@@ -73,17 +73,21 @@ class KungfuAliasService:
for row in rows
]
async def save(self, pzid: Any, name: Any, aliases: Iterable[Any]):
async def save_aliases(self, pzid: Any, aliases: Iterable[Any]):
"""只更新已有心法的别名，不允许修改 ID 和标准名称。"""
normalized_pzid = self._parse_pzid(pzid)
normalized_name = self._clean(name)
if not normalized_name:
raise ValueError("标准心法名不能为空")
if len(normalized_name) > 64:
raise ValueError("标准心法名过长")
rows = await self.list_kungfu()
current = next(
(row for row in rows if row["pzid"] == normalized_pzid),
None,
)
if current is None:
raise ValueError("心法不存在")
normalized_name = current["name"]
normalized_aliases = self._normalize_aliases(normalized_name, aliases)
occupied: dict[str, str] = {}
for row in await self.list_kungfu():
for row in rows:
if row["pzid"] == normalized_pzid:
continue
for value in [row["name"], *row["aliases"]]:
@@ -100,17 +104,11 @@ class KungfuAliasService:
]
await self.sql.execute(
"""
INSERT INTO kungfu (pzid, name, name1, name2, name3, name4, name5)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(pzid) DO UPDATE SET
name=excluded.name,
name1=excluded.name1,
name2=excluded.name2,
name3=excluded.name3,
name4=excluded.name4,
name5=excluded.name5
UPDATE kungfu
SET name1=?, name2=?, name3=?, name4=?, name5=?
WHERE pzid=?
""",
(normalized_pzid, normalized_name, *values),
(*values, normalized_pzid),
)
def _normalize_aliases(self, name: str, aliases: Iterable[Any]) -> list[str]:
core/message.py
+92 -82
View File
@@ -1,5 +1,6 @@
from typing import Any
from aiocqhttp.exceptions import ActionFailed
from astrbot.core import html_renderer
from astrbot.api import logger
from astrbot.api.event import AstrMessageEvent, MessageChain
@@ -16,18 +17,67 @@ from .bilei_data import BiLeidata
class MessageBuilder:
"""回复消息构建"""
_RENDER_FORMATS = {"jpeg", "png"}
_DEVICE_SCALE_FACTOR_LEVELS = {
1.0: "normal",
1.3: "high",
1.8: "ultra",
}
def __init__(self,
jx3api: JX3APIService,
jx3box: JX3BOXService,
bilei: BiLeidata,
event_push: EventPushService,
icons: dict[str, dict[str, str]]
icons: dict[str, dict[str, str]],
render_config: dict[str, Any] | None = None,
):
self.jx3api = jx3api
self.jx3box = jx3box
self.bilei = bilei
self.event_push = event_push
self.icons = icons
self.render_config = render_config if isinstance(render_config, dict) else {}
def _build_render_options(
self,
overrides: dict[str, Any] | None = None,
) -> dict[str, Any]:
"""根据插件配置构造 AstrBot HTML 截图参数。"""
image_format = str(self.render_config.get("format", "jpeg")).lower()
if image_format not in self._RENDER_FORMATS:
image_format = "jpeg"
try:
device_scale_factor = float(
self.render_config.get("device_scale_factor", 1.3)
)
except (TypeError, ValueError):
device_scale_factor = 1.3
scale_factor_level = self._DEVICE_SCALE_FACTOR_LEVELS.get(
device_scale_factor,
"high",
)
try:
jpeg_quality = int(self.render_config.get("jpeg_quality", 100))
except (TypeError, ValueError):
jpeg_quality = 100
jpeg_quality = max(1, min(100, jpeg_quality))
options: dict[str, Any] = {
"quality": jpeg_quality,
"device_scale_factor_level": scale_factor_level,
"scale": "device",
"full_page": True,
"omit_background": False,
"type": image_format,
}
options.update(overrides or {})
if options.get("type") == "png":
options.pop("quality", None)
return options
async def html_render(
@@ -42,7 +92,7 @@ class MessageBuilder:
tmpl,
data,
return_url=return_url,
options=options,
options=self._build_render_options(options),
)
@@ -56,7 +106,7 @@ class MessageBuilder:
await event.send(event.plain_result(data["msg"]))
except Exception as e:
logger.error(f"功能函数执行错误: {e}")
await event.send(event.plain_result("猪脑过载，请稍后再试"))
await event.send(event.plain_result("猪脑过载，请稍后再试"))
async def T2I_image_msg(
@@ -69,22 +119,25 @@ class MessageBuilder:
data = await action()
try:
if data["code"] == 200:
options = {
"quality": 100,
"device_scale_factor_level": "normal",
"full_page": True,
"omit_background": False,
"type": "jpeg"
}
options.update(render_options or {})
if options.get("type") == "png":
options.pop("quality", None)
data["data"]["icons"] = self.icons
url = await self.html_render(data["temp"], data["data"], options=options)
url = await self.html_render(
data["temp"],
data["data"],
options=render_options,
)
await event.send(event.image_result(url))
else:
await event.send(event.plain_result(data["msg"]))
except ActionFailed as e:
if e.retcode == 1200:
logger.warning(
"图片消息发送回执超时，消息可能已经成功送达，"
"不再发送错误提示。"
)
return
logger.error(f"功能函数执行错误: {e}")
await event.send(event.plain_result("猪脑过载，请稍后再试"))
except Exception as e:
logger.error(f"功能函数执行错误: {e}")
await event.send(event.plain_result("猪脑过载，请稍后再试"))
@@ -131,7 +184,7 @@ class MessageBuilder:
if data.get("data"):
chain.message(str(data["data"]))
if data.get("temp"):
url = await self.html_render(data["temp"], {}, options={})
url = await self.html_render(data["temp"], {})
chain.url_image(url)
await event.send(chain)
except Exception as e:
@@ -191,6 +244,9 @@ class MessageBuilder:
controller.stop()
return
# 已取得有效选择，先结束等待计时，再执行可能超过 timeout 的
# API 请求与图片渲染，避免超时分支重复发送默认第一项。
controller.stop()
try:
await send_selected(
new_event,
@@ -200,8 +256,6 @@ class MessageBuilder:
logger.error(f"二轮查询执行错误: {e}")
await new_event.send(MessageChain().message("猪脑过载，请稍后再试"))
controller.stop()
try:
await option_select_waiter(event)
except TimeoutError:
@@ -291,73 +345,29 @@ class MessageBuilder:
""" 名剑统计 模式"""
return await self.T2I_image_msg(event, lambda: self.jx3api.mingjiantongji(mode))
async def mingshiwushiqiang(self, event: AstrMessageEvent, server: str):
""" 名士五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("名士五十强", server))
async def banghuipaihang(self, event: AstrMessageEvent, server: str):
"""帮会排行 服务器。"""
return await self.handler_plain_image_msg(
event,
lambda: self.jx3api.banghui_rank_menu(),
lambda selected: self.jx3api.rank_statistical_select(server,selected,),
)
async def laojianghuwushiqiang(self, event: AstrMessageEvent, server: str):
""" 老江湖五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("老江湖五十强", server))
async def zhenyingpaihang(self, event: AstrMessageEvent, server: str):
"""阵营排行 服务器。"""
return await self.handler_plain_image_msg(
event,
lambda: self.jx3api.zhenying_rank_menu(),
lambda selected: self.jx3api.rank_statistical_select(server,selected,),
)
async def bingjiacangjiawushiqiang(self, event: AstrMessageEvent, server: str):
""" 兵甲藏家五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("兵甲藏家五十强", server))
async def mingshiwushiqiang_mentor(self, event: AstrMessageEvent, server: str):
""" 名师五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("名师五十强", server))
async def zhengyingyingxiongwushiqiang(self, event: AstrMessageEvent, server: str):
""" 阵营英雄五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("阵营英雄五十强", server))
async def xinhuoxiangchuanwushiqiang(self, event: AstrMessageEvent, server: str):
""" 薪火相传五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("薪火相传五十强", server))
async def luyuanguangjiyibaiqiang(self, event: AstrMessageEvent, server: str):
""" 庐园广记一百强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("庐园广记一百强", server))
async def haoqishenbingbaojiawushiqiang(self, event: AstrMessageEvent, server: str):
""" 浩气神兵宝甲五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("浩气神兵宝甲五十强", server))
async def erenshenbingbaojiawushiqiang(self, event: AstrMessageEvent, server: str ):
""" 恶人神兵宝甲五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("恶人神兵宝甲五十强", server))
async def haoqiaixinbanghuiwushiqiang(self, event: AstrMessageEvent, server: str ):
""" 浩气爱心帮会五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("浩气爱心帮会五十强", server))
async def erenaixinbanghuiwushiqiang(self, event: AstrMessageEvent, server: str ):
""" 恶人爱心帮会五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("恶人爱心帮会五十强", server))
async def saijierenwushiqiang(self, event: AstrMessageEvent, server: str):
""" 赛季恶人五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("赛季恶人五十强", server))
async def saijihaoqiwushiqiang(self, event: AstrMessageEvent, server: str ):
""" 赛季浩气五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("赛季浩气五十强", server))
async def shangzhouerenwushiqiang(self, event: AstrMessageEvent, server: str ):
""" 上周恶人五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("上周恶人五十强", server))
async def shangzhouhaoqiwushiqiang(self, event: AstrMessageEvent, server: str ):
""" 上周浩气五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("上周浩气五十强", server))
async def benzhouerenwushiqiang(self, event: AstrMessageEvent, server: str ):
""" 本周恶人五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("本周恶人五十强", server))
async def benzhouhaoqiwushiqiang(self, event: AstrMessageEvent, server: str ):
""" 本周浩气五十强 服务器"""
return await self.T2I_image_msg(event, lambda: self.jx3api.rank_statistical("本周浩气五十强", server))
async def qitapaihang(self, event: AstrMessageEvent, server: str):
"""其他排行 服务器。"""
return await self.handler_plain_image_msg(
event,
lambda: self.jx3api.qita_rank_menu(),
lambda selected: self.jx3api.rank_statistical_select(server,selected,),
)
async def shilianpaixing(self, event: AstrMessageEvent, server: str , name: str):
""" 试炼排行 服务器 心法 """