可针对不同平台选择发送图片类型为linkbase64link_remotelink_local

This commit is contained in:
KimigaiiWuyi 2023-11-19 03:44:10 +08:00
parent bc3fb49938
commit fa2054d15c
8 changed files with 144 additions and 87 deletions

View File

@ -11,9 +11,13 @@ from gsuid_core.models import Event, Message, MessageSend
from gsuid_core.utils.plugins_config.gs_config import core_plugins_config from gsuid_core.utils.plugins_config.gs_config import core_plugins_config
from gsuid_core.segment import MessageSegment, to_markdown, convert_message from gsuid_core.segment import MessageSegment, to_markdown, convert_message
is_sp_msg_id: str = core_plugins_config.get_config('EnableSpecificMsgId').data
sp_msg_id: str = core_plugins_config.get_config('SpecificMsgId').data sp_msg_id: str = core_plugins_config.get_config('SpecificMsgId').data
is_markdown: List = core_plugins_config.get_config('SendMDPlatform').data is_sp_msg_id: str = core_plugins_config.get_config('EnableSpecificMsgId').data
ism: List = core_plugins_config.get_config('SendMDPlatform').data
isb: List = core_plugins_config.get_config('SendButtonsPlatform').data
enable_buttons_platform = isb
enable_markdown_platform = ism
class _Bot: class _Bot:
@ -36,9 +40,9 @@ class _Bot:
sender_id: str = '', sender_id: str = '',
group_id: Optional[str] = None, group_id: Optional[str] = None,
): ):
_message = await convert_message(message) _message = await convert_message(message, bot_id)
if bot_id in is_markdown: if bot_id in enable_markdown_platform:
_message = await to_markdown(_message) _message = await to_markdown(_message)
if at_sender and sender_id: if at_sender and sender_id:
@ -160,9 +164,9 @@ class Bot:
if reply is None: if reply is None:
reply = f'请在{timeout}秒内做出选择...' reply = f'请在{timeout}秒内做出选择...'
_reply = await convert_message(reply) _reply = await convert_message(reply, self.bot_id)
if self.ev.real_bot_id in ['qqgroup']: if self.ev.real_bot_id in enable_buttons_platform:
_buttons = [] _buttons = []
for option in option_list: for option in option_list:
if isinstance(option, List): if isinstance(option, List):

View File

@ -49,7 +49,7 @@ class GsClient:
else: else:
content = [Message(type='text', data=intent)] content = [Message(type='text', data=intent)]
msg = MessageReceive( msg = MessageReceive(
bot_id='Nonebot222', bot_id='console',
user_type='direct', user_type='direct',
user_pm=1, user_pm=1,
group_id=None, group_id=None,

View File

@ -59,7 +59,7 @@ async def get_help(
font: Callable[[int], ImageFont.FreeTypeFont], font: Callable[[int], ImageFont.FreeTypeFont],
is_dark: bool = True, is_dark: bool = True,
text_color: Tuple[int, int, int] = (250, 250, 250), text_color: Tuple[int, int, int] = (250, 250, 250),
sub_color: Optional[Tuple[int, int, int]] = None, sub_c: Optional[Tuple[int, int, int]] = None,
op_color: Optional[Tuple[int, int, int]] = None, op_color: Optional[Tuple[int, int, int]] = None,
column: int = 5, column: int = 5,
is_gaussian: bool = False, is_gaussian: bool = False,
@ -74,15 +74,15 @@ async def get_help(
if help_path.exists() and name in cache and cache[name] and enable_cache: if help_path.exists() and name in cache and cache[name] and enable_cache:
return await convert_img(Image.open(help_path)) return await convert_img(Image.open(help_path))
if sub_color is None and is_dark: if sub_c is None and is_dark:
sub_color = tuple(x - 50 for x in text_color if x > 50) sub_c = tuple(x - 50 for x in text_color if x > 50) # type: ignore
elif sub_color is None and not is_dark: elif sub_c is None and not is_dark:
sub_color = tuple(x + 50 for x in text_color if x < 205) sub_c = tuple(x + 50 for x in text_color if x < 205) # type: ignore
if op_color is None and is_dark: if op_color is None and is_dark:
op_color = tuple(x - 90 for x in text_color if x > 90) op_color = tuple(x - 90 for x in text_color if x > 90) # type: ignore
elif op_color is None and not is_dark: elif op_color is None and not is_dark:
op_color = tuple(x + 90 for x in text_color if x < 160) op_color = tuple(x + 90 for x in text_color if x < 160) # type: ignore
_h = 600 _h = 600
@ -109,12 +109,10 @@ async def get_help(
for _i, message in enumerate(extra_message): for _i, message in enumerate(extra_message):
_x = int(first_x + _i * 300) _x = int(first_x + _i * 300)
title.paste(badge_s, (_x, 556), badge_s) title.paste(badge_s, (_x, 556), badge_s)
title_draw.text( title_draw.text((_x + 360, 596), message, sub_c, font(26), 'mm')
(_x + 360, 596), message, sub_color, font(26), 'mm'
)
title_draw.text((cx(w, 0), 440), f'{name} 帮助', text_color, font(36), 'mm') title_draw.text((cx(w, 0), 440), f'{name} 帮助', text_color, font(36), 'mm')
title_draw.text((cx(w, 0), 520), sub_text, sub_color, font(26), 'mm') title_draw.text((cx(w, 0), 520), sub_text, sub_c, font(26), 'mm')
if is_dark: if is_dark:
icon_mask = Image.new('RGBA', (36, 36), (255, 255, 255)) icon_mask = Image.new('RGBA', (36, 36), (255, 255, 255))
@ -142,7 +140,7 @@ async def get_help(
bbox = font(35).getbbox(sv_name) bbox = font(35).getbbox(sv_name)
size, _ = bbox[2] - bbox[0], bbox[3] - bbox[1] size, _ = bbox[2] - bbox[0], bbox[3] - bbox[1]
bc_draw.text((42 + size, 30), sv_desc, sub_color, font(20), 'lm') bc_draw.text((42 + size, 30), sv_desc, sub_c, font(20), 'lm')
sv_img.paste(bc, (0, 10), bc) sv_img.paste(bc, (0, 10), bc)
# sv_img = easy_alpha_composite(sv_img, bc, (0, 10)) # sv_img = easy_alpha_composite(sv_img, bc, (0, 10))
@ -169,7 +167,7 @@ async def get_help(
# 标题 # 标题
bt_draw.text((20 + f, 28), tr_name, text_color, font(26), 'lm') bt_draw.text((20 + f, 28), tr_name, text_color, font(26), 'lm')
# 使用范例 # 使用范例
bt_draw.text((20 + f, 50), tr['eg'], sub_color, font(17), 'lm') bt_draw.text((20 + f, 50), tr['eg'], sub_c, font(17), 'lm')
# 简单介绍 # 简单介绍
bt_draw.text((20, 78), tr['desc'], op_color, font(16), 'lm') bt_draw.text((20, 78), tr['desc'], op_color, font(16), 'lm')

View File

@ -13,6 +13,7 @@ from gsuid_core.data_store import image_res
from gsuid_core.message_models import Button from gsuid_core.message_models import Button
from gsuid_core.utils.image.convert import text2pic from gsuid_core.utils.image.convert import text2pic
from gsuid_core.utils.plugins_config.gs_config import ( from gsuid_core.utils.plugins_config.gs_config import (
send_pic_config,
pic_upload_config, pic_upload_config,
core_plugins_config, core_plugins_config,
) )
@ -68,13 +69,7 @@ class MessageSegment:
with open(img, 'rb') as fp: with open(img, 'rb') as fp:
img = fp.read() img = fp.read()
if enable_pic_srv: data = f'base64://{b64encode(img).decode()}'
name = f'{uuid.uuid1()}.jpg'
path = image_res / name
path.write_bytes(img)
data = f'link://{pic_srv}/genshinuid/image/{name}'
else:
data = f'base64://{b64encode(img).decode()}'
msg = Message(type='image', data=data) msg = Message(type='image', data=data)
return msg return msg
@ -174,35 +169,93 @@ class MessageSegment:
return Message(type=f'log_{type}', data=content) return Message(type=f'log_{type}', data=content)
async def _conver_image_to_url(message: Message) -> List[Message]: async def _conver_image_to_url(image: Union[bytes, str]) -> List[Message]:
if pclient is not None: if pclient is not None:
img_text: str = message.data # type: ignore if isinstance(image, str) and image.startswith('base64://'):
if img_text.startswith('base64://'): image_bytes = b64decode(image[9:])
bio = BytesIO(b64decode(img_text[9:])) else:
img = Image.open(bio) image_bytes = image
img_url = await pclient.upload(f'{uuid.uuid4()}.jpg', bio) assert isinstance(image_bytes, bytes)
_message = [
MessageSegment.image(img_url if img_url else img_text), bio = BytesIO(image_bytes)
MessageSegment.image_size(img.size), img = Image.open(bio)
] img_url = await pclient.upload(f'{uuid.uuid4()}.jpg', bio)
return _message _message = [
return [message] MessageSegment.image(img_url if img_url else image_bytes),
MessageSegment.image_size(img.size),
]
return _message
return []
async def _convert_message_to_image(
message: Message, bot_id: str
) -> List[Message]:
if message.data is None:
return []
image_b64 = None
if (
message.type == 'text'
and is_text2pic
and len(message.data) >= int(text2pic_limit)
):
image_bytes = await text2pic(message.data)
elif message.type == 'image':
img: Union[bytes, str] = message.data
if isinstance(img, str) and img.startswith('base64://'):
image_b64 = img
image_bytes = b64decode(img[9:])
else:
image_bytes = img
else:
return [message]
assert isinstance(image_bytes, bytes)
send_type = send_pic_config.get_config(bot_id).data
if send_type == 'base64':
return (
[Message(type='image', data=image_b64)]
if image_b64
else [MessageSegment.image(image_bytes)]
)
if send_type == 'link_remote':
return await _conver_image_to_url(image_bytes)
elif (send_type == 'link_local') or enable_pic_srv:
bio = BytesIO(image_bytes)
image = Image.open(bio)
name = f'{uuid.uuid1()}.jpg'
path = image_res / name
path.write_bytes(image_bytes)
data = f'link://{pic_srv}/genshinuid/image/{name}'
return [
Message(type='image', data=data),
MessageSegment.image_size(image.size),
]
elif pclient is not None:
return await _conver_image_to_url(image_bytes)
else:
return [message]
async def _convert_message( async def _convert_message(
message: Union[Message, str, bytes] message: Union[Message, str, bytes], bot_id: str
) -> List[Message]: ) -> List[Message]:
_message = [message]
if isinstance(message, Message): if isinstance(message, Message):
if message.data is None: if message.data is None:
return [message] return [message]
if message.type == 'image': if message.type == 'image':
_message = await _conver_image_to_url(message) _message = await _convert_message_to_image(message, bot_id)
elif message.type == 'node': elif message.type == 'node':
_temp = [] _temp = []
for i in message.data: for i in message.data:
if i.type == 'image': if i.type == 'image':
_temp.extend(await _conver_image_to_url(i)) _temp.extend(await _convert_message_to_image(i, bot_id))
else: else:
_temp.append(i) _temp.append(i)
_message = [MessageSegment.node(_temp)] _message = [MessageSegment.node(_temp)]
@ -210,51 +263,35 @@ async def _convert_message(
_message = [message] _message = [message]
elif isinstance(message, str): elif isinstance(message, str):
if message.startswith('base64://'): if message.startswith('base64://'):
bio = BytesIO(b64decode(message[9:])) _str_message = Message(type='image', data=message)
img = Image.open(bio)
img_url = None
if pclient is not None:
img_url = await pclient.upload(f'{uuid.uuid4()}.jpg', bio)
_message = [
MessageSegment.image(img_url if img_url else message),
MessageSegment.image_size(img.size),
]
else: else:
_message = [MessageSegment.text(message)] _str_message = MessageSegment.text(message)
elif isinstance(message, bytes): _message = await _convert_message_to_image(_str_message, bot_id)
bio = BytesIO(message) elif isinstance(message, (bytes, bytearray, memoryview)):
img = Image.open(bio) message = bytes(message)
_bytes_message = Message(type='image', data=message)
img_url = None _message = await _convert_message_to_image(_bytes_message, bot_id)
if pclient is not None: return _message
img_url = await pclient.upload(f'{uuid.uuid4()}.jpg', bio)
_message = [
MessageSegment.image(img_url if img_url else message),
MessageSegment.image_size(img.size),
]
elif isinstance(message, (bytearray, memoryview)):
_message = [MessageSegment.image(bytes(message))]
else:
_message = [message]
return _message # type: ignore
async def convert_message( async def convert_message(
message: Union[Message, List[Message], List[str], str, bytes] message: Union[Message, List[Message], List[str], str, bytes], bot_id: str
) -> List[Message]: ) -> List[Message]:
# 转换消息类型为bot标准输出类型
_message: List[Message] = [] _message: List[Message] = []
if isinstance(message, List): if isinstance(message, List):
# 如果要转换的消息类型为列表且全都是string则作为合并转发消息发送
if all(isinstance(x, str) for x in message): if all(isinstance(x, str) for x in message):
_message.extend([MessageSegment.node(message)]) _message.extend([MessageSegment.node(message)])
else: else:
# 如果不是,则针对每条消息都进行转换
for i in message: for i in message:
_message.extend(await _convert_message(i)) _message.extend(await _convert_message(i, bot_id))
else: else:
_message = await _convert_message(message) _message = await _convert_message(message, bot_id)
# 启用了随机字符的话,随机加入字符
if R_enabled: if R_enabled:
result = ''.join( result = ''.join(
random.choice(R_text) random.choice(R_text)
@ -262,16 +299,6 @@ async def convert_message(
) )
_message.append(MessageSegment.text(result)) _message.append(MessageSegment.text(result))
if is_text2pic:
if (
len(_message) == 1
and _message[0].type == 'text'
and isinstance(_message[0].data, str)
and len(_message[0].data) >= int(text2pic_limit)
):
img = await text2pic(_message[0].data)
_message = [MessageSegment.image(img)]
return _message return _message

View File

@ -3,6 +3,7 @@ import json
from pathlib import Path from pathlib import Path
from openpyxl import load_workbook from openpyxl import load_workbook
from openpyxl.worksheet.worksheet import Worksheet
sample = { sample = {
'name': '', 'name': '',
@ -20,6 +21,7 @@ OUTPUT_PATH = Path(__file__).parent / 'Help.json'
wb = load_workbook(str(HELP_PATH)) wb = load_workbook(str(HELP_PATH))
ws = wb.active ws = wb.active
assert isinstance(ws, Worksheet)
module_name_str = '' module_name_str = ''
for row in range(2, 999): for row in range(2, 999):

View File

@ -62,5 +62,8 @@ CONIFG_DEFAULT: Dict[str, GSC] = {
'EnablePicSrv': GsBoolConfig('启用将图片转链接发送(需公网)', '发送图片转链接', False), 'EnablePicSrv': GsBoolConfig('启用将图片转链接发送(需公网)', '发送图片转链接', False),
'PicSrv': GsStrConfig('图片转链接为(需公网)', '发送图片转链接', ''), 'PicSrv': GsStrConfig('图片转链接为(需公网)', '发送图片转链接', ''),
'ProxyURL': GsStrConfig('安装插件时使用git代理地址', 'git代理地址', ''), 'ProxyURL': GsStrConfig('安装插件时使用git代理地址', 'git代理地址', ''),
'SendMDPlatform': GsListStrConfig('发送MD的平台列表(用:连接)', '发送MD的平台列表', []), 'SendMDPlatform': GsListStrConfig('默认发送MD的平台列表(用:连接)', '发送MD的平台列表', []),
'SendButtonsPlatform': GsListStrConfig(
'默认发送按钮的平台列表(用:连接)', '发送按钮的平台列表', []
),
} }

View File

@ -8,6 +8,7 @@ from gsuid_core.data_store import get_res_path
from .models import GSC, GsBoolConfig from .models import GSC, GsBoolConfig
from .config_default import CONIFG_DEFAULT from .config_default import CONIFG_DEFAULT
from .send_pic_config import SEND_PIC_CONIFG
from .pic_server_config import PIC_UPLOAD_CONIFG from .pic_server_config import PIC_UPLOAD_CONIFG
@ -121,5 +122,9 @@ core_plugins_config = StringConfig(
) )
pic_upload_config = StringConfig( pic_upload_config = StringConfig(
'Core_Pic', get_res_path() / 'pic_upload_config.json', PIC_UPLOAD_CONIFG 'GsCore图片上传', get_res_path() / 'pic_upload_config.json', PIC_UPLOAD_CONIFG
)
send_pic_config = StringConfig(
'GsCore发送图片', get_res_path() / 'send_pic_config.json', SEND_PIC_CONIFG
) )

View File

@ -0,0 +1,18 @@
from typing import Dict
from .models import GSC, GsStrConfig
SEND_PIC_CONIFG: Dict[str, GSC] = {
'onebot': GsStrConfig('OneBot图片发送方式', '可选link或base64', 'base64'),
'onebotv12': GsStrConfig('OneBot V12图片发送方式', '可选link或base64', 'base64'),
'qqguild': GsStrConfig('QQ Guild图片发送方式', '可选link或base64', 'base64'),
'qqgroup': GsStrConfig('QQ Group图片发送方式', '可选link或base64', 'link'),
'telegram': GsStrConfig('Telegram图片发送方式', '可选link或base64', 'base64'),
'discord': GsStrConfig('Discord图片发送方式', '可选link或base64', 'base64'),
'kook': GsStrConfig('KOOK图片发送方式', '可选link或base64', 'base64'),
'dodo': GsStrConfig('DoDo图片发送方式', '可选link或base64', 'base64'),
'feishu': GsStrConfig('飞书图片发送方式', '可选link或base64', 'base64'),
'ntchat': GsStrConfig('NtChat图片发送方式', '可选link或base64', 'base64'),
'villa': GsStrConfig('米游社大别野图片发送方式', '可选link或base64', 'base64'),
'console': GsStrConfig('本地client.py图片发送方式', '可选link或base64', 'link_local'),
}