ElapsingDreams/LiteyukiBot
1
0
Fork 0
forked from bot/app
Code Pull Requests Activity

🐛 修复npm无法显示的问题

Browse Source
This commit is contained in:
远野千束 2024-08-20 21:39:01 +08:00
parent 598bff8c49
commit 9e2bbe2e5c
10 changed files with 31 additions and 489 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
src/liteyuki_main/core.py
View File

@ -25,7 +25,9 @@ from ..utils.base.ly_function import get_function
require("nonebot_plugin_alconna")
require("nonebot_plugin_apscheduler")
from nonebot_plugin_alconna import on_alconna, Alconna, Args, Arparma, MultiVar
require("nonebot_plugin_htmlrender")
from nonebot_plugin_htmlrender import md_to_pic
from nonebot_plugin_alconna import UniMessage, on_alconna, Alconna, Args, Arparma, MultiVar
from nonebot_plugin_apscheduler import scheduler
driver = get_driver()
@ -54,7 +56,7 @@ async def _(bot: T_Bot, matcher: Matcher, result: Arparma):
permission=SUPERUSER
).handle()
# Satori OK
async def _(bot: T_Bot, event: T_MessageEvent):
async def _(bot: T_Bot, event: T_MessageEvent, matcher: Matcher):
# 使用git pull更新
ulang = get_user_lang(str(event.user.id if isinstance(event, satori.event.Event) else event.user_id))
@ -64,7 +66,9 @@ async def _(bot: T_Bot, event: T_MessageEvent):
btn_restart = md.btn_cmd(ulang.get("liteyuki.restart_now"), "reload-liteyuki")
pip.main(["install", "-r", "requirements.txt"])
reply += f"{ulang.get('liteyuki.update_restart', RESTART=btn_restart)}"
await md.send_md(reply, bot)
# await md.send_md(reply, bot)
img_bytes = await md_to_pic(reply)
await UniMessage.send(UniMessage.image(raw=img_bytes))
@on_alconna(

125
src/nonebot_plugins/liteyuki_eventpush.py
View File

@ -1,125 +0,0 @@
import nonebot
from nonebot import on_message, require
from nonebot.plugin import PluginMetadata
from src.utils.base.data import Database, LiteModel
from src.utils.base.ly_typing import T_Bot, T_MessageEvent
from src.utils.message.message import MarkdownMessage as md
require("nonebot_plugin_alconna")
from nonebot_plugin_alconna import on_alconna
from arclet.alconna import Arparma, Alconna, Args, Option, Subcommand
class Node(LiteModel):
TABLE_NAME: str = "node"
bot_id: str = ""
session_type: str = ""
session_id: str = ""
def __str__(self):
return f"{self.bot_id}.{self.session_type}.{self.session_id}"
class Push(LiteModel):
TABLE_NAME: str = "push"
source: Node = Node()
target: Node = Node()
inde: int = 0
pushes_db = Database("data/pushes.ldb")
pushes_db.auto_migrate(Push(), Node())
alc = Alconna(
"lep",
Subcommand(
"add",
Args["source", str],
Args["target", str],
Option("bidirectional", Args["bidirectional", bool])
),
Subcommand(
"rm",
Args["index", int],
),
Subcommand(
"list",
)
)
add_push = on_alconna(alc)
@add_push.handle()
async def _(result: Arparma):
"""bot_id.session_type.session_id"""
if result.subcommands.get("add"):
source = result.subcommands["add"].args.get("source")
target = result.subcommands["add"].args.get("target")
if source and target:
source = source.split(".")
target = target.split(".")
push1 = Push(
source=Node(bot_id=source[0], session_type=source[1], session_id=source[2]),
target=Node(bot_id=target[0], session_type=target[1], session_id=target[2]),
inde=len(pushes_db.where_all(Push(), default=[]))
)
pushes_db.save(push1)
if result.subcommands["add"].args.get("bidirectional"):
push2 = Push(
source=Node(bot_id=target[0], session_type=target[1], session_id=target[2]),
target=Node(bot_id=source[0], session_type=source[1], session_id=source[2]),
inde=len(pushes_db.where_all(Push(), default=[]))
)
pushes_db.save(push2)
await add_push.finish("添加成功")
else:
await add_push.finish("参数缺失")
elif result.subcommands.get("rm"):
index = result.subcommands["rm"].args.get("index")
if index is not None:
try:
pushes_db.delete(Push(), "inde = ?", index)
await add_push.finish("删除成功")
except IndexError:
await add_push.finish("索引错误")
else:
await add_push.finish("参数缺失")
elif result.subcommands.get("list"):
await add_push.finish(
"\n".join([f"{push.inde} {push.source.bot_id}.{push.source.session_type}.{push.source.session_id} -> "
f"{push.target.bot_id}.{push.target.session_type}.{push.target.session_id}" for i, push in
enumerate(pushes_db.where_all(Push(), default=[]))]))
else:
await add_push.finish("参数错误")
@on_message(block=False).handle()
async def _(event: T_MessageEvent, bot: T_Bot):
for push in pushes_db.where_all(Push(), default=[]):
if str(push.source) == f"{bot.self_id}.{event.message_type}.{event.user_id if event.message_type == 'private' else event.group_id}":
bot2 = nonebot.get_bot(push.target.bot_id)
msg_formatted = ""
for line in str(event.message).split("\n"):
msg_formatted += f"**{line.strip()}**\n"
push_message = (
f"> From {event.sender.nickname}@{push.source.session_type}.{push.source.session_id}\n> Bot {bot.self_id}\n\n"
f"{msg_formatted}")
await md.send_md(push_message, bot2, message_type=push.target.session_type,
session_id=push.target.session_id)
return
__author__ = "snowykami"
__plugin_meta__ = PluginMetadata(
name="轻雪事件推送",
description="事件推送插件，支持单向和双向推送，支持跨Bot推送",
usage="",
homepage="https://github.com/snowykami/LiteyukiBot",
extra={
"liteyuki": True,
}
)

52
src/nonebot_plugins/liteyuki_markdowntest.py
View File

@ -1,52 +0,0 @@
from nonebot import on_command, require
from nonebot.adapters.onebot.v11 import MessageSegment
from nonebot.params import CommandArg
from nonebot.permission import SUPERUSER
from nonebot.plugin import PluginMetadata
from src.utils.base.ly_typing import T_Bot, T_MessageEvent, v11
from src.utils.message.message import MarkdownMessage as md, broadcast_to_superusers
from src.utils.message.html_tool import *
md_test = on_command("mdts", permission=SUPERUSER)
btn_test = on_command("btnts", permission=SUPERUSER)
latex_test = on_command("latex", permission=SUPERUSER)
@md_test.handle()
async def _(bot: T_Bot, event: T_MessageEvent, arg: v11.Message = CommandArg()):
await md.send_md(
v11.utils.unescape(str(arg)),
bot,
message_type=event.message_type,
session_id=event.user_id if event.message_type == "private" else event.group_id
)
@btn_test.handle()
async def _(bot: T_Bot, event: T_MessageEvent, arg: v11.Message = CommandArg()):
await md.send_btn(
str(arg),
bot,
message_type=event.message_type,
session_id=event.user_id if event.message_type == "private" else event.group_id
)
@latex_test.handle()
async def _(bot: T_Bot, event: T_MessageEvent, arg: v11.Message = CommandArg()):
latex_text = f"$${v11.utils.unescape(str(arg))}$$"
img = await md_to_pic(latex_text)
await bot.send(event=event, message=MessageSegment.image(img))
__author__ = "snowykami"
__plugin_meta__ = PluginMetadata(
name="轻雪Markdown测试",
description="用于测试Markdown的插件",
usage="",
homepage="https://github.com/snowykami/LiteyukiBot",
extra={
"liteyuki": True,
}
)

15
src/nonebot_plugins/liteyuki_minigame/__init__.py
View File

@ -1,15 +0,0 @@
from nonebot.plugin import PluginMetadata
from .minesweeper import *
__plugin_meta__ = PluginMetadata(
name="轻雪小游戏",
description="内置了一些小游戏",
usage="",
type="application",
homepage="https://github.com/snowykami/LiteyukiBot",
extra={
"liteyuki": True,
"toggleable" : True,
"default_enable" : True,
}
)

168
src/nonebot_plugins/liteyuki_minigame/game.py
View File

@ -1,168 +0,0 @@
import random
from pydantic import BaseModel
from src.utils.message.message import MarkdownMessage as md
class Dot(BaseModel):
row: int
col: int
mask: bool = True
value: int = 0
flagged: bool = False
class Minesweeper:
# 0-8: number of mines around, 9: mine, -1: undefined
NUMS = "⓪①②③④⑤⑥⑦⑧🅑⑩⑪⑫⑬⑭⑮⑯⑰⑱⑲⑳"
MASK = "🅜"
FLAG = "🅕"
MINE = "🅑"
def __init__(self, rows, cols, num_mines, session_type, session_id):
assert rows > 0 and cols > 0 and 0 < num_mines < rows * cols
self.session_type = session_type
self.session_id = session_id
self.rows = rows
self.cols = cols
self.num_mines = num_mines
self.board: list[list[Dot]] = [[Dot(row=i, col=j) for j in range(cols)] for i in range(rows)]
self.is_first = True
def reveal(self, row, col) -> bool:
"""
展开
Args:
row:
col:
Returns:
游戏是否继续
"""
if self.is_first:
# 第一次展开，生成地雷
self.generate_board(self.board[row][col])
self.is_first = False
if self.board[row][col].value == 9:
self.board[row][col].mask = False
return False
if not self.board[row][col].mask:
return True
self.board[row][col].mask = False
if self.board[row][col].value == 0:
self.reveal_neighbors(row, col)
return True
def is_win(self) -> bool:
"""
是否胜利
Returns:
"""
for row in range(self.rows):
for col in range(self.cols):
if self.board[row][col].mask and self.board[row][col].value != 9:
return False
return True
def generate_board(self, first_dot: Dot):
"""
避开第一个点，生成地雷
Args:
first_dot: 第一个点
Returns:
"""
generate_count = 0
while generate_count < self.num_mines:
row = random.randint(0, self.rows - 1)
col = random.randint(0, self.cols - 1)
if self.board[row][col].value == 9 or (row, col) == (first_dot.row, first_dot.col):
continue
self.board[row][col] = Dot(row=row, col=col, mask=True, value=9)
generate_count += 1
for row in range(self.rows):
for col in range(self.cols):
if self.board[row][col].value != 9:
self.board[row][col].value = self.count_adjacent_mines(row, col)
def count_adjacent_mines(self, row, col):
"""
计算周围地雷数量
Args:
row:
col:
Returns:
"""
count = 0
for r in range(max(0, row - 1), min(self.rows, row + 2)):
for c in range(max(0, col - 1), min(self.cols, col + 2)):
if self.board[r][c].value == 9:
count += 1
return count
def reveal_neighbors(self, row, col):
"""
递归展开，使用深度优先搜索
Args:
row:
col:
Returns:
"""
for r in range(max(0, row - 1), min(self.rows, row + 2)):
for c in range(max(0, col - 1), min(self.cols, col + 2)):
if self.board[r][c].mask:
self.board[r][c].mask = False
if self.board[r][c].value == 0:
self.reveal_neighbors(r, c)
def mark(self, row, col) -> bool:
"""
标记
Args:
row:
col:
Returns:
是否标记成功，如果已经展开则无法标记
"""
if self.board[row][col].mask:
self.board[row][col].flagged = not self.board[row][col].flagged
return self.board[row][col].flagged
def board_markdown(self) -> str:
"""
打印地雷板
Returns:
"""
dis = " "
start = "> " if self.cols >= 10 else ""
text = start + self.NUMS[0] + dis*2
# 横向两个雷之间的间隔字符
# 生成横向索引
for i in range(self.cols):
text += f"{self.NUMS[i]}" + dis
text += "\n\n"
for i, row in enumerate(self.board):
text += start + f"{self.NUMS[i]}" + dis*2
for dot in row:
if dot.mask and not dot.flagged:
text += md.btn_cmd(self.MASK, f"minesweeper reveal {dot.row} {dot.col}")
elif dot.flagged:
text += md.btn_cmd(self.FLAG, f"minesweeper mark {dot.row} {dot.col}")
else:
text += self.NUMS[dot.value]
text += dis
text += "\n"
btn_mark = md.btn_cmd("标记", f"minesweeper mark ", enter=False)
btn_end = md.btn_cmd("结束", "minesweeper end", enter=True)
text += f" {btn_mark} {btn_end}"
return text

103
src/nonebot_plugins/liteyuki_minigame/minesweeper.py
View File

@ -1,103 +0,0 @@
from nonebot import require
from src.utils.base.ly_typing import T_Bot, T_MessageEvent
from src.utils.message.message import MarkdownMessage as md
require("nonebot_plugin_alconna")
from .game import Minesweeper
from nonebot_plugin_alconna import Alconna, on_alconna, Subcommand, Args, Arparma
minesweeper = on_alconna(
aliases={"扫雷"},
command=Alconna(
"minesweeper",
Subcommand(
"start",
Args["row", int, 8]["col", int, 8]["mines", int, 10],
alias=["开始"],
),
Subcommand(
"end",
alias=["结束"]
),
Subcommand(
"reveal",
Args["row", int]["col", int],
alias=["展开"]
),
Subcommand(
"mark",
Args["row", int]["col", int],
alias=["标记"]
),
),
)
minesweeper_cache: list[Minesweeper] = []
def get_minesweeper_cache(event: T_MessageEvent) -> Minesweeper | None:
for i in minesweeper_cache:
if i.session_type == event.message_type:
if i.session_id == event.user_id or i.session_id == event.group_id:
return i
return None
@minesweeper.handle()
async def _(event: T_MessageEvent, result: Arparma, bot: T_Bot):
game = get_minesweeper_cache(event)
if result.subcommands.get("start"):
if game:
await minesweeper.finish("当前会话不能同时进行多个扫雷游戏")
else:
try:
new_game = Minesweeper(
rows=result.subcommands["start"].args["row"],
cols=result.subcommands["start"].args["col"],
num_mines=result.subcommands["start"].args["mines"],
session_type=event.message_type,
session_id=event.user_id if event.message_type == "private" else event.group_id,
)
minesweeper_cache.append(new_game)
await minesweeper.send("游戏开始")
await md.send_md(new_game.board_markdown(), bot)
except AssertionError:
await minesweeper.finish("参数错误")
elif result.subcommands.get("end"):
if game:
minesweeper_cache.remove(game)
await minesweeper.finish("游戏结束")
else:
await minesweeper.finish("当前没有扫雷游戏")
elif result.subcommands.get("reveal"):
if not game:
await minesweeper.finish("当前没有扫雷游戏")
else:
row = result.subcommands["reveal"].args["row"]
col = result.subcommands["reveal"].args["col"]
if not (0 <= row < game.rows and 0 <= col < game.cols):
await minesweeper.finish("参数错误")
if not game.reveal(row, col):
minesweeper_cache.remove(game)
await md.send_md(game.board_markdown(), bot)
await minesweeper.finish("游戏结束")
await md.send_md(game.board_markdown(), bot)
if game.is_win():
minesweeper_cache.remove(game)
await minesweeper.finish("游戏胜利")
elif result.subcommands.get("mark"):
if not game:
await minesweeper.finish("当前没有扫雷游戏")
else:
row = result.subcommands["mark"].args["row"]
col = result.subcommands["mark"].args["col"]
if not (0 <= row < game.rows and 0 <= col < game.cols):
await minesweeper.finish("参数错误")
game.board[row][col].flagged = not game.board[row][col].flagged
await md.send_md(game.board_markdown(), bot)
else:
await minesweeper.finish("参数错误")

25
src/nonebot_plugins/liteyuki_pacman/npm.py
View File

@ -14,6 +14,7 @@ from nonebot.permission import SUPERUSER
from nonebot.plugin import Plugin, PluginMetadata
from nonebot.utils import run_sync
from src.utils.base.data_manager import InstalledPlugin
from src.utils.base.language import get_user_lang
from src.utils.base.ly_typing import T_Bot
@ -24,8 +25,10 @@ from src.utils.message.tools import clamp
from .common import *
require("nonebot_plugin_alconna")
require("nonebot_plugin_htmlrender")
from nonebot_plugin_htmlrender import md_to_pic
from nonebot_plugin_alconna import (
on_alconna,
UniMessage, on_alconna,
Alconna,
Args,
Arparma,
@ -292,7 +295,8 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot, npm: Matcher):
reply += f"\n{ulang.get('npm.too_many_results', HIDE_NUM=len(rs) - max_show)}"
else:
reply = ulang.get("npm.search_no_result")
await md.send_md(reply, bot)
img_bytes = await md_to_pic(reply)
await UniMessage.send(UniMessage.image(raw=img_bytes))
elif sc.get("install") and perm_s:
plugin_name: str = result.subcommands["install"].args.get("plugin_name")
@ -320,7 +324,7 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot, npm: Matcher):
info = md.escape(
ulang.get("npm.install_success", NAME=store_plugin.name)
) # markdown转义
await md.send_md(f"{info}\n\n" f"```\n{log}\n```", bot)
await npm.send(f"{info}\n\n" + f"\n{log}\n")
else:
await npm.finish(
ulang.get(
@ -331,12 +335,12 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot, npm: Matcher):
info = ulang.get(
"npm.load_failed", NAME=plugin_name, HOMEPAGE=homepage_btn
).replace("_", r"\\_")
await md.send_md(f"{info}\n\n" f"```\n{log}\n```\n", bot)
await npm.finish(f"{info}\n\n" f"```\n{log}\n```\n")
else:
info = ulang.get(
"npm.install_failed", NAME=plugin_name, HOMEPAGE=homepage_btn
).replace("_", r"\\_")
await md.send_md(f"{info}\n\n" f"```\n{log}\n```", bot)
await npm.send(f"{info}\n\n" f"```\n{log}\n```")
elif sc.get("uninstall") and perm_s:
plugin_name: str = result.subcommands["uninstall"].args.get("plugin_name") # type: ignore
@ -464,7 +468,8 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot, npm: Matcher):
else ulang.get("npm.next_page")
)
reply += f"\n{btn_prev} {page}/{total} {btn_next}"
await md.send_md(reply, bot)
img_bytes = await md_to_pic(reply)
await UniMessage.send(UniMessage.image(raw=img_bytes))
else:
if await SUPERUSER(bot, event):
@ -517,7 +522,8 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot, npm: Matcher):
f"\n\n>page为页数，num为每页显示数量"
f"\n\n>*{md.escape('npm list [page] [num]')}*"
)
await md.send_md(reply, bot)
img_bytes = await md_to_pic(reply)
await UniMessage.send(UniMessage.image(raw=img_bytes))
else:
btn_list = md.btn_cmd(
@ -539,7 +545,8 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot, npm: Matcher):
f"\n\n>page为页数，num为每页显示数量"
f"\n\n>*{md.escape('npm list [page] [num]')}*"
)
await md.send_md(reply, bot)
img_bytes = await md_to_pic(reply)
await UniMessage.send(UniMessage.image(raw=img_bytes))
@on_alconna(
@ -679,7 +686,7 @@ async def _(result: Arparma, matcher: Matcher, event: T_MessageEvent, bot: T_Bot
else mdc.paragraph(ulang.get("npm.homepage"))
),
]
await md.send_md(compile_md(reply), bot)
await matcher.finish(compile_md(reply))
else:
await matcher.finish(ulang.get("npm.plugin_not_found", NAME=plugin_name))
else:

2
src/nonebot_plugins/liteyuki_pacman/rpm.py
View File

@ -181,6 +181,6 @@ async def _(bot: T_Bot, event: T_MessageEvent, result: Arparma, matcher: Matcher
else:
pass
if send_as_md:
await md.send_md(reply, bot)
await matcher.send(reply)
else:
await matcher.finish(reply)

8
src/nonebot_plugins/liteyuki_user/profile_manager.py
View File

@ -12,7 +12,9 @@ from .const import representative_timezones_list
from src.utils import event as event_utils
require("nonebot_plugin_alconna")
require("nonebot_plugin_htmlrender")
from nonebot_plugin_alconna import Alconna, Args, Arparma, Subcommand, on_alconna
from nonebot_plugin_htmlrender import md_to_pic
profile_alc = on_alconna(
Alconna(
@ -65,7 +67,8 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
# 未输入值，尝试呼出菜单
menu = get_profile_menu(result.args["key"], ulang)
if menu:
await md.send_md(menu, bot)
img_bytes = await md_to_pic(menu)
await profile_alc.finish(menu)
else:
await profile_alc.finish(ulang.get("user.profile.input_value", ATTR=ulang.get(f"user.profile.{result.args['key']}")))
@ -97,7 +100,8 @@ async def _(result: Arparma, event: T_MessageEvent, bot: T_Bot):
reply += (f"\n**{key_text}** **{val}**\n"
f"\n> {ulang.get(f'user.profile.{key}.desc')}"
f"\n> {btn_set} \n\n***\n")
await md.send_md(reply, bot)
img_bytes = await md_to_pic(reply)
await profile_alc.finish(reply)
def get_profile_menu(key: str, ulang: Language) -> Optional[str]:

12
src/utils/message/message.py
View File

@ -19,9 +19,6 @@ from nonebot_plugin_htmlrender import md_to_pic
config = load_from_yaml("config.yml")
can_send_markdown = {} # 用于存储机器人是否支持发送markdown消息，id->bool
async def broadcast_to_superusers(message: str | T_Message, markdown: bool = False):
"""广播消息给超级用户"""
@ -57,14 +54,7 @@ class MarkdownMessage:
width=540,
device_scale_factor=4
)
if md_image_bytes is None:
data = await bot.send_msg(
message_type=message_type,
group_id=session_id,
user_id=session_id,
message=markdown,
)
return data
print(md_image_bytes)
data = await bot.send_msg(
message_type=message_type,
group_id=session_id,