mirror of
https://github.com/RockChinQ/QChatGPT.git
synced 2024-11-16 03:32:33 +08:00
chore: 删除 qqbot 包中的流程代码
This commit is contained in:
RockChinQ
parent
8d084427d2
commit
1900ddacbb
|
|
@ -1,70 +0,0 @@
|
|||
# 处理对会话的禁用配置
|
||||
# 过去的 banlist
|
||||
from __future__ import annotations
|
||||
import re
|
||||
|
||||
from ...core import app
|
||||
from ...config import manager as cfg_mgr
|
||||
|
||||
|
||||
class SessionBanManager:
|
||||
|
||||
ap: app.Application = None
|
||||
|
||||
banlist_mgr: cfg_mgr.ConfigManager
|
||||
|
||||
def __init__(self, ap: app.Application):
|
||||
self.ap = ap
|
||||
|
||||
async def initialize(self):
|
||||
self.banlist_mgr = await cfg_mgr.load_python_module_config(
|
||||
"banlist.py",
|
||||
"res/templates/banlist-template.py"
|
||||
)
|
||||
|
||||
async def is_banned(
|
||||
self, launcher_type: str, launcher_id: int, sender_id: int
|
||||
) -> bool:
|
||||
if not self.banlist_mgr.data['enable']:
|
||||
return False
|
||||
|
||||
result = False
|
||||
|
||||
if launcher_type == 'group':
|
||||
if not self.banlist_mgr.data['enable_group']: # 未启用群聊响应
|
||||
result = True
|
||||
# 检查是否显式声明发起人QQ要被person忽略
|
||||
elif sender_id in self.banlist_mgr.data['person']:
|
||||
result = True
|
||||
else:
|
||||
for group_rule in self.banlist_mgr.data['group']:
|
||||
if type(group_rule) == int:
|
||||
if group_rule == launcher_id:
|
||||
result = True
|
||||
elif type(group_rule) == str:
|
||||
if group_rule.startswith('!'):
|
||||
reg_str = group_rule[1:]
|
||||
if re.match(reg_str, str(launcher_id)):
|
||||
result = False
|
||||
break
|
||||
else:
|
||||
if re.match(group_rule, str(launcher_id)):
|
||||
result = True
|
||||
elif launcher_type == 'person':
|
||||
if not self.banlist_mgr.data['enable_private']:
|
||||
result = True
|
||||
else:
|
||||
for person_rule in self.banlist_mgr.data['person']:
|
||||
if type(person_rule) == int:
|
||||
if person_rule == launcher_id:
|
||||
result = True
|
||||
elif type(person_rule) == str:
|
||||
if person_rule.startswith('!'):
|
||||
reg_str = person_rule[1:]
|
||||
if re.match(reg_str, str(launcher_id)):
|
||||
result = False
|
||||
break
|
||||
else:
|
||||
if re.match(person_rule, str(launcher_id)):
|
||||
result = True
|
||||
return result
|
||||
|
|
@ -1,93 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from ...core import app
|
||||
from . import entities
|
||||
from . import filter
|
||||
from .filters import cntignore, banwords, baiduexamine
|
||||
|
||||
|
||||
class ContentFilterManager:
|
||||
|
||||
ao: app.Application
|
||||
|
||||
filter_chain: list[filter.ContentFilter]
|
||||
|
||||
def __init__(self, ap: app.Application) -> None:
|
||||
self.ap = ap
|
||||
self.filter_chain = []
|
||||
|
||||
async def initialize(self):
|
||||
self.filter_chain.append(cntignore.ContentIgnore(self.ap))
|
||||
|
||||
if self.ap.cfg_mgr.data['sensitive_word_filter']:
|
||||
self.filter_chain.append(banwords.BanWordFilter(self.ap))
|
||||
|
||||
if self.ap.cfg_mgr.data['baidu_check']:
|
||||
self.filter_chain.append(baiduexamine.BaiduCloudExamine(self.ap))
|
||||
|
||||
for filter in self.filter_chain:
|
||||
await filter.initialize()
|
||||
|
||||
async def pre_process(self, message: str) -> entities.FilterManagerResult:
|
||||
"""请求llm前处理消息
|
||||
只要有一个不通过就不放行，只放行 PASS 的消息
|
||||
"""
|
||||
if not self.ap.cfg_mgr.data['income_msg_check']: # 不检查收到的消息，直接放行
|
||||
return entities.FilterManagerResult(
|
||||
level=entities.ManagerResultLevel.CONTINUE,
|
||||
replacement=message,
|
||||
user_notice='',
|
||||
console_notice=''
|
||||
)
|
||||
else:
|
||||
for filter in self.filter_chain:
|
||||
if entities.EnableStage.PRE in filter.enable_stages:
|
||||
result = await filter.process(message)
|
||||
|
||||
if result.level in [
|
||||
entities.ResultLevel.BLOCK,
|
||||
entities.ResultLevel.MASKED
|
||||
]:
|
||||
return entities.FilterManagerResult(
|
||||
level=entities.ManagerResultLevel.INTERRUPT,
|
||||
replacement=result.replacement,
|
||||
user_notice=result.user_notice,
|
||||
console_notice=result.console_notice
|
||||
)
|
||||
elif result.level == entities.ResultLevel.PASS:
|
||||
message = result.replacement
|
||||
|
||||
return entities.FilterManagerResult(
|
||||
level=entities.ManagerResultLevel.CONTINUE,
|
||||
replacement=message,
|
||||
user_notice='',
|
||||
console_notice=''
|
||||
)
|
||||
|
||||
async def post_process(self, message: str) -> entities.FilterManagerResult:
|
||||
"""请求llm后处理响应
|
||||
只要是 PASS 或者 MASKED 的就通过此 filter，将其 replacement 设置为message，进入下一个 filter
|
||||
"""
|
||||
for filter in self.filter_chain:
|
||||
if entities.EnableStage.POST in filter.enable_stages:
|
||||
result = await filter.process(message)
|
||||
|
||||
if result.level == entities.ResultLevel.BLOCK:
|
||||
return entities.FilterManagerResult(
|
||||
level=entities.ManagerResultLevel.INTERRUPT,
|
||||
replacement=result.replacement,
|
||||
user_notice=result.user_notice,
|
||||
console_notice=result.console_notice
|
||||
)
|
||||
elif result.level in [
|
||||
entities.ResultLevel.PASS,
|
||||
entities.ResultLevel.MASKED
|
||||
]:
|
||||
message = result.replacement
|
||||
|
||||
return entities.FilterManagerResult(
|
||||
level=entities.ManagerResultLevel.CONTINUE,
|
||||
replacement=message,
|
||||
user_notice='',
|
||||
console_notice=''
|
||||
)
|
||||
|
|
@ -1,64 +0,0 @@
|
|||
|
||||
import typing
|
||||
import enum
|
||||
|
||||
import pydantic
|
||||
|
||||
|
||||
class ResultLevel(enum.Enum):
|
||||
"""结果等级"""
|
||||
PASS = enum.auto()
|
||||
"""通过"""
|
||||
|
||||
WARN = enum.auto()
|
||||
"""警告"""
|
||||
|
||||
MASKED = enum.auto()
|
||||
"""已掩去"""
|
||||
|
||||
BLOCK = enum.auto()
|
||||
"""阻止"""
|
||||
|
||||
|
||||
class EnableStage(enum.Enum):
|
||||
"""启用阶段"""
|
||||
PRE = enum.auto()
|
||||
"""预处理"""
|
||||
|
||||
POST = enum.auto()
|
||||
"""后处理"""
|
||||
|
||||
|
||||
class FilterResult(pydantic.BaseModel):
|
||||
level: ResultLevel
|
||||
|
||||
replacement: str
|
||||
"""替换后的消息"""
|
||||
|
||||
user_notice: str
|
||||
"""不通过时，用户提示消息"""
|
||||
|
||||
console_notice: str
|
||||
"""不通过时，控制台提示消息"""
|
||||
|
||||
|
||||
class ManagerResultLevel(enum.Enum):
|
||||
"""处理器结果等级"""
|
||||
CONTINUE = enum.auto()
|
||||
"""继续"""
|
||||
|
||||
INTERRUPT = enum.auto()
|
||||
"""中断"""
|
||||
|
||||
class FilterManagerResult(pydantic.BaseModel):
|
||||
|
||||
level: ManagerResultLevel
|
||||
|
||||
replacement: str
|
||||
"""替换后的消息"""
|
||||
|
||||
user_notice: str
|
||||
"""用户提示消息"""
|
||||
|
||||
console_notice: str
|
||||
"""控制台提示消息"""
|
||||
|
|
@ -1,34 +0,0 @@
|
|||
# 内容过滤器的抽象类
|
||||
from __future__ import annotations
|
||||
import abc
|
||||
|
||||
from ...core import app
|
||||
from . import entities
|
||||
|
||||
|
||||
class ContentFilter(metaclass=abc.ABCMeta):
|
||||
|
||||
ap: app.Application
|
||||
|
||||
def __init__(self, ap: app.Application):
|
||||
self.ap = ap
|
||||
|
||||
@property
|
||||
def enable_stages(self):
|
||||
"""启用的阶段
|
||||
"""
|
||||
return [
|
||||
entities.EnableStage.PRE,
|
||||
entities.EnableStage.POST
|
||||
]
|
||||
|
||||
async def initialize(self):
|
||||
"""初始化过滤器
|
||||
"""
|
||||
pass
|
||||
|
||||
@abc.abstractmethod
|
||||
async def process(self, message: str) -> entities.FilterResult:
|
||||
"""处理消息
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
|
@ -1,61 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import aiohttp
|
||||
|
||||
from .. import entities
|
||||
from .. import filter as filter_model
|
||||
|
||||
|
||||
BAIDU_EXAMINE_URL = "https://aip.baidubce.com/rest/2.0/solution/v1/text_censor/v2/user_defined?access_token={}"
|
||||
BAIDU_EXAMINE_TOKEN_URL = "https://aip.baidubce.com/oauth/2.0/token"
|
||||
|
||||
|
||||
class BaiduCloudExamine(filter_model.ContentFilter):
|
||||
"""百度云内容审核"""
|
||||
|
||||
async def _get_token(self) -> str:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(
|
||||
BAIDU_EXAMINE_TOKEN_URL,
|
||||
params={
|
||||
"grant_type": "client_credentials",
|
||||
"client_id": self.ap.cfg_mgr.data['baidu_api_key'],
|
||||
"client_secret": self.ap.cfg_mgr.data['baidu_secret_key']
|
||||
}
|
||||
) as resp:
|
||||
return (await resp.json())['access_token']
|
||||
|
||||
async def process(self, message: str) -> entities.FilterResult:
|
||||
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.post(
|
||||
BAIDU_EXAMINE_URL.format(await self._get_token()),
|
||||
headers={'Content-Type': 'application/x-www-form-urlencoded', 'Accept': 'application/json'},
|
||||
data=f"text={message}".encode('utf-8')
|
||||
) as resp:
|
||||
result = await resp.json()
|
||||
|
||||
if "error_code" in result:
|
||||
return entities.FilterResult(
|
||||
level=entities.ResultLevel.BLOCK,
|
||||
replacement=message,
|
||||
user_notice='',
|
||||
console_notice=f"百度云判定出错，错误信息：{result['error_msg']}"
|
||||
)
|
||||
else:
|
||||
conclusion = result["conclusion"]
|
||||
|
||||
if conclusion in ("合规"):
|
||||
return entities.FilterResult(
|
||||
level=entities.ResultLevel.PASS,
|
||||
replacement=message,
|
||||
user_notice='',
|
||||
console_notice=f"百度云判定结果：{conclusion}"
|
||||
)
|
||||
else:
|
||||
return entities.FilterResult(
|
||||
level=entities.ResultLevel.BLOCK,
|
||||
replacement=message,
|
||||
user_notice=self.ap.cfg_mgr.data['inappropriate_message_tips'],
|
||||
console_notice=f"百度云判定结果：{conclusion}"
|
||||
)
|
||||
|
|
@ -1,44 +0,0 @@
|
|||
from __future__ import annotations
|
||||
import re
|
||||
|
||||
from .. import filter as filter_model
|
||||
from .. import entities
|
||||
from ....config import manager as cfg_mgr
|
||||
|
||||
|
||||
class BanWordFilter(filter_model.ContentFilter):
|
||||
"""根据内容禁言"""
|
||||
|
||||
sensitive: cfg_mgr.ConfigManager
|
||||
|
||||
async def initialize(self):
|
||||
self.sensitive = await cfg_mgr.load_json_config(
|
||||
"sensitive.json",
|
||||
"res/templates/sensitive-template.json"
|
||||
)
|
||||
|
||||
async def process(self, message: str) -> entities.FilterResult:
|
||||
found = False
|
||||
|
||||
for word in self.sensitive.data['words']:
|
||||
match = re.findall(word, message)
|
||||
|
||||
if len(match) > 0:
|
||||
found = True
|
||||
|
||||
for i in range(len(match)):
|
||||
if self.sensitive.data['mask_word'] == "":
|
||||
message = message.replace(
|
||||
match[i], self.sensitive.data['mask'] * len(match[i])
|
||||
)
|
||||
else:
|
||||
message = message.replace(
|
||||
match[i], self.sensitive.data['mask_word']
|
||||
)
|
||||
|
||||
return entities.FilterResult(
|
||||
level=entities.ResultLevel.MASKED if found else entities.ResultLevel.PASS,
|
||||
replacement=message,
|
||||
user_notice='[bot] 消息中存在不合适的内容, 请更换措辞' if found else '',
|
||||
console_notice=''
|
||||
)
|
||||
|
|
@ -1,43 +0,0 @@
|
|||
from __future__ import annotations
|
||||
import re
|
||||
|
||||
from .. import entities
|
||||
from .. import filter as filter_model
|
||||
|
||||
|
||||
class ContentIgnore(filter_model.ContentFilter):
|
||||
"""根据内容忽略消息"""
|
||||
|
||||
@property
|
||||
def enable_stages(self):
|
||||
return [
|
||||
entities.EnableStage.PRE,
|
||||
]
|
||||
|
||||
async def process(self, message: str) -> entities.FilterResult:
|
||||
if 'prefix' in self.ap.cfg_mgr.data['ignore_rules']:
|
||||
for rule in self.ap.cfg_mgr.data['ignore_rules']['prefix']:
|
||||
if message.startswith(rule):
|
||||
return entities.FilterResult(
|
||||
level=entities.ResultLevel.BLOCK,
|
||||
replacement='',
|
||||
user_notice='',
|
||||
console_notice='根据 ignore_rules 中的 prefix 规则，忽略消息'
|
||||
)
|
||||
|
||||
if 'regexp' in self.ap.cfg_mgr.data['ignore_rules']:
|
||||
for rule in self.ap.cfg_mgr.data['ignore_rules']['regexp']:
|
||||
if re.search(rule, message):
|
||||
return entities.FilterResult(
|
||||
level=entities.ResultLevel.BLOCK,
|
||||
replacement='',
|
||||
user_notice='',
|
||||
console_notice='根据 ignore_rules 中的 regexp 规则，忽略消息'
|
||||
)
|
||||
|
||||
return entities.FilterResult(
|
||||
level=entities.ResultLevel.PASS,
|
||||
replacement=message,
|
||||
user_notice='',
|
||||
console_notice=''
|
||||
)
|
||||
|
|
@ -1,56 +0,0 @@
|
|||
from __future__ import annotations
|
||||
import os
|
||||
import traceback
|
||||
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
from mirai.models.message import MessageComponent, Plain
|
||||
|
||||
from ...core import app
|
||||
from . import strategy
|
||||
from .strategies import image, forward
|
||||
|
||||
|
||||
class LongTextProcessor:
|
||||
|
||||
ap: app.Application
|
||||
|
||||
strategy_impl: strategy.LongTextStrategy
|
||||
|
||||
def __init__(self, ap: app.Application):
|
||||
self.ap = ap
|
||||
|
||||
async def initialize(self):
|
||||
config = self.ap.cfg_mgr.data
|
||||
if self.ap.cfg_mgr.data['blob_message_strategy'] == 'image':
|
||||
use_font = config['font_path']
|
||||
try:
|
||||
# 检查是否存在
|
||||
if not os.path.exists(use_font):
|
||||
# 若是windows系统，使用微软雅黑
|
||||
if os.name == "nt":
|
||||
use_font = "C:/Windows/Fonts/msyh.ttc"
|
||||
if not os.path.exists(use_font):
|
||||
self.ap.logger.warn("未找到字体文件，且无法使用Windows自带字体，更换为转发消息组件以发送长消息，您可以在config.py中调整相关设置。")
|
||||
config['blob_message_strategy'] = "forward"
|
||||
else:
|
||||
self.ap.logger.info("使用Windows自带字体：" + use_font)
|
||||
self.ap.cfg_mgr.data['font_path'] = use_font
|
||||
else:
|
||||
self.ap.logger.warn("未找到字体文件，且无法使用系统自带字体，更换为转发消息组件以发送长消息，您可以在config.py中调整相关设置。")
|
||||
self.ap.cfg_mgr.data['blob_message_strategy'] = "forward"
|
||||
except:
|
||||
traceback.print_exc()
|
||||
self.ap.logger.error("加载字体文件失败({})，更换为转发消息组件以发送长消息，您可以在config.py中调整相关设置。".format(use_font))
|
||||
self.ap.cfg_mgr.data['blob_message_strategy'] = "forward"
|
||||
|
||||
if self.ap.cfg_mgr.data['blob_message_strategy'] == 'image':
|
||||
self.strategy_impl = image.Text2ImageStrategy(self.ap)
|
||||
elif self.ap.cfg_mgr.data['blob_message_strategy'] == 'forward':
|
||||
self.strategy_impl = forward.ForwardComponentStrategy(self.ap)
|
||||
await self.strategy_impl.initialize()
|
||||
|
||||
async def check_and_process(self, message: str) -> list[MessageComponent]:
|
||||
if len(message) > self.ap.cfg_mgr.data['blob_message_threshold']:
|
||||
return await self.strategy_impl.process(message)
|
||||
else:
|
||||
return [Plain(message)]
|
||||
|
|
@ -1,62 +0,0 @@
|
|||
# 转发消息组件
|
||||
from __future__ import annotations
|
||||
import typing
|
||||
|
||||
from mirai.models import MessageChain
|
||||
from mirai.models.message import MessageComponent, ForwardMessageNode
|
||||
from mirai.models.base import MiraiBaseModel
|
||||
|
||||
from .. import strategy as strategy_model
|
||||
|
||||
|
||||
class ForwardMessageDiaplay(MiraiBaseModel):
|
||||
title: str = "群聊的聊天记录"
|
||||
brief: str = "[聊天记录]"
|
||||
source: str = "聊天记录"
|
||||
preview: typing.List[str] = []
|
||||
summary: str = "查看x条转发消息"
|
||||
|
||||
|
||||
class Forward(MessageComponent):
|
||||
"""合并转发。"""
|
||||
type: str = "Forward"
|
||||
"""消息组件类型。"""
|
||||
display: ForwardMessageDiaplay
|
||||
"""显示信息"""
|
||||
node_list: typing.List[ForwardMessageNode]
|
||||
"""转发消息节点列表。"""
|
||||
def __init__(self, *args, **kwargs):
|
||||
if len(args) == 1:
|
||||
self.node_list = args[0]
|
||||
super().__init__(**kwargs)
|
||||
super().__init__(*args, **kwargs)
|
||||
|
||||
def __str__(self):
|
||||
return '[聊天记录]'
|
||||
|
||||
|
||||
class ForwardComponentStrategy(strategy_model.LongTextStrategy):
|
||||
|
||||
async def process(self, message: str) -> list[MessageComponent]:
|
||||
display = ForwardMessageDiaplay(
|
||||
title="群聊的聊天记录",
|
||||
brief="[聊天记录]",
|
||||
source="聊天记录",
|
||||
preview=["QQ用户: "+message],
|
||||
summary="查看1条转发消息"
|
||||
)
|
||||
|
||||
node_list = [
|
||||
ForwardMessageNode(
|
||||
sender_id=self.ap.im_mgr.bot_account_id,
|
||||
sender_name='QQ用户',
|
||||
message_chain=MessageChain([message])
|
||||
)
|
||||
]
|
||||
|
||||
forward = Forward(
|
||||
display=display,
|
||||
node_list=node_list
|
||||
)
|
||||
|
||||
return [forward]
|
||||
|
|
@ -1,197 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import typing
|
||||
import os
|
||||
import base64
|
||||
import time
|
||||
import re
|
||||
|
||||
from PIL import Image, ImageDraw, ImageFont
|
||||
|
||||
from mirai.models import MessageChain, Image as ImageComponent
|
||||
from mirai.models.message import MessageComponent
|
||||
|
||||
from .. import strategy as strategy_model
|
||||
|
||||
|
||||
class Text2ImageStrategy(strategy_model.LongTextStrategy):
|
||||
|
||||
text_render_font: ImageFont.FreeTypeFont
|
||||
|
||||
async def initialize(self):
|
||||
self.text_render_font = ImageFont.truetype(self.ap.cfg_mgr.data['font_path'], 32, encoding="utf-8")
|
||||
|
||||
async def process(self, message: str) -> list[MessageComponent]:
|
||||
img_path = self.text_to_image(
|
||||
text_str=message,
|
||||
save_as='temp/{}.png'.format(int(time.time()))
|
||||
)
|
||||
|
||||
compressed_path, size = self.compress_image(
|
||||
img_path,
|
||||
outfile="temp/{}_compressed.png".format(int(time.time()))
|
||||
)
|
||||
|
||||
with open(compressed_path, 'rb') as f:
|
||||
img = f.read()
|
||||
|
||||
b64 = base64.b64encode(img)
|
||||
|
||||
# 删除图片
|
||||
os.remove(img_path)
|
||||
|
||||
if os.path.exists(compressed_path):
|
||||
os.remove(compressed_path)
|
||||
|
||||
return [
|
||||
ImageComponent(
|
||||
base64=b64.decode('utf-8'),
|
||||
)
|
||||
]
|
||||
|
||||
def indexNumber(self, path=''):
|
||||
"""
|
||||
查找字符串中数字所在串中的位置
|
||||
:param path:目标字符串
|
||||
:return:<class 'list'>: <class 'list'>: [['1', 16], ['2', 35], ['1', 51]]
|
||||
"""
|
||||
kv = []
|
||||
nums = []
|
||||
beforeDatas = re.findall('[\d]+', path)
|
||||
for num in beforeDatas:
|
||||
indexV = []
|
||||
times = path.count(num)
|
||||
if times > 1:
|
||||
if num not in nums:
|
||||
indexs = re.finditer(num, path)
|
||||
for index in indexs:
|
||||
iV = []
|
||||
i = index.span()[0]
|
||||
iV.append(num)
|
||||
iV.append(i)
|
||||
kv.append(iV)
|
||||
nums.append(num)
|
||||
else:
|
||||
index = path.find(num)
|
||||
indexV.append(num)
|
||||
indexV.append(index)
|
||||
kv.append(indexV)
|
||||
# 根据数字位置排序
|
||||
indexSort = []
|
||||
resultIndex = []
|
||||
for vi in kv:
|
||||
indexSort.append(vi[1])
|
||||
indexSort.sort()
|
||||
for i in indexSort:
|
||||
for v in kv:
|
||||
if i == v[1]:
|
||||
resultIndex.append(v)
|
||||
return resultIndex
|
||||
|
||||
|
||||
def get_size(self, file):
|
||||
# 获取文件大小:KB
|
||||
size = os.path.getsize(file)
|
||||
return size / 1024
|
||||
|
||||
|
||||
def get_outfile(self, infile, outfile):
|
||||
if outfile:
|
||||
return outfile
|
||||
dir, suffix = os.path.splitext(infile)
|
||||
outfile = '{}-out{}'.format(dir, suffix)
|
||||
return outfile
|
||||
|
||||
|
||||
def compress_image(self, infile, outfile='', kb=100, step=20, quality=90):
|
||||
"""不改变图片尺寸压缩到指定大小
|
||||
:param infile: 压缩源文件
|
||||
:param outfile: 压缩文件保存地址
|
||||
:param mb: 压缩目标,KB
|
||||
:param step: 每次调整的压缩比率
|
||||
:param quality: 初始压缩比率
|
||||
:return: 压缩文件地址，压缩文件大小
|
||||
"""
|
||||
o_size = self.get_size(infile)
|
||||
if o_size <= kb:
|
||||
return infile, o_size
|
||||
outfile = self.get_outfile(infile, outfile)
|
||||
while o_size > kb:
|
||||
im = Image.open(infile)
|
||||
im.save(outfile, quality=quality)
|
||||
if quality - step < 0:
|
||||
break
|
||||
quality -= step
|
||||
o_size = self.get_size(outfile)
|
||||
return outfile, self.get_size(outfile)
|
||||
|
||||
|
||||
def text_to_image(self, text_str: str, save_as="temp.png", width=800):
|
||||
|
||||
text_str = text_str.replace("\t", " ")
|
||||
|
||||
# 分行
|
||||
lines = text_str.split('\n')
|
||||
|
||||
# 计算并分割
|
||||
final_lines = []
|
||||
|
||||
text_width = width-80
|
||||
|
||||
self.ap.logger.debug("lines: {}, text_width: {}".format(lines, text_width))
|
||||
for line in lines:
|
||||
# 如果长了就分割
|
||||
line_width = self.text_render_font.getlength(line)
|
||||
self.ap.logger.debug("line_width: {}".format(line_width))
|
||||
if line_width < text_width:
|
||||
final_lines.append(line)
|
||||
continue
|
||||
else:
|
||||
rest_text = line
|
||||
while True:
|
||||
# 分割最前面的一行
|
||||
point = int(len(rest_text) * (text_width / line_width))
|
||||
|
||||
# 检查断点是否在数字中间
|
||||
numbers = self.indexNumber(rest_text)
|
||||
|
||||
for number in numbers:
|
||||
if number[1] < point < number[1] + len(number[0]) and number[1] != 0:
|
||||
point = number[1]
|
||||
break
|
||||
|
||||
final_lines.append(rest_text[:point])
|
||||
rest_text = rest_text[point:]
|
||||
line_width = self.text_render_font.getlength(rest_text)
|
||||
if line_width < text_width:
|
||||
final_lines.append(rest_text)
|
||||
break
|
||||
else:
|
||||
continue
|
||||
# 准备画布
|
||||
img = Image.new('RGBA', (width, max(280, len(final_lines) * 35 + 65)), (255, 255, 255, 255))
|
||||
draw = ImageDraw.Draw(img, mode='RGBA')
|
||||
|
||||
self.ap.logger.debug("正在绘制图片...")
|
||||
# 绘制正文
|
||||
line_number = 0
|
||||
offset_x = 20
|
||||
offset_y = 30
|
||||
for final_line in final_lines:
|
||||
draw.text((offset_x, offset_y + 35 * line_number), final_line, fill=(0, 0, 0), font=self.text_render_font)
|
||||
# 遍历此行,检查是否有emoji
|
||||
idx_in_line = 0
|
||||
for ch in final_line:
|
||||
# 检查字符占位宽
|
||||
char_code = ord(ch)
|
||||
if char_code >= 127:
|
||||
idx_in_line += 1
|
||||
else:
|
||||
idx_in_line += 0.5
|
||||
|
||||
line_number += 1
|
||||
|
||||
self.ap.logger.debug("正在保存图片...")
|
||||
img.save(save_as)
|
||||
|
||||
return save_as
|
||||
|
|
@ -1,22 +0,0 @@
|
|||
from __future__ import annotations
|
||||
import abc
|
||||
import typing
|
||||
|
||||
import mirai
|
||||
from mirai.models.message import MessageComponent
|
||||
|
||||
from ...core import app
|
||||
|
||||
|
||||
class LongTextStrategy(metaclass=abc.ABCMeta):
|
||||
ap: app.Application
|
||||
|
||||
def __init__(self, ap: app.Application):
|
||||
self.ap = ap
|
||||
|
||||
async def initialize(self):
|
||||
pass
|
||||
|
||||
@abc.abstractmethod
|
||||
async def process(self, message: str) -> list[MessageComponent]:
|
||||
return []
|
||||
|
|
@ -1,9 +0,0 @@
|
|||
import pydantic
|
||||
import mirai
|
||||
|
||||
|
||||
class RuleJudgeResult(pydantic.BaseModel):
|
||||
|
||||
matching: bool = False
|
||||
|
||||
replacement: mirai.MessageChain = None
|
||||
|
|
@ -1,58 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import mirai
|
||||
|
||||
from ...core import app
|
||||
from . import entities, rule
|
||||
from .rules import atbot, prefix, regexp, random
|
||||
|
||||
|
||||
class GroupRespondRuleChecker:
|
||||
"""群组响应规则检查器
|
||||
"""
|
||||
|
||||
ap: app.Application
|
||||
|
||||
rule_matchers: list[rule.GroupRespondRule]
|
||||
|
||||
def __init__(self, ap: app.Application):
|
||||
self.ap = ap
|
||||
|
||||
async def initialize(self):
|
||||
"""初始化检查器
|
||||
"""
|
||||
self.rule_matchers = [
|
||||
atbot.AtBotRule(self.ap),
|
||||
prefix.PrefixRule(self.ap),
|
||||
regexp.RegExpRule(self.ap),
|
||||
random.RandomRespRule(self.ap),
|
||||
]
|
||||
|
||||
for rule_matcher in self.rule_matchers:
|
||||
await rule_matcher.initialize()
|
||||
|
||||
async def check(
|
||||
self,
|
||||
message_text: str,
|
||||
message_chain: mirai.MessageChain,
|
||||
launcher_id: int,
|
||||
sender_id: int,
|
||||
) -> entities.RuleJudgeResult:
|
||||
"""检查消息是否匹配规则
|
||||
"""
|
||||
rules = self.ap.cfg_mgr.data['response_rules']
|
||||
|
||||
use_rule = rules['default']
|
||||
|
||||
if str(launcher_id) in use_rule:
|
||||
use_rule = use_rule[str(launcher_id)]
|
||||
|
||||
for rule_matcher in self.rule_matchers:
|
||||
res = await rule_matcher.match(message_text, message_chain, use_rule)
|
||||
if res.matching:
|
||||
return res
|
||||
|
||||
return entities.RuleJudgeResult(
|
||||
matching=False,
|
||||
replacement=message_chain
|
||||
)
|
||||
|
|
@ -1,31 +0,0 @@
|
|||
from __future__ import annotations
|
||||
import abc
|
||||
|
||||
import mirai
|
||||
|
||||
from ...core import app
|
||||
from . import entities
|
||||
|
||||
|
||||
class GroupRespondRule(metaclass=abc.ABCMeta):
|
||||
"""群组响应规则的抽象类
|
||||
"""
|
||||
|
||||
ap: app.Application
|
||||
|
||||
def __init__(self, ap: app.Application):
|
||||
self.ap = ap
|
||||
|
||||
async def initialize(self):
|
||||
pass
|
||||
|
||||
@abc.abstractmethod
|
||||
async def match(
|
||||
self,
|
||||
message_text: str,
|
||||
message_chain: mirai.MessageChain,
|
||||
rule_dict: dict
|
||||
) -> entities.RuleJudgeResult:
|
||||
"""判断消息是否匹配规则
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
|
@ -1,28 +0,0 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import mirai
|
||||
|
||||
from .. import rule as rule_model
|
||||
from .. import entities
|
||||
|
||||
|
||||
class AtBotRule(rule_model.GroupRespondRule):
|
||||
|
||||
async def match(
|
||||
self,
|
||||
message_text: str,
|
||||
message_chain: mirai.MessageChain,
|
||||
rule_dict: dict
|
||||
) -> entities.RuleJudgeResult:
|
||||
|
||||
if message_chain.has(mirai.At(self.ap.im_mgr.bot_account_id)) and rule_dict['at']:
|
||||
message_chain.remove(mirai.At(self.ap.im_mgr.bot_account_id))
|
||||
return entities.RuleJudgeResult(
|
||||
matching=True,
|
||||
replacement=message_chain,
|
||||
)
|
||||
|
||||
return entities.RuleJudgeResult(
|
||||
matching=False,
|
||||
replacement = message_chain
|
||||
)
|
||||
|
|
@ -1,29 +0,0 @@
|
|||
import mirai
|
||||
|
||||
from .. import rule as rule_model
|
||||
from .. import entities
|
||||
|
||||
|
||||
class PrefixRule(rule_model.GroupRespondRule):
|
||||
|
||||
async def match(
|
||||
self,
|
||||
message_text: str,
|
||||
message_chain: mirai.MessageChain,
|
||||
rule_dict: dict
|
||||
) -> entities.RuleJudgeResult:
|
||||
prefixes = rule_dict['prefix']
|
||||
|
||||
for prefix in prefixes:
|
||||
if message_text.startswith(prefix):
|
||||
return entities.RuleJudgeResult(
|
||||
matching=True,
|
||||
replacement=mirai.MessageChain([
|
||||
mirai.Plain(message_text[len(prefix):])
|
||||
]),
|
||||
)
|
||||
|
||||
return entities.RuleJudgeResult(
|
||||
matching=False,
|
||||
replacement=message_chain
|
||||
)
|
||||
|
|
@ -1,22 +0,0 @@
|
|||
import random
|
||||
|
||||
import mirai
|
||||
|
||||
from .. import rule as rule_model
|
||||
from .. import entities
|
||||
|
||||
|
||||
class RandomRespRule(rule_model.GroupRespondRule):
|
||||
|
||||
async def match(
|
||||
self,
|
||||
message_text: str,
|
||||
message_chain: mirai.MessageChain,
|
||||
rule_dict: dict
|
||||
) -> entities.RuleJudgeResult:
|
||||
random_rate = rule_dict['random_rate']
|
||||
|
||||
return entities.RuleJudgeResult(
|
||||
matching=random.random() < random_rate,
|
||||
replacement=message_chain
|
||||
)
|
||||
|
|
@ -1,31 +0,0 @@
|
|||
import re
|
||||
|
||||
import mirai
|
||||
|
||||
from .. import rule as rule_model
|
||||
from .. import entities
|
||||
|
||||
|
||||
class RegExpRule(rule_model.GroupRespondRule):
|
||||
|
||||
async def match(
|
||||
self,
|
||||
message_text: str,
|
||||
message_chain: mirai.MessageChain,
|
||||
rule_dict: dict
|
||||
) -> entities.RuleJudgeResult:
|
||||
regexps = rule_dict['regexp']
|
||||
|
||||
for regexp in regexps:
|
||||
match = re.match(regexp, message_text)
|
||||
|
||||
if match:
|
||||
return entities.RuleJudgeResult(
|
||||
matching=True,
|
||||
replacement=message_chain,
|
||||
)
|
||||
|
||||
return entities.RuleJudgeResult(
|
||||
matching=False,
|
||||
replacement=message_chain
|
||||
)
|
||||
Loading…
Reference in New Issue
Block a user