diff --git a/.env.example b/.env.example index 647c5d2..59dbe2a 100644 --- a/.env.example +++ b/.env.example @@ -7,46 +7,46 @@ PROXY="http://127.0.0.1:7890" # ==================== # 行为设置 -# 使用.env.*中配置的NICKNAME作为图片上的 Bot 昵称（可空） +# 使用.env.*中配置的NICKNAME作为图片上的 Bot 昵称（可不填） PS_USE_ENV_NICK=False -# 分区列表里忽略的盘符（挂载点）（可空） +# 分区列表里忽略的盘符（挂载点）（可不填） # 使用正则表达式匹配 # 由于配置项使用JSON解析，所以需要使用双反斜杠转义， # 如："sda\\d" 解析为 sda\d（代表 sda<一位阿拉伯数字>）； # "C:\\\\Windows" 解析为 C:\\Windows（代表 C:\Windows） PS_IGNORE_PARTS=[] -# 忽略获取容量状态失败的磁盘分区（可空） +# 忽略获取容量状态失败的磁盘分区（可不填） PS_IGNORE_BAD_PARTS=False -# 是否排序分区列表（按照已用大小比例倒序）（可空） +# 是否排序分区列表（按照已用大小比例倒序）（可不填） PS_SORT_PARTS=True -# 是否反转分区列表排序（可空） +# 是否反转分区列表排序（可不填） PS_SORT_PARTS_REVERSE=False -# 磁盘 IO 统计列表中忽略的磁盘名（可空） +# 磁盘 IO 统计列表中忽略的磁盘名（可不填） # 使用正则表达式匹配（注意事项同上） PS_IGNORE_DISK_IOS=[] -# 是否忽略 IO 都为 0B/s 的磁盘（可空） +# 是否忽略 IO 都为 0B/s 的磁盘（可不填） PS_IGNORE_NO_IO_DISK=False -# 是否排序磁盘 IO 统计列表（按照读写速度总和倒序）（可空） +# 是否排序磁盘 IO 统计列表（按照读写速度总和倒序）（可不填） PS_SORT_DISK_IOS=True -# 网速列表中忽略的网络名称（可空） +# 网速列表中忽略的网络名称（可不填） # 使用正则表达式匹配（注意事项同上） PS_IGNORE_NETS=["^lo$", "^Loopback"] -# 是否忽略上下行都为 0B/s 的网卡（可空） +# 是否忽略上下行都为 0B/s 的网卡（可不填） PS_IGNORE_0B_NET=False -# 是否排序网速列表（按照上下行速度总和倒序）（可空） +# 是否排序网速列表（按照上下行速度总和倒序）（可不填） PS_SORT_NETS=True -# 需要进行测试响应速度的网址列表（可空） +# 需要进行测试响应速度的网址列表（可不填） PS_TEST_SITES=' [ {"name": "百度", "url": "https://www.baidu.com/"}, @@ -54,70 +54,79 @@ PS_TEST_SITES=' ] ' -# 是否将测试网址的结果排序（按照响应时间正序）（可空） +# 是否将测试网址的结果排序（按照响应时间正序）（可不填） PS_SORT_SITES=True -# 进程列表的最大项目数量（可空） +# 进程列表的最大项目数量（可不填） # 设为 0 可禁用显示 PS_PROC_LEN=5 -# 要忽略的进程名（可空） +# 要忽略的进程名（可不填） # 使用正则表达式匹配（注意事项同上） PS_IGNORE_PROCS=[] -# 进程列表的排序方式（可空） +# 进程列表的排序方式（可不填） # 可选：cpu、mem PS_PROC_SORT_BY=cpu -# 是否将进程 CPU 占用率显示为类似 Windows 任务管理器的百分比（最高 100%）（可空） +# 是否将进程 CPU 占用率显示为类似 Windows 任务管理器的百分比（最高 100%）（可不填） # 例：当你的 CPU 总共有 4 线程时，如果该进程吃满了两个线程， # Linux 会显示为 200%（每个线程算 100%），而 Windows 会显示为 50%（总占用率算 100%） PS_PROC_CPU_MAX_100P=False -# 网址测试访问时的超时时间（秒）（可空） +# 网址测试访问时的超时时间（秒）（可不填） PS_TEST_TIMEOUT=5 -# 请求头像等其他 URL 时的超时时间（秒）（可空） +# 请求头像等其他 URL 时的超时时间（秒）（可不填） PS_REQ_TIMEOUT=10 # ==================== # 个性化设置 -# 自定义字体路径（可空） -# 默认会自动选择系统内中文字体，这里填的是 Windows 自带的微软雅黑粗体 -PS_FONT="C:/Windows/Fonts/msjhbd.ttc" +# 自定义字体路径（可不填） +# 默认会自动选择系统内中文字体 +PS_FONT= -# 自定义背景图 URL 列表（可空） +# 自定义背景图 URL 列表（可不填） # 本地图请使用 file:///文件路径 PS_CUSTOM_BG=[] -# 各状态矩形背景底色（可空） +# 各状态矩形背景底色（可不填） PS_BG_COLOR=[255, 255, 255, 150] -# 背景图遮罩颜色（可空） +# 背景图遮罩颜色（可不填） PS_MASK_COLOR=[255, 255, 255, 125] -# 背景图高斯模糊半径（可空） +# 背景图高斯模糊半径（可不填） PS_BLUR_RADIUS=4 -# 底部脚注文字大小（可空） +# 底部脚注文字大小（可不填） PS_FOOTER_SIZE=22 -# 硬盘列表、网卡列表文本最长长度（可空） +# 硬盘列表、网卡列表文本最长长度（可不填） # 过长会被截断，长度包含 "..." 的长度 PS_MAX_TEXT_LEN=18 +# 默认头像路径（可不填） +PS_DEFAULT_AVATAR= + +# 默认背景图路径（可不填） +PS_DEFAULT_BG= + # ==================== # 平台相关设置 -# 是否只能由SuperUsers触发指令（可空） +# 触发插件功能的指令列表（可不填） +PS_COMMAND=["运行状态", "状态", "zt", "yxzt", "status"] + +# 是否只能由SuperUsers触发指令（可不填） PS_ONLY_SU=False -# 触发指令是否需要@Bot（可空） +# 触发指令是否需要@Bot（可不填） PS_NEED_AT=False -# 是否回复目标用户（可空） +# 是否回复目标用户（可不填） PS_REPLY_TARGET=True -# Telegram 平台接受自定义背景图的最大文件大小（单位 MB）（可空） +# Telegram 平台接受自定义背景图的最大文件大小（单位 MB）（可不填） PS_TG_MAX_FILE_SIZE=15 diff --git a/README.md b/README.md index d34ba35..1ff1099 100644 --- a/README.md +++ b/README.md @@ -152,6 +152,7 @@ Telegram：[@lgc2333](https://t.me/lgc2333) ### 0.5.5 - 一些不影响使用的小更改 +- 添加配置项 `PS_DEFAULT_AVATAR`、`PS_DEFAULT_BG`、`PS_COMMAND` ### 0.5.4 diff --git a/nonebot_plugin_picstatus/__main__.py b/nonebot_plugin_picstatus/__main__.py index 34c52b9..0b58824 100644 --- a/nonebot_plugin_picstatus/__main__.py +++ b/nonebot_plugin_picstatus/__main__.py @@ -1,110 +1,108 @@ -from typing import Optional - -from nonebot import logger, on_command -from nonebot.adapters import Bot, Event, Message -from nonebot.matcher import Matcher -from nonebot.params import CommandArg -from nonebot.permission import SUPERUSER -from nonebot.rule import Rule, to_me -from nonebot_plugin_saa import Image, MessageFactory, extract_target - -from .config import config -from .draw import get_stat_pic -from .util import async_request, download_telegram_file - -try: - from nonebot.adapters.onebot.v11 import MessageEvent as OBV11MessageEvent -except ImportError: - OBV11MessageEvent = None - -try: - from nonebot.adapters.telegram.event import MessageEvent as TGMessageEvent -except ImportError: - TGMessageEvent = None - - -def check_empty_arg_rule(arg: Message = CommandArg()): - return not arg.extract_plain_text() - - -def trigger_rule(): - rule = Rule(check_empty_arg_rule) - if config.ps_need_at: - rule = rule & to_me() - return rule - - -async def extract_msg_pic(bot: Bot, event: Event) -> Optional[bytes]: - if OBV11MessageEvent and isinstance(event, OBV11MessageEvent): - if (event.reply and (img := event.reply.message["image"])) or ( - img := event.message["image"] - ): - url = img[0].data["url"] - return await async_request(url) - - elif TGMessageEvent and isinstance(event, TGMessageEvent): - msg = event.message - if event.reply_to_message and event.reply_to_message.message: - msg += event.reply_to_message.message - - file_id = None - if photos := msg["photo"]: - file_id = photos[0].data["file"] - - elif documents := msg["document"]: - doc = next( - ( - doc - for doc in documents - if doc.data["mime_type"].startswith("image/") - ), - None, - ) - if doc: - data = doc.data - if data["file_size"] > (config.ps_tg_max_file_size * 1024 * 1024): - logger.warning("附带图片文件过大，回退到默认行为") - await MessageFactory("附带图片文件过大，将使用默认图片").send( - reply=config.ps_reply_target, - ) - else: - file_id = data["file_id"] - - if file_id: - return await download_telegram_file(bot, file_id) - - return None - - -_cmd, _alias = config.ps_command -stat_matcher = on_command( - _cmd, - aliases=set(_alias), - rule=trigger_rule(), - permission=SUPERUSER if config.ps_only_su else None, -) - - -@stat_matcher.handle() -async def _(bot: Bot, event: Event, matcher: Matcher): - try: - extract_target(event) - except RuntimeError: - logger.warning("SAA 不支持的平台，取消响应") - return - - pic = None - try: - pic = await extract_msg_pic(bot, event) - except Exception: - logger.exception("获取消息中附带图片失败，回退到默认行为") - await MessageFactory("获取消息中附带图片失败，将使用默认图片").send(reply=config.ps_reply_target) - - try: - ret = await get_stat_pic(bot, pic) - except Exception: - logger.exception("获取运行状态图失败") - await MessageFactory("获取运行状态图片失败，请检查后台输出").send(reply=config.ps_reply_target) - await matcher.finish() - - await MessageFactory(Image(ret)).finish(reply=config.ps_reply_target) +from typing import Optional + +from nonebot import logger, on_command +from nonebot.adapters import Bot, Event, Message +from nonebot.params import CommandArg +from nonebot.permission import SUPERUSER +from nonebot.rule import Rule, to_me +from nonebot_plugin_saa import Image, MessageFactory, extract_target + +from .config import config +from .draw import get_stat_pic +from .util import async_request, download_telegram_file + +try: + from nonebot.adapters.onebot.v11 import MessageEvent as OBV11MessageEvent +except ImportError: + OBV11MessageEvent = None + +try: + from nonebot.adapters.telegram.event import MessageEvent as TGMessageEvent +except ImportError: + TGMessageEvent = None + + +def check_empty_arg_rule(arg: Message = CommandArg()): + return not arg.extract_plain_text() + + +def trigger_rule(): + rule = Rule(check_empty_arg_rule) + if config.ps_need_at: + rule = rule & to_me() + return rule + + +async def extract_msg_pic(bot: Bot, event: Event) -> Optional[bytes]: + if OBV11MessageEvent and isinstance(event, OBV11MessageEvent): + if (event.reply and (img := event.reply.message["image"])) or ( + img := event.message["image"] + ): + url = img[0].data["url"] + return await async_request(url) + + elif TGMessageEvent and isinstance(event, TGMessageEvent): + msg = event.message + if event.reply_to_message and event.reply_to_message.message: + msg += event.reply_to_message.message + + file_id = None + if photos := msg["photo"]: + file_id = photos[0].data["file"] + + elif documents := msg["document"]: + doc = next( + ( + doc + for doc in documents + if doc.data["mime_type"].startswith("image/") + ), + None, + ) + if doc: + data = doc.data + if data["file_size"] > (config.ps_tg_max_file_size * 1024 * 1024): + logger.warning("附带图片文件过大，回退到默认行为") + await MessageFactory("附带图片文件过大，将使用默认图片").send( + reply=config.ps_reply_target, + ) + else: + file_id = data["file_id"] + + if file_id: + return await download_telegram_file(bot, file_id) + + return None + + +_cmd, _alias = config.ps_command +stat_matcher = on_command( + _cmd, + aliases=set(_alias), + rule=trigger_rule(), + permission=SUPERUSER if config.ps_only_su else None, +) + + +@stat_matcher.handle() +async def _(bot: Bot, event: Event): + try: + extract_target(event) + except RuntimeError: + logger.warning("SAA 不支持的平台，取消响应") + return + + pic = None + try: + pic = await extract_msg_pic(bot, event) + except Exception: + logger.exception("获取消息中附带图片失败，回退到默认行为") + await MessageFactory("获取消息中附带图片失败，将使用默认图片").send(reply=config.ps_reply_target) + + try: + ret = await get_stat_pic(bot, pic) + except Exception: + logger.exception("获取运行状态图失败") + await MessageFactory("获取运行状态图片失败，请检查后台输出").finish(reply=config.ps_reply_target) + + await MessageFactory(Image(ret)).finish(reply=config.ps_reply_target) diff --git a/nonebot_plugin_picstatus/config.py b/nonebot_plugin_picstatus/config.py index 2c29bd8..2b184e7 100644 --- a/nonebot_plugin_picstatus/config.py +++ b/nonebot_plugin_picstatus/config.py @@ -4,7 +4,9 @@ from nonebot import get_driver from pydantic import BaseModel -from .const import DEFAULT_AVATAR_PATH, DEFAULT_BG_PATH +RES_PATH = Path(__file__).parent / "res" +DEFAULT_BG_PATH = RES_PATH / "default_bg.webp" +DEFAULT_AVATAR_PATH = RES_PATH / "default_avatar.webp" class TestSiteCfg(BaseModel): @@ -30,7 +32,7 @@ class Cfg(BaseModel): ps_ignore_disk_ios: List[str] = [] ps_ignore_no_io_disk: bool = False ps_sort_disk_ios: bool = True - ps_ignore_nets: List[str] = ["^lo$", "^Loopback"] + ps_ignore_nets: List[str] = [r"^lo(op)?\d*$", "^Loopback"] ps_ignore_0b_net: bool = False ps_sort_nets: bool = True ps_test_sites: List[TestSiteCfg] = [ diff --git a/nonebot_plugin_picstatus/const.py b/nonebot_plugin_picstatus/const.py deleted file mode 100644 index 097ca2e..0000000 --- a/nonebot_plugin_picstatus/const.py +++ /dev/null @@ -1,5 +0,0 @@ -from pathlib import Path - -RES_PATH = Path(__file__).parent / "res" -DEFAULT_BG_PATH = RES_PATH / "default_bg.png" -DEFAULT_AVATAR_PATH = RES_PATH / "default_avatar.png" diff --git a/nonebot_plugin_picstatus/draw.py b/nonebot_plugin_picstatus/draw.py index 2f12343..9a40e33 100644 --- a/nonebot_plugin_picstatus/draw.py +++ b/nonebot_plugin_picstatus/draw.py @@ -478,12 +478,11 @@ async def get_stat_pic(bot: Bot, bg_arg: Optional[bytes] = None) -> bytes: ) bg = ret[0] - ret = ret[1:] + ret = [x for x in ret[1:] if x] # 统计图片高度 for p in ret: - if p: - img_h += p.size[1] + 50 + img_h += p.size[1] + 50 # 居中裁剪背景 bg = bg.convert("RGBA") diff --git a/nonebot_plugin_picstatus/res/default_avatar.png b/nonebot_plugin_picstatus/res/default_avatar.png deleted file mode 100644 index 3d9cc96..0000000 Binary files a/nonebot_plugin_picstatus/res/default_avatar.png and /dev/null differ diff --git a/nonebot_plugin_picstatus/res/default_avatar.webp b/nonebot_plugin_picstatus/res/default_avatar.webp new file mode 100644 index 0000000..00419fc Binary files /dev/null and b/nonebot_plugin_picstatus/res/default_avatar.webp differ diff --git a/nonebot_plugin_picstatus/res/default_bg.png b/nonebot_plugin_picstatus/res/default_bg.png deleted file mode 100644 index bb93cba..0000000 Binary files a/nonebot_plugin_picstatus/res/default_bg.png and /dev/null differ diff --git a/nonebot_plugin_picstatus/res/default_bg.webp b/nonebot_plugin_picstatus/res/default_bg.webp new file mode 100644 index 0000000..6ac7eeb Binary files /dev/null and b/nonebot_plugin_picstatus/res/default_bg.webp differ diff --git a/nonebot_plugin_picstatus/util.py b/nonebot_plugin_picstatus/util.py index 28b0ced..64e6b62 100644 --- a/nonebot_plugin_picstatus/util.py +++ b/nonebot_plugin_picstatus/util.py @@ -1,153 +1,153 @@ -import json -import re -from datetime import timedelta -from io import BytesIO -from pathlib import Path -from typing import List, Literal, Optional, Union, cast, overload - -import anyio -from httpx import AsyncClient -from nonebot import logger -from nonebot.adapters import Bot -from PIL import Image - -from .config import config - -try: - from nonebot.adapters.onebot.v11 import Bot as OBV11Bot -except ImportError: - OBV11Bot = None - -try: - from nonebot.adapters.telegram import Bot as TGBot -except ImportError: - TGBot = None - - -def format_timedelta(t: timedelta): - mm, ss = divmod(t.seconds, 60) - hh, mm = divmod(mm, 60) - s = f"{hh}:{mm:02d}:{ss:02d}" - if t.days: - s = f"{t.days}天 {s}" - # if t.microseconds: - # s += f" {t.microseconds / 1000:.3f}毫秒" - return s - - -@overload -async def async_request( - url: str, - *args, - is_text: Literal[False] = False, - proxy: Optional[str] = None, - **kwargs, -) -> bytes: - ... - - -@overload -async def async_request( - url: str, - *args, - is_text: Literal[True] = True, - proxy: Optional[str] = None, - **kwargs, -) -> str: - ... - - -async def async_request(url: str, *args, is_text=False, proxy=None, **kwargs): - async with AsyncClient( - proxies=proxy, - follow_redirects=True, - timeout=config.ps_req_timeout, - ) as cli: - res = await cli.get(url, *args, **kwargs) - return res.text if is_text else res.content - - -async def get_anime_pic() -> bytes: - data = json.loads( - await async_request("https://api.gumengya.com/Api/DmImg", is_text=True), - ) - assert str(data.get("code")) == "200" - return await async_request(data["data"]["url"]) - - -async def async_open_img( - fp: Union[str, Path, anyio.Path], - *args, - **kwargs, -) -> Image.Image: - p = BytesIO(await anyio.Path(fp).read_bytes()) - return Image.open(p, *args, **kwargs) - - -def format_byte_count(b: int) -> str: - units = ["B", "K", "M", "G", "T", "P", "E", "Z", "Y"] - multiplier = 1024 - - value = b - for unit in units: - if value < multiplier: - return f"{value:.2f}{unit}" - value /= multiplier - - return f"{value:.2f}{units[-1]}" - - -def match_list_regexp(reg_list: List[str], txt: str) -> Optional[re.Match]: - return next((match for r in reg_list if (match := re.search(r, txt))), None) - - -def process_text_len(text: str) -> str: - real_max_len = config.ps_max_text_len - 3 - if len(text) > real_max_len: - text = f"{text[:real_max_len]}..." - return text - - -async def download_telegram_file(bot: Bot, file_id: str) -> bytes: - assert TGBot and isinstance(bot, TGBot), "仅 Telegram Bot 可调用此函数" - - res = await bot.get_file(file_id=file_id) - file_path = cast(str, res.file_path) - - if await (p := anyio.Path(file_path)).exists(): - return await p.read_bytes() - - url = f"{bot.bot_config.api_server}file/bot{bot.bot_config.token}/{file_path}" - return await async_request(url, proxy=config.proxy) - - -async def get_qq_avatar(qq) -> bytes: - return await async_request(f"https://q2.qlogo.cn/headimg_dl?dst_uin={qq}&spec=640") - - -async def get_tg_avatar(bot: Bot) -> bytes: - assert TGBot and isinstance(bot, TGBot), "仅 Telegram Bot 可调用此函数" - - res = await bot.get_user_profile_photos(user_id=int(bot.self_id), limit=1) - file_id = res.photos[0][-1].file_id - - return await download_telegram_file(bot, file_id) - - -async def get_bot_avatar(bot: Bot) -> Image.Image: - avatar = None - - try: - if OBV11Bot and isinstance(bot, OBV11Bot): - avatar = await get_qq_avatar(bot.self_id) - elif TGBot and isinstance(bot, TGBot): - avatar = await get_tg_avatar(bot) - else: - logger.info("暂不支持获取该平台Bot头像，使用默认头像替代") - except Exception: - logger.exception("获取Bot头像失败，使用默认头像替代") - - if avatar: - return Image.open(BytesIO(avatar)) - - return await async_open_img(config.ps_default_avatar) +import json +import re +from datetime import timedelta +from io import BytesIO +from pathlib import Path +from typing import List, Literal, Optional, Union, cast, overload + +import anyio +from httpx import AsyncClient +from nonebot import logger +from nonebot.adapters import Bot +from PIL import Image + +from .config import config + +try: + from nonebot.adapters.onebot.v11 import Bot as OBV11Bot +except ImportError: + OBV11Bot = None + +try: + from nonebot.adapters.telegram import Bot as TGBot +except ImportError: + TGBot = None + + +def format_timedelta(t: timedelta): + mm, ss = divmod(t.seconds, 60) + hh, mm = divmod(mm, 60) + s = f"{hh}:{mm:02d}:{ss:02d}" + if t.days: + s = f"{t.days}天 {s}" + # if t.microseconds: + # s += f" {t.microseconds / 1000:.3f}毫秒" + return s + + +@overload +async def async_request( + url: str, + *args, + is_text: Literal[False] = False, + proxy: Optional[str] = None, + **kwargs, +) -> bytes: + ... + + +@overload +async def async_request( + url: str, + *args, + is_text: Literal[True] = True, + proxy: Optional[str] = None, + **kwargs, +) -> str: + ... + + +async def async_request(url: str, *args, is_text=False, proxy=None, **kwargs): + async with AsyncClient( + proxies=proxy, + follow_redirects=True, + timeout=config.ps_req_timeout, + ) as cli: + res = await cli.get(url, *args, **kwargs) + return res.text if is_text else res.content + + +async def get_anime_pic() -> bytes: + data = json.loads( + await async_request("https://api.gumengya.com/Api/DmImg", is_text=True), + ) + assert str(data.get("code")) == "200" + return await async_request(data["data"]["url"]) + + +async def async_open_img( + fp: Union[str, Path, anyio.Path], + *args, + **kwargs, +) -> Image.Image: + p = BytesIO(await anyio.Path(fp).read_bytes()) + return Image.open(p, *args, **kwargs) + + +def format_byte_count(b: int) -> str: + units = ["B", "K", "M", "G", "T", "P", "E", "Z", "Y"] + multiplier = 1024 + + value = b + for unit in units: + if value < multiplier: + return f"{value:.2f}{unit}" + value /= multiplier + + return f"{value:.2f}{units[-1]}" + + +def match_list_regexp(reg_list: List[str], txt: str) -> Optional[re.Match]: + return next((match for r in reg_list if (match := re.search(r, txt))), None) + + +def process_text_len(text: str) -> str: + real_max_len = config.ps_max_text_len - 3 + if len(text) > real_max_len: + text = f"{text[:real_max_len]}..." + return text + + +async def download_telegram_file(bot: Bot, file_id: str) -> bytes: + assert TGBot and isinstance(bot, TGBot), "仅 Telegram Bot 可调用此函数" + + res = await bot.get_file(file_id=file_id) + file_path = cast(str, res.file_path) + + if await (p := anyio.Path(file_path)).exists(): + return await p.read_bytes() + + url = f"{bot.bot_config.api_server}file/bot{bot.bot_config.token}/{file_path}" + return await async_request(url, proxy=config.proxy) + + +async def get_qq_avatar(qq) -> bytes: + return await async_request(f"https://q2.qlogo.cn/headimg_dl?dst_uin={qq}&spec=640") + + +async def get_tg_avatar(bot: Bot) -> bytes: + assert TGBot and isinstance(bot, TGBot), "仅 Telegram Bot 可调用此函数" + + res = await bot.get_user_profile_photos(user_id=int(bot.self_id), limit=1) + file_id = res.photos[0][-1].file_id + + return await download_telegram_file(bot, file_id) + + +async def get_bot_avatar(bot: Bot) -> Image.Image: + avatar = None + + try: + if OBV11Bot and isinstance(bot, OBV11Bot): + avatar = await get_qq_avatar(bot.self_id) + elif TGBot and isinstance(bot, TGBot): + avatar = await get_tg_avatar(bot) + else: + logger.info("暂不支持获取该平台Bot头像，使用默认头像替代") + except Exception: + logger.exception("获取Bot头像失败，使用默认头像替代") + + if avatar: + return Image.open(BytesIO(avatar)) + + return await async_open_img(config.ps_default_avatar)