mirror of
https://github.com/LiteyukiStudio/LiteyukiBot.git
synced 2025-09-12 20:36:22 +00:00
✨ 分离magicoca和croterline
This commit is contained in:
@ -1,89 +1,89 @@
|
||||
import os
|
||||
import aiofiles # type: ignore
|
||||
import nonebot
|
||||
from nonebot import require
|
||||
|
||||
# require("nonebot_plugin_htmlrender")
|
||||
|
||||
from nonebot_plugin_htmlrender import ( # type: ignore
|
||||
template_to_html,
|
||||
template_to_pic,
|
||||
md_to_pic
|
||||
) # type: ignore
|
||||
|
||||
|
||||
async def template2html(
|
||||
template: str,
|
||||
templates: dict,
|
||||
) -> str:
|
||||
"""
|
||||
Args:
|
||||
template: str: 模板文件
|
||||
**templates: dict: 模板参数
|
||||
Returns:
|
||||
HTML 正文
|
||||
"""
|
||||
template_path = os.path.dirname(template)
|
||||
template_name = os.path.basename(template)
|
||||
return await template_to_html(template_path, template_name, **templates)
|
||||
|
||||
|
||||
async def template2image(
|
||||
template: str,
|
||||
templates: dict,
|
||||
pages=None,
|
||||
wait: int = 0,
|
||||
scale_factor: float = 1,
|
||||
debug: bool = False,
|
||||
) -> bytes:
|
||||
"""
|
||||
template -> html -> image
|
||||
Args:
|
||||
debug: 输入渲染好的 html
|
||||
wait: 等待时间，单位秒
|
||||
pages: 页面参数
|
||||
template: str: 模板文件
|
||||
templates: dict: 模板参数
|
||||
scale_factor: 缩放因子，越高越清晰
|
||||
Returns:
|
||||
图片二进制数据
|
||||
"""
|
||||
|
||||
###
|
||||
if pages is None:
|
||||
pages = {
|
||||
"viewport": {
|
||||
"width" : 1080,
|
||||
"height": 10
|
||||
},
|
||||
}
|
||||
|
||||
template_path = os.path.dirname(template)
|
||||
template_name = os.path.basename(template)
|
||||
|
||||
if debug:
|
||||
# 重载资源
|
||||
raw_html = await template_to_html(
|
||||
template_name=template_name,
|
||||
template_path=template_path,
|
||||
**templates,
|
||||
)
|
||||
random_file_name = f"debug.html"
|
||||
async with aiofiles.open(
|
||||
os.path.join(template_path, random_file_name), "w", encoding="utf-8"
|
||||
) as f:
|
||||
await f.write(raw_html)
|
||||
nonebot.logger.info("Debug HTML: %s" % f"{random_file_name}")
|
||||
return await template_to_pic(
|
||||
template_name=template_name,
|
||||
template_path=template_path,
|
||||
templates=templates,
|
||||
wait=wait,
|
||||
|
||||
###
|
||||
pages=pages,
|
||||
device_scale_factor=scale_factor
|
||||
###
|
||||
)
|
||||
|
||||
|
||||
import os
|
||||
import aiofiles # type: ignore
|
||||
import nonebot
|
||||
from nonebot import require
|
||||
|
||||
# require("nonebot_plugin_htmlrender")
|
||||
|
||||
from nonebot_plugin_htmlrender import ( # type: ignore
|
||||
template_to_html,
|
||||
template_to_pic,
|
||||
md_to_pic
|
||||
) # type: ignore
|
||||
|
||||
|
||||
async def template2html(
|
||||
template: str,
|
||||
templates: dict,
|
||||
) -> str:
|
||||
"""
|
||||
Args:
|
||||
template: str: 模板文件
|
||||
**templates: dict: 模板参数
|
||||
Returns:
|
||||
HTML 正文
|
||||
"""
|
||||
template_path = os.path.dirname(template)
|
||||
template_name = os.path.basename(template)
|
||||
return await template_to_html(template_path, template_name, **templates)
|
||||
|
||||
|
||||
async def template2image(
|
||||
template: str,
|
||||
templates: dict,
|
||||
pages=None,
|
||||
wait: int = 0,
|
||||
scale_factor: float = 1,
|
||||
debug: bool = False,
|
||||
) -> bytes:
|
||||
"""
|
||||
template -> html -> image
|
||||
Args:
|
||||
debug: 输入渲染好的 html
|
||||
wait: 等待时间，单位秒
|
||||
pages: 页面参数
|
||||
template: str: 模板文件
|
||||
templates: dict: 模板参数
|
||||
scale_factor: 缩放因子，越高越清晰
|
||||
Returns:
|
||||
图片二进制数据
|
||||
"""
|
||||
|
||||
###
|
||||
if pages is None:
|
||||
pages = {
|
||||
"viewport": {
|
||||
"width" : 1080,
|
||||
"height": 10
|
||||
},
|
||||
}
|
||||
|
||||
template_path = os.path.dirname(template)
|
||||
template_name = os.path.basename(template)
|
||||
|
||||
if debug:
|
||||
# 重载资源
|
||||
raw_html = await template_to_html(
|
||||
template_name=template_name,
|
||||
template_path=template_path,
|
||||
**templates,
|
||||
)
|
||||
random_file_name = f"debug.html"
|
||||
async with aiofiles.open(
|
||||
os.path.join(template_path, random_file_name), "w", encoding="utf-8"
|
||||
) as f:
|
||||
await f.write(raw_html)
|
||||
nonebot.logger.info("Debug HTML: %s" % f"{random_file_name}")
|
||||
return await template_to_pic(
|
||||
template_name=template_name,
|
||||
template_path=template_path,
|
||||
templates=templates,
|
||||
wait=wait,
|
||||
|
||||
###
|
||||
pages=pages,
|
||||
device_scale_factor=scale_factor
|
||||
###
|
||||
)
|
||||
|
||||
|
||||
|
||||
@ -1,209 +1,209 @@
|
||||
import base64
|
||||
from io import BytesIO
|
||||
from urllib.parse import quote
|
||||
|
||||
import aiohttp
|
||||
from PIL import Image
|
||||
|
||||
from ..base.config import get_config
|
||||
from ..base.data import LiteModel
|
||||
from ..base.ly_typing import T_Bot
|
||||
|
||||
|
||||
def escape_md(text: str) -> str:
|
||||
"""
|
||||
转义Markdown特殊字符
|
||||
Args:
|
||||
text: str: 文本
|
||||
|
||||
Returns:
|
||||
str: 转义后文本
|
||||
"""
|
||||
spacial_chars = r"\`*_{}[]()#+-.!"
|
||||
for char in spacial_chars:
|
||||
text = text.replace(char, "\\\\" + char)
|
||||
return text.replace("\n", r"\n").replace('"', r'\\\"')
|
||||
|
||||
|
||||
def escape_decorator(func):
|
||||
def wrapper(text: str):
|
||||
return func(escape_md(text))
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
def compile_md(comps: list[str]) -> str:
|
||||
"""
|
||||
合成Markdown文本
|
||||
Args:
|
||||
comps: list[str]: 组件列表
|
||||
|
||||
Returns:
|
||||
str: 编译后文本
|
||||
"""
|
||||
return "".join(comps)
|
||||
|
||||
|
||||
class MarkdownComponent:
|
||||
@staticmethod
|
||||
def heading(text: str, level: int = 1) -> str:
|
||||
"""标题"""
|
||||
assert 1 <= level <= 6, "标题级别应在 1-6 之间"
|
||||
return f"{'#' * level} {text}\n"
|
||||
|
||||
@staticmethod
|
||||
def bold(text: str) -> str:
|
||||
"""粗体"""
|
||||
return f"**{text}**"
|
||||
|
||||
@staticmethod
|
||||
def italic(text: str) -> str:
|
||||
"""斜体"""
|
||||
return f"*{text}*"
|
||||
|
||||
@staticmethod
|
||||
def strike(text: str) -> str:
|
||||
"""删除线"""
|
||||
return f"~~{text}~~"
|
||||
|
||||
@staticmethod
|
||||
def code(text: str) -> str:
|
||||
"""行内代码"""
|
||||
return f"`{text}`"
|
||||
|
||||
@staticmethod
|
||||
def code_block(text: str, language: str = "") -> str:
|
||||
"""代码块"""
|
||||
return f"```{language}\n{text}\n```\n"
|
||||
|
||||
@staticmethod
|
||||
def quote(text: str) -> str:
|
||||
"""引用"""
|
||||
return f"> {text}\n\n"
|
||||
|
||||
@staticmethod
|
||||
def link(text: str, url: str, symbol: bool = True) -> str:
|
||||
"""
|
||||
链接
|
||||
|
||||
Args:
|
||||
text: 链接文本
|
||||
url: 链接地址
|
||||
symbol: 是否显示链接图标, mqqapi请使用False
|
||||
"""
|
||||
return f"[{'🔗' if symbol else ''}{text}]({url})"
|
||||
|
||||
@staticmethod
|
||||
def image(url: str, *, size: tuple[int, int]) -> str:
|
||||
"""
|
||||
图片，本地图片不建议直接使用
|
||||
Args:
|
||||
url: 图片链接
|
||||
size: 图片大小
|
||||
|
||||
Returns:
|
||||
markdown格式的图片
|
||||
"""
|
||||
return f"![image #{size[0]}px #{size[1]}px]({url})"
|
||||
|
||||
@staticmethod
|
||||
async def auto_image(image: str | bytes, bot: T_Bot) -> str:
|
||||
"""
|
||||
自动获取图片大小
|
||||
Args:
|
||||
image: 本地图片路径 | 图片url http/file | 图片bytes
|
||||
bot: bot对象，用于上传图片到图床
|
||||
|
||||
Returns:
|
||||
markdown格式的图片
|
||||
"""
|
||||
if isinstance(image, bytes):
|
||||
# 传入为二进制图片
|
||||
image_obj = Image.open(BytesIO(image))
|
||||
base64_string = base64.b64encode(image_obj.tobytes()).decode("utf-8")
|
||||
url = await bot.call_api("upload_image", file=f"base64://{base64_string}")
|
||||
size = image_obj.size
|
||||
elif isinstance(image, str):
|
||||
# 传入链接或本地路径
|
||||
if image.startswith("http"):
|
||||
# 网络请求
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(image) as resp:
|
||||
image_data = await resp.read()
|
||||
url = image
|
||||
size = Image.open(BytesIO(image_data)).size
|
||||
|
||||
else:
|
||||
# 本地路径/file://
|
||||
image_obj = Image.open(image.replace("file://", ""))
|
||||
base64_string = base64.b64encode(image_obj.tobytes()).decode("utf-8")
|
||||
url = await bot.call_api("upload_image", file=f"base64://{base64_string}")
|
||||
size = image_obj.size
|
||||
else:
|
||||
raise ValueError("图片类型错误")
|
||||
|
||||
return MarkdownComponent.image(url, size=size)
|
||||
|
||||
@staticmethod
|
||||
def table(data: list[list[any]]) -> str:
|
||||
"""
|
||||
表格
|
||||
Args:
|
||||
data: 表格数据，二维列表
|
||||
Returns:
|
||||
markdown格式的表格
|
||||
"""
|
||||
# 表头
|
||||
table = "|".join(map(str, data[0])) + "\n"
|
||||
table += "|".join([":-:" for _ in range(len(data[0]))]) + "\n"
|
||||
# 表内容
|
||||
for row in data[1:]:
|
||||
table += "|".join(map(str, row)) + "\n"
|
||||
return table
|
||||
|
||||
@staticmethod
|
||||
def paragraph(text: str) -> str:
|
||||
"""
|
||||
段落
|
||||
Args:
|
||||
text: 段落内容
|
||||
Returns:
|
||||
markdown格式的段落
|
||||
"""
|
||||
return f"{text}\n"
|
||||
|
||||
|
||||
class Mqqapi:
|
||||
@staticmethod
|
||||
@escape_decorator
|
||||
def cmd(text: str, cmd: str, enter: bool = True, reply: bool = False, use_cmd_start: bool = True) -> str:
|
||||
"""
|
||||
生成点击回调文本
|
||||
Args:
|
||||
text: 显示内容
|
||||
cmd: 命令
|
||||
enter: 是否自动发送
|
||||
reply: 是否回复
|
||||
use_cmd_start: 是否使用配置的命令前缀
|
||||
|
||||
Returns:
|
||||
[text](mqqapi://) markdown格式的可点击回调文本，类似于链接
|
||||
"""
|
||||
|
||||
if use_cmd_start:
|
||||
command_start = get_config("command_start", [])
|
||||
if command_start:
|
||||
# 若命令前缀不为空，则使用配置的第一个命令前缀
|
||||
cmd = f"{command_start[0]}{cmd}"
|
||||
return f"[{text}](mqqapi://aio/inlinecmd?command={quote(cmd)}&reply={str(reply).lower()}&enter={str(enter).lower()})"
|
||||
|
||||
|
||||
class RenderData(LiteModel):
|
||||
label: str
|
||||
visited_label: str
|
||||
style: int
|
||||
|
||||
|
||||
class Button(LiteModel):
|
||||
id: int
|
||||
render_data: RenderData
|
||||
import base64
|
||||
from io import BytesIO
|
||||
from urllib.parse import quote
|
||||
|
||||
import aiohttp
|
||||
from PIL import Image
|
||||
|
||||
from ..base.config import get_config
|
||||
from ..base.data import LiteModel
|
||||
from ..base.ly_typing import T_Bot
|
||||
|
||||
|
||||
def escape_md(text: str) -> str:
|
||||
"""
|
||||
转义Markdown特殊字符
|
||||
Args:
|
||||
text: str: 文本
|
||||
|
||||
Returns:
|
||||
str: 转义后文本
|
||||
"""
|
||||
spacial_chars = r"\`*_{}[]()#+-.!"
|
||||
for char in spacial_chars:
|
||||
text = text.replace(char, "\\\\" + char)
|
||||
return text.replace("\n", r"\n").replace('"', r'\\\"')
|
||||
|
||||
|
||||
def escape_decorator(func):
|
||||
def wrapper(text: str):
|
||||
return func(escape_md(text))
|
||||
|
||||
return wrapper
|
||||
|
||||
|
||||
def compile_md(comps: list[str]) -> str:
|
||||
"""
|
||||
合成Markdown文本
|
||||
Args:
|
||||
comps: list[str]: 组件列表
|
||||
|
||||
Returns:
|
||||
str: 编译后文本
|
||||
"""
|
||||
return "".join(comps)
|
||||
|
||||
|
||||
class MarkdownComponent:
|
||||
@staticmethod
|
||||
def heading(text: str, level: int = 1) -> str:
|
||||
"""标题"""
|
||||
assert 1 <= level <= 6, "标题级别应在 1-6 之间"
|
||||
return f"{'#' * level} {text}\n"
|
||||
|
||||
@staticmethod
|
||||
def bold(text: str) -> str:
|
||||
"""粗体"""
|
||||
return f"**{text}**"
|
||||
|
||||
@staticmethod
|
||||
def italic(text: str) -> str:
|
||||
"""斜体"""
|
||||
return f"*{text}*"
|
||||
|
||||
@staticmethod
|
||||
def strike(text: str) -> str:
|
||||
"""删除线"""
|
||||
return f"~~{text}~~"
|
||||
|
||||
@staticmethod
|
||||
def code(text: str) -> str:
|
||||
"""行内代码"""
|
||||
return f"`{text}`"
|
||||
|
||||
@staticmethod
|
||||
def code_block(text: str, language: str = "") -> str:
|
||||
"""代码块"""
|
||||
return f"```{language}\n{text}\n```\n"
|
||||
|
||||
@staticmethod
|
||||
def quote(text: str) -> str:
|
||||
"""引用"""
|
||||
return f"> {text}\n\n"
|
||||
|
||||
@staticmethod
|
||||
def link(text: str, url: str, symbol: bool = True) -> str:
|
||||
"""
|
||||
链接
|
||||
|
||||
Args:
|
||||
text: 链接文本
|
||||
url: 链接地址
|
||||
symbol: 是否显示链接图标, mqqapi请使用False
|
||||
"""
|
||||
return f"[{'🔗' if symbol else ''}{text}]({url})"
|
||||
|
||||
@staticmethod
|
||||
def image(url: str, *, size: tuple[int, int]) -> str:
|
||||
"""
|
||||
图片，本地图片不建议直接使用
|
||||
Args:
|
||||
url: 图片链接
|
||||
size: 图片大小
|
||||
|
||||
Returns:
|
||||
markdown格式的图片
|
||||
"""
|
||||
return f"![image #{size[0]}px #{size[1]}px]({url})"
|
||||
|
||||
@staticmethod
|
||||
async def auto_image(image: str | bytes, bot: T_Bot) -> str:
|
||||
"""
|
||||
自动获取图片大小
|
||||
Args:
|
||||
image: 本地图片路径 | 图片url http/file | 图片bytes
|
||||
bot: bot对象，用于上传图片到图床
|
||||
|
||||
Returns:
|
||||
markdown格式的图片
|
||||
"""
|
||||
if isinstance(image, bytes):
|
||||
# 传入为二进制图片
|
||||
image_obj = Image.open(BytesIO(image))
|
||||
base64_string = base64.b64encode(image_obj.tobytes()).decode("utf-8")
|
||||
url = await bot.call_api("upload_image", file=f"base64://{base64_string}")
|
||||
size = image_obj.size
|
||||
elif isinstance(image, str):
|
||||
# 传入链接或本地路径
|
||||
if image.startswith("http"):
|
||||
# 网络请求
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(image) as resp:
|
||||
image_data = await resp.read()
|
||||
url = image
|
||||
size = Image.open(BytesIO(image_data)).size
|
||||
|
||||
else:
|
||||
# 本地路径/file://
|
||||
image_obj = Image.open(image.replace("file://", ""))
|
||||
base64_string = base64.b64encode(image_obj.tobytes()).decode("utf-8")
|
||||
url = await bot.call_api("upload_image", file=f"base64://{base64_string}")
|
||||
size = image_obj.size
|
||||
else:
|
||||
raise ValueError("图片类型错误")
|
||||
|
||||
return MarkdownComponent.image(url, size=size)
|
||||
|
||||
@staticmethod
|
||||
def table(data: list[list[any]]) -> str:
|
||||
"""
|
||||
表格
|
||||
Args:
|
||||
data: 表格数据，二维列表
|
||||
Returns:
|
||||
markdown格式的表格
|
||||
"""
|
||||
# 表头
|
||||
table = "|".join(map(str, data[0])) + "\n"
|
||||
table += "|".join([":-:" for _ in range(len(data[0]))]) + "\n"
|
||||
# 表内容
|
||||
for row in data[1:]:
|
||||
table += "|".join(map(str, row)) + "\n"
|
||||
return table
|
||||
|
||||
@staticmethod
|
||||
def paragraph(text: str) -> str:
|
||||
"""
|
||||
段落
|
||||
Args:
|
||||
text: 段落内容
|
||||
Returns:
|
||||
markdown格式的段落
|
||||
"""
|
||||
return f"{text}\n"
|
||||
|
||||
|
||||
class Mqqapi:
|
||||
@staticmethod
|
||||
@escape_decorator
|
||||
def cmd(text: str, cmd: str, enter: bool = True, reply: bool = False, use_cmd_start: bool = True) -> str:
|
||||
"""
|
||||
生成点击回调文本
|
||||
Args:
|
||||
text: 显示内容
|
||||
cmd: 命令
|
||||
enter: 是否自动发送
|
||||
reply: 是否回复
|
||||
use_cmd_start: 是否使用配置的命令前缀
|
||||
|
||||
Returns:
|
||||
[text](mqqapi://) markdown格式的可点击回调文本，类似于链接
|
||||
"""
|
||||
|
||||
if use_cmd_start:
|
||||
command_start = get_config("command_start", [])
|
||||
if command_start:
|
||||
# 若命令前缀不为空，则使用配置的第一个命令前缀
|
||||
cmd = f"{command_start[0]}{cmd}"
|
||||
return f"[{text}](mqqapi://aio/inlinecmd?command={quote(cmd)}&reply={str(reply).lower()}&enter={str(enter).lower()})"
|
||||
|
||||
|
||||
class RenderData(LiteModel):
|
||||
label: str
|
||||
visited_label: str
|
||||
style: int
|
||||
|
||||
|
||||
class Button(LiteModel):
|
||||
id: int
|
||||
render_data: RenderData
|
||||
|
||||
@ -1,202 +1,202 @@
|
||||
import base64
|
||||
import io
|
||||
from typing import Any
|
||||
from urllib.parse import quote
|
||||
|
||||
import aiofiles
|
||||
import aiohttp
|
||||
import nonebot
|
||||
from PIL import Image
|
||||
from nonebot.adapters.onebot import v11
|
||||
|
||||
from .html_tool import md_to_pic
|
||||
from .. import load_from_yaml
|
||||
from ..base.ly_typing import T_Bot, T_Message, T_MessageEvent
|
||||
|
||||
config = load_from_yaml("config.yml")
|
||||
|
||||
|
||||
async def broadcast_to_superusers(message: str | T_Message, markdown: bool = False):
|
||||
"""广播消息给超级用户"""
|
||||
for bot in nonebot.get_bots().values():
|
||||
for user_id in config.get("superusers", []):
|
||||
if markdown:
|
||||
await MarkdownMessage.send_md(message, bot, message_type="private", session_id=user_id)
|
||||
else:
|
||||
await bot.send_private_msg(user_id=user_id, message=message)
|
||||
|
||||
|
||||
class MarkdownMessage:
|
||||
@staticmethod
|
||||
async def send_md(
|
||||
markdown: str,
|
||||
bot: T_Bot, *,
|
||||
message_type: str = None,
|
||||
session_id: str | int = None
|
||||
) -> dict[str, Any] | None:
|
||||
"""
|
||||
发送Markdown消息，支持自动转为图片发送
|
||||
Args:
|
||||
markdown:
|
||||
bot:
|
||||
message_type:
|
||||
session_id:
|
||||
Returns:
|
||||
|
||||
"""
|
||||
plain_markdown = markdown.replace("[🔗", "[")
|
||||
md_image_bytes = await md_to_pic(
|
||||
md=plain_markdown,
|
||||
width=540,
|
||||
device_scale_factor=4
|
||||
)
|
||||
print(md_image_bytes)
|
||||
data = await bot.send_msg(
|
||||
message_type=message_type,
|
||||
group_id=session_id,
|
||||
user_id=session_id,
|
||||
message=v11.MessageSegment.image(md_image_bytes),
|
||||
)
|
||||
return data
|
||||
|
||||
@staticmethod
|
||||
async def send_image(
|
||||
image: bytes | str,
|
||||
bot: T_Bot, *,
|
||||
message_type: str = None,
|
||||
session_id: str | int = None,
|
||||
event: T_MessageEvent = None,
|
||||
**kwargs
|
||||
) -> dict:
|
||||
"""
|
||||
发送单张装逼大图
|
||||
Args:
|
||||
image: 图片字节流或图片本地路径，链接请使用Markdown.image_async方法获取后通过send_md发送
|
||||
bot: bot instance
|
||||
message_type: message message_type
|
||||
session_id: session id
|
||||
event: event
|
||||
kwargs: other arguments
|
||||
Returns:
|
||||
dict: response data
|
||||
"""
|
||||
if isinstance(image, str):
|
||||
async with aiofiles.open(image, "rb") as f:
|
||||
image = await f.read()
|
||||
method = 2
|
||||
if method == 2:
|
||||
base64_string = base64.b64encode(image).decode("utf-8")
|
||||
data = await bot.call_api("upload_image", file=f"base64://{base64_string}")
|
||||
await MarkdownMessage.send_md(MarkdownMessage.image(data, Image.open(io.BytesIO(image)).size), bot,
|
||||
message_type=message_type,
|
||||
session_id=session_id)
|
||||
|
||||
# 其他实现端方案
|
||||
else:
|
||||
image_message_id = (await bot.send_private_msg(
|
||||
user_id=bot.self_id,
|
||||
message=[
|
||||
v11.MessageSegment.image(file=image)
|
||||
]
|
||||
))["message_id"]
|
||||
image_url = (await bot.get_msg(message_id=image_message_id))["message"][0]["data"]["url"]
|
||||
image_size = Image.open(io.BytesIO(image)).size
|
||||
image_md = MarkdownMessage.image(image_url, image_size)
|
||||
return await MarkdownMessage.send_md(image_md, bot, message_type=message_type, session_id=session_id)
|
||||
|
||||
if data is None:
|
||||
data = await bot.send_msg(
|
||||
message_type=message_type,
|
||||
group_id=session_id,
|
||||
user_id=session_id,
|
||||
message=v11.MessageSegment.image(image),
|
||||
**kwargs
|
||||
)
|
||||
return data
|
||||
|
||||
@staticmethod
|
||||
async def get_image_url(image: bytes | str, bot: T_Bot) -> str:
|
||||
"""把图片上传到图床，返回链接
|
||||
Args:
|
||||
bot: 发送的bot
|
||||
image: 图片字节流或图片本地路径
|
||||
Returns:
|
||||
"""
|
||||
# 等林文轩修好Lagrange.OneBot再说
|
||||
|
||||
@staticmethod
|
||||
def btn_cmd(name: str, cmd: str, reply: bool = False, enter: bool = True) -> str:
|
||||
"""生成点击回调按钮
|
||||
Args:
|
||||
name: 按钮显示内容
|
||||
cmd: 发送的命令，已在函数内url编码，不需要再次编码
|
||||
reply: 是否以回复的方式发送消息
|
||||
enter: 自动发送消息则为True，否则填充到输入框
|
||||
|
||||
Returns:
|
||||
markdown格式的可点击回调按钮
|
||||
|
||||
"""
|
||||
if "" not in config.get("command_start", ["/"]) and config.get("alconna_use_command_start", False):
|
||||
cmd = f"{config['command_start'][0]}{cmd}"
|
||||
return f"[{name}](mqqapi://aio/inlinecmd?command={quote(cmd)}&reply={str(reply).lower()}&enter={str(enter).lower()})"
|
||||
|
||||
@staticmethod
|
||||
def btn_link(name: str, url: str) -> str:
|
||||
"""生成点击链接按钮
|
||||
Args:
|
||||
name: 链接显示内容
|
||||
url: 链接地址
|
||||
|
||||
Returns:
|
||||
markdown格式的链接
|
||||
|
||||
"""
|
||||
return f"[🔗{name}]({url})"
|
||||
|
||||
@staticmethod
|
||||
def image(url: str, size: tuple[int, int]) -> str:
|
||||
"""构建图片链接
|
||||
Args:
|
||||
size:
|
||||
url: 图片链接
|
||||
|
||||
Returns:
|
||||
markdown格式的图片
|
||||
|
||||
"""
|
||||
return f"![image #{size[0]}px #{size[1]}px]({url})"
|
||||
|
||||
@staticmethod
|
||||
async def image_async(url: str) -> str:
|
||||
"""获取图片，自动请求获取大小，异步
|
||||
Args:
|
||||
url: 图片链接
|
||||
|
||||
Returns:
|
||||
图片Markdown语法: 
|
||||
|
||||
"""
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as resp:
|
||||
image = Image.open(io.BytesIO(await resp.read()))
|
||||
return MarkdownMessage.image(url, image.size)
|
||||
except Exception as e:
|
||||
nonebot.logger.error(f"get image error: {e}")
|
||||
return "[Image Error]"
|
||||
|
||||
@staticmethod
|
||||
def escape(text: str) -> str:
|
||||
"""转义特殊字符
|
||||
Args:
|
||||
text: 需要转义的文本，请勿直接把整个markdown文本传入，否则会转义掉所有字符
|
||||
|
||||
Returns:
|
||||
转义后的文本
|
||||
|
||||
"""
|
||||
chars = "*[]()~_`>#+=|{}.!"
|
||||
for char in chars:
|
||||
text = text.replace(char, f"\\\\{char}")
|
||||
return text
|
||||
import base64
|
||||
import io
|
||||
from typing import Any
|
||||
from urllib.parse import quote
|
||||
|
||||
import aiofiles
|
||||
import aiohttp
|
||||
import nonebot
|
||||
from PIL import Image
|
||||
from nonebot.adapters.onebot import v11
|
||||
|
||||
from .html_tool import md_to_pic
|
||||
from .. import load_from_yaml
|
||||
from ..base.ly_typing import T_Bot, T_Message, T_MessageEvent
|
||||
|
||||
config = load_from_yaml("config.yml")
|
||||
|
||||
|
||||
async def broadcast_to_superusers(message: str | T_Message, markdown: bool = False):
|
||||
"""广播消息给超级用户"""
|
||||
for bot in nonebot.get_bots().values():
|
||||
for user_id in config.get("superusers", []):
|
||||
if markdown:
|
||||
await MarkdownMessage.send_md(message, bot, message_type="private", session_id=user_id)
|
||||
else:
|
||||
await bot.send_private_msg(user_id=user_id, message=message)
|
||||
|
||||
|
||||
class MarkdownMessage:
|
||||
@staticmethod
|
||||
async def send_md(
|
||||
markdown: str,
|
||||
bot: T_Bot, *,
|
||||
message_type: str = None,
|
||||
session_id: str | int = None
|
||||
) -> dict[str, Any] | None:
|
||||
"""
|
||||
发送Markdown消息，支持自动转为图片发送
|
||||
Args:
|
||||
markdown:
|
||||
bot:
|
||||
message_type:
|
||||
session_id:
|
||||
Returns:
|
||||
|
||||
"""
|
||||
plain_markdown = markdown.replace("[🔗", "[")
|
||||
md_image_bytes = await md_to_pic(
|
||||
md=plain_markdown,
|
||||
width=540,
|
||||
device_scale_factor=4
|
||||
)
|
||||
print(md_image_bytes)
|
||||
data = await bot.send_msg(
|
||||
message_type=message_type,
|
||||
group_id=session_id,
|
||||
user_id=session_id,
|
||||
message=v11.MessageSegment.image(md_image_bytes),
|
||||
)
|
||||
return data
|
||||
|
||||
@staticmethod
|
||||
async def send_image(
|
||||
image: bytes | str,
|
||||
bot: T_Bot, *,
|
||||
message_type: str = None,
|
||||
session_id: str | int = None,
|
||||
event: T_MessageEvent = None,
|
||||
**kwargs
|
||||
) -> dict:
|
||||
"""
|
||||
发送单张装逼大图
|
||||
Args:
|
||||
image: 图片字节流或图片本地路径，链接请使用Markdown.image_async方法获取后通过send_md发送
|
||||
bot: bot instance
|
||||
message_type: message message_type
|
||||
session_id: session id
|
||||
event: event
|
||||
kwargs: other arguments
|
||||
Returns:
|
||||
dict: response data
|
||||
"""
|
||||
if isinstance(image, str):
|
||||
async with aiofiles.open(image, "rb") as f:
|
||||
image = await f.read()
|
||||
method = 2
|
||||
if method == 2:
|
||||
base64_string = base64.b64encode(image).decode("utf-8")
|
||||
data = await bot.call_api("upload_image", file=f"base64://{base64_string}")
|
||||
await MarkdownMessage.send_md(MarkdownMessage.image(data, Image.open(io.BytesIO(image)).size), bot,
|
||||
message_type=message_type,
|
||||
session_id=session_id)
|
||||
|
||||
# 其他实现端方案
|
||||
else:
|
||||
image_message_id = (await bot.send_private_msg(
|
||||
user_id=bot.self_id,
|
||||
message=[
|
||||
v11.MessageSegment.image(file=image)
|
||||
]
|
||||
))["message_id"]
|
||||
image_url = (await bot.get_msg(message_id=image_message_id))["message"][0]["data"]["url"]
|
||||
image_size = Image.open(io.BytesIO(image)).size
|
||||
image_md = MarkdownMessage.image(image_url, image_size)
|
||||
return await MarkdownMessage.send_md(image_md, bot, message_type=message_type, session_id=session_id)
|
||||
|
||||
if data is None:
|
||||
data = await bot.send_msg(
|
||||
message_type=message_type,
|
||||
group_id=session_id,
|
||||
user_id=session_id,
|
||||
message=v11.MessageSegment.image(image),
|
||||
**kwargs
|
||||
)
|
||||
return data
|
||||
|
||||
@staticmethod
|
||||
async def get_image_url(image: bytes | str, bot: T_Bot) -> str:
|
||||
"""把图片上传到图床，返回链接
|
||||
Args:
|
||||
bot: 发送的bot
|
||||
image: 图片字节流或图片本地路径
|
||||
Returns:
|
||||
"""
|
||||
# 等林文轩修好Lagrange.OneBot再说
|
||||
|
||||
@staticmethod
|
||||
def btn_cmd(name: str, cmd: str, reply: bool = False, enter: bool = True) -> str:
|
||||
"""生成点击回调按钮
|
||||
Args:
|
||||
name: 按钮显示内容
|
||||
cmd: 发送的命令，已在函数内url编码，不需要再次编码
|
||||
reply: 是否以回复的方式发送消息
|
||||
enter: 自动发送消息则为True，否则填充到输入框
|
||||
|
||||
Returns:
|
||||
markdown格式的可点击回调按钮
|
||||
|
||||
"""
|
||||
if "" not in config.get("command_start", ["/"]) and config.get("alconna_use_command_start", False):
|
||||
cmd = f"{config['command_start'][0]}{cmd}"
|
||||
return f"[{name}](mqqapi://aio/inlinecmd?command={quote(cmd)}&reply={str(reply).lower()}&enter={str(enter).lower()})"
|
||||
|
||||
@staticmethod
|
||||
def btn_link(name: str, url: str) -> str:
|
||||
"""生成点击链接按钮
|
||||
Args:
|
||||
name: 链接显示内容
|
||||
url: 链接地址
|
||||
|
||||
Returns:
|
||||
markdown格式的链接
|
||||
|
||||
"""
|
||||
return f"[🔗{name}]({url})"
|
||||
|
||||
@staticmethod
|
||||
def image(url: str, size: tuple[int, int]) -> str:
|
||||
"""构建图片链接
|
||||
Args:
|
||||
size:
|
||||
url: 图片链接
|
||||
|
||||
Returns:
|
||||
markdown格式的图片
|
||||
|
||||
"""
|
||||
return f"![image #{size[0]}px #{size[1]}px]({url})"
|
||||
|
||||
@staticmethod
|
||||
async def image_async(url: str) -> str:
|
||||
"""获取图片，自动请求获取大小，异步
|
||||
Args:
|
||||
url: 图片链接
|
||||
|
||||
Returns:
|
||||
图片Markdown语法: 
|
||||
|
||||
"""
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as resp:
|
||||
image = Image.open(io.BytesIO(await resp.read()))
|
||||
return MarkdownMessage.image(url, image.size)
|
||||
except Exception as e:
|
||||
nonebot.logger.error(f"get image error: {e}")
|
||||
return "[Image Error]"
|
||||
|
||||
@staticmethod
|
||||
def escape(text: str) -> str:
|
||||
"""转义特殊字符
|
||||
Args:
|
||||
text: 需要转义的文本，请勿直接把整个markdown文本传入，否则会转义掉所有字符
|
||||
|
||||
Returns:
|
||||
转义后的文本
|
||||
|
||||
"""
|
||||
chars = "*[]()~_`>#+=|{}.!"
|
||||
for char in chars:
|
||||
text = text.replace(char, f"\\\\{char}")
|
||||
return text
|
||||
|
||||
@ -1,101 +1,101 @@
|
||||
import nonebot
|
||||
|
||||
|
||||
def convert_duration(text: str, default) -> float:
|
||||
"""
|
||||
转换自然语言时间为秒数
|
||||
Args:
|
||||
text: 1d2h3m
|
||||
default: 出错时返回
|
||||
|
||||
Returns:
|
||||
float: 总秒数
|
||||
"""
|
||||
units = {
|
||||
"d" : 86400,
|
||||
"h" : 3600,
|
||||
"m" : 60,
|
||||
"s" : 1,
|
||||
"ms": 0.001
|
||||
}
|
||||
|
||||
duration = 0
|
||||
current_number = ''
|
||||
current_unit = ''
|
||||
try:
|
||||
for char in text:
|
||||
if char.isdigit():
|
||||
current_number += char
|
||||
else:
|
||||
if current_number:
|
||||
duration += int(current_number) * units[current_unit]
|
||||
current_number = ''
|
||||
if char in units:
|
||||
current_unit = char
|
||||
else:
|
||||
current_unit = ''
|
||||
|
||||
if current_number:
|
||||
duration += int(current_number) * units[current_unit]
|
||||
|
||||
return duration
|
||||
|
||||
except BaseException as e:
|
||||
nonebot.logger.info(f"convert_duration error: {e}")
|
||||
return default
|
||||
|
||||
|
||||
def convert_time_to_seconds(time_str):
|
||||
"""转换自然语言时长为秒数
|
||||
Args:
|
||||
time_str: 1d2m3s
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
seconds = 0
|
||||
current_number = ''
|
||||
|
||||
for char in time_str:
|
||||
if char.isdigit() or char == '.':
|
||||
current_number += char
|
||||
elif char == 'd':
|
||||
seconds += float(current_number) * 24 * 60 * 60
|
||||
current_number = ''
|
||||
elif char == 'h':
|
||||
seconds += float(current_number) * 60 * 60
|
||||
current_number = ''
|
||||
elif char == 'm':
|
||||
seconds += float(current_number) * 60
|
||||
current_number = ''
|
||||
elif char == 's':
|
||||
seconds += float(current_number)
|
||||
current_number = ''
|
||||
|
||||
return int(seconds)
|
||||
|
||||
|
||||
def convert_seconds_to_time(seconds):
|
||||
"""转换秒数为自然语言时长
|
||||
Args:
|
||||
seconds: 10000
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
d = seconds // (24 * 60 * 60)
|
||||
h = (seconds % (24 * 60 * 60)) // (60 * 60)
|
||||
m = (seconds % (60 * 60)) // 60
|
||||
s = seconds % 60
|
||||
|
||||
# 若值为0则不显示
|
||||
time_str = ''
|
||||
if d:
|
||||
time_str += f"{d}d"
|
||||
if h:
|
||||
time_str += f"{h}h"
|
||||
if m:
|
||||
time_str += f"{m}m"
|
||||
if not time_str:
|
||||
time_str = f"{s}s"
|
||||
return time_str
|
||||
import nonebot
|
||||
|
||||
|
||||
def convert_duration(text: str, default) -> float:
|
||||
"""
|
||||
转换自然语言时间为秒数
|
||||
Args:
|
||||
text: 1d2h3m
|
||||
default: 出错时返回
|
||||
|
||||
Returns:
|
||||
float: 总秒数
|
||||
"""
|
||||
units = {
|
||||
"d" : 86400,
|
||||
"h" : 3600,
|
||||
"m" : 60,
|
||||
"s" : 1,
|
||||
"ms": 0.001
|
||||
}
|
||||
|
||||
duration = 0
|
||||
current_number = ''
|
||||
current_unit = ''
|
||||
try:
|
||||
for char in text:
|
||||
if char.isdigit():
|
||||
current_number += char
|
||||
else:
|
||||
if current_number:
|
||||
duration += int(current_number) * units[current_unit]
|
||||
current_number = ''
|
||||
if char in units:
|
||||
current_unit = char
|
||||
else:
|
||||
current_unit = ''
|
||||
|
||||
if current_number:
|
||||
duration += int(current_number) * units[current_unit]
|
||||
|
||||
return duration
|
||||
|
||||
except BaseException as e:
|
||||
nonebot.logger.info(f"convert_duration error: {e}")
|
||||
return default
|
||||
|
||||
|
||||
def convert_time_to_seconds(time_str):
|
||||
"""转换自然语言时长为秒数
|
||||
Args:
|
||||
time_str: 1d2m3s
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
seconds = 0
|
||||
current_number = ''
|
||||
|
||||
for char in time_str:
|
||||
if char.isdigit() or char == '.':
|
||||
current_number += char
|
||||
elif char == 'd':
|
||||
seconds += float(current_number) * 24 * 60 * 60
|
||||
current_number = ''
|
||||
elif char == 'h':
|
||||
seconds += float(current_number) * 60 * 60
|
||||
current_number = ''
|
||||
elif char == 'm':
|
||||
seconds += float(current_number) * 60
|
||||
current_number = ''
|
||||
elif char == 's':
|
||||
seconds += float(current_number)
|
||||
current_number = ''
|
||||
|
||||
return int(seconds)
|
||||
|
||||
|
||||
def convert_seconds_to_time(seconds):
|
||||
"""转换秒数为自然语言时长
|
||||
Args:
|
||||
seconds: 10000
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
d = seconds // (24 * 60 * 60)
|
||||
h = (seconds % (24 * 60 * 60)) // (60 * 60)
|
||||
m = (seconds % (60 * 60)) // 60
|
||||
s = seconds % 60
|
||||
|
||||
# 若值为0则不显示
|
||||
time_str = ''
|
||||
if d:
|
||||
time_str += f"{d}d"
|
||||
if h:
|
||||
time_str += f"{h}h"
|
||||
if m:
|
||||
time_str += f"{m}m"
|
||||
if not time_str:
|
||||
time_str = f"{s}s"
|
||||
return time_str
|
||||
|
||||
@ -1,99 +1,99 @@
|
||||
import random
|
||||
from importlib.metadata import PackageNotFoundError, version
|
||||
|
||||
|
||||
def clamp(value: float, min_value: float, max_value: float) -> float | int:
|
||||
"""将值限制在最小值和最大值之间
|
||||
|
||||
Args:
|
||||
value (float): 要限制的值
|
||||
min_value (float): 最小值
|
||||
max_value (float): 最大值
|
||||
|
||||
Returns:
|
||||
float: 限制后的值
|
||||
"""
|
||||
return max(min(value, max_value), min_value)
|
||||
|
||||
|
||||
def convert_size(size: int, precision: int = 2, add_unit: bool = True, suffix: str = " XiB") -> str | float:
|
||||
"""把字节数转换为人类可读的字符串，计算正负
|
||||
|
||||
Args:
|
||||
|
||||
add_unit: 是否添加单位，False后则suffix无效
|
||||
suffix: XiB或XB
|
||||
precision: 浮点数的小数点位数
|
||||
size (int): 字节数
|
||||
|
||||
Returns:
|
||||
|
||||
str: The human-readable string, e.g. "1.23 GB".
|
||||
"""
|
||||
is_negative = size < 0
|
||||
size = abs(size)
|
||||
for unit in ("", "K", "M", "G", "T", "P", "E", "Z"):
|
||||
if size < 1024:
|
||||
break
|
||||
size /= 1024
|
||||
if is_negative:
|
||||
size = -size
|
||||
if add_unit:
|
||||
return f"{size:.{precision}f}{suffix.replace('X', unit)}"
|
||||
else:
|
||||
return size
|
||||
|
||||
|
||||
def keywords_in_text(keywords: list[str], text: str, all_matched: bool) -> bool:
|
||||
"""
|
||||
检查关键词是否在文本中
|
||||
Args:
|
||||
keywords: 关键词列表
|
||||
text: 文本
|
||||
all_matched: 是否需要全部匹配
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
if all_matched:
|
||||
for keyword in keywords:
|
||||
if keyword not in text:
|
||||
return False
|
||||
return True
|
||||
else:
|
||||
for keyword in keywords:
|
||||
if keyword in text:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def check_for_package(package_name: str) -> bool:
|
||||
try:
|
||||
version(package_name)
|
||||
return True
|
||||
except PackageNotFoundError:
|
||||
return False
|
||||
|
||||
|
||||
def random_ascii_string(length: int) -> str:
|
||||
"""
|
||||
生成随机ASCII字符串
|
||||
Args:
|
||||
length:
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
return "".join([chr(random.randint(33, 126)) for _ in range(length)])
|
||||
|
||||
|
||||
def random_hex_string(length: int) -> str:
|
||||
"""
|
||||
生成随机十六进制字符串
|
||||
Args:
|
||||
length:
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
return "".join([random.choice("0123456789abcdef") for _ in range(length)])
|
||||
import random
|
||||
from importlib.metadata import PackageNotFoundError, version
|
||||
|
||||
|
||||
def clamp(value: float, min_value: float, max_value: float) -> float | int:
|
||||
"""将值限制在最小值和最大值之间
|
||||
|
||||
Args:
|
||||
value (float): 要限制的值
|
||||
min_value (float): 最小值
|
||||
max_value (float): 最大值
|
||||
|
||||
Returns:
|
||||
float: 限制后的值
|
||||
"""
|
||||
return max(min(value, max_value), min_value)
|
||||
|
||||
|
||||
def convert_size(size: int, precision: int = 2, add_unit: bool = True, suffix: str = " XiB") -> str | float:
|
||||
"""把字节数转换为人类可读的字符串，计算正负
|
||||
|
||||
Args:
|
||||
|
||||
add_unit: 是否添加单位，False后则suffix无效
|
||||
suffix: XiB或XB
|
||||
precision: 浮点数的小数点位数
|
||||
size (int): 字节数
|
||||
|
||||
Returns:
|
||||
|
||||
str: The human-readable string, e.g. "1.23 GB".
|
||||
"""
|
||||
is_negative = size < 0
|
||||
size = abs(size)
|
||||
for unit in ("", "K", "M", "G", "T", "P", "E", "Z"):
|
||||
if size < 1024:
|
||||
break
|
||||
size /= 1024
|
||||
if is_negative:
|
||||
size = -size
|
||||
if add_unit:
|
||||
return f"{size:.{precision}f}{suffix.replace('X', unit)}"
|
||||
else:
|
||||
return size
|
||||
|
||||
|
||||
def keywords_in_text(keywords: list[str], text: str, all_matched: bool) -> bool:
|
||||
"""
|
||||
检查关键词是否在文本中
|
||||
Args:
|
||||
keywords: 关键词列表
|
||||
text: 文本
|
||||
all_matched: 是否需要全部匹配
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
if all_matched:
|
||||
for keyword in keywords:
|
||||
if keyword not in text:
|
||||
return False
|
||||
return True
|
||||
else:
|
||||
for keyword in keywords:
|
||||
if keyword in text:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def check_for_package(package_name: str) -> bool:
|
||||
try:
|
||||
version(package_name)
|
||||
return True
|
||||
except PackageNotFoundError:
|
||||
return False
|
||||
|
||||
|
||||
def random_ascii_string(length: int) -> str:
|
||||
"""
|
||||
生成随机ASCII字符串
|
||||
Args:
|
||||
length:
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
return "".join([chr(random.randint(33, 126)) for _ in range(length)])
|
||||
|
||||
|
||||
def random_hex_string(length: int) -> str:
|
||||
"""
|
||||
生成随机十六进制字符串
|
||||
Args:
|
||||
length:
|
||||
|
||||
Returns:
|
||||
|
||||
"""
|
||||
return "".join([random.choice("0123456789abcdef") for _ in range(length)])
|
||||
|
||||
Reference in New Issue
Block a user