Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2025-12-15 06:12:53 +08:00
Code Issues Actions 7 Packages Projects Releases Wiki Activity

fix bug

Browse Source
This commit is contained in:
HibiKier 2022-05-21 13:15:53 +08:00
parent 86a8fe7096
commit 188ae38525
25 changed files with 153 additions and 124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
README.md
View File

@ -238,6 +238,23 @@ __Docker 最新版本由 [Sakuracio](https://github.com/Sakuracio) 提供__
## 更新
### 2022/5/
* 修复搜番无结果时报错无正确反馈
* 解锁了windows上无法使用playwright的限制
* 修复p搜对应pid有多张图时出错，改为连续发送图片
* 修复p搜对数字的错误判断
* 修复添加商品折扣无法正确添加
* 修复了bilibili订阅直播间订阅up名称不一致的问题
* 修复原神玩家查询没开地图时报错
* 最低priority修改为 999
### 2022/5/19
* fix: mihoyo bbs api changed [@pull/357](https://github.com/HibiKier/zhenxun_bot/pull/357)
* Add word_clouds [@pull/265](https://github.com/HibiKier/zhenxun_bot/pull/265)
* Fix wrong live streamer name [@pull/284](https://github.com/HibiKier/zhenxun_bot/pull/284)
### 2022/5/16
* 词条支持图片和@问题 [@pull/160](https://github.com/HibiKier/zhenxun_bot/pull/160)
@ -511,4 +528,4 @@ __..... 更多更新信息请查看文档__
[NothAmor / nonebot2_luxun_says](https://github.com/NothAmor/nonebot2_luxun_says) ：基于nonebot2机器人框架的鲁迅说插件
[Kyomotoi / AnimeThesaurus](https://github.com/Kyomotoi/AnimeThesaurus) ：一个~~特二刺螈~~（文爱）的适用于任何bot的词库
[Ailitonia / omega-miya](https://github.com/Ailitonia/omega-miya) ：基于nonebot2的qq机器人
[KimigaiiWuyi / GenshinUID]("https://github.com/KimigaiiWuyi/GenshinUID") ：一个基于HoshinoBot/NoneBot2的原神UID查询插件
[KimigaiiWuyi / GenshinUID](https://github.com/KimigaiiWuyi/GenshinUID) ：一个基于HoshinoBot/NoneBot2的原神UID查询插件

4
basic_plugins/hooks/auth_hook.py
View File

@ -106,7 +106,7 @@ async def _(matcher: Matcher, bot: Bot, event: Event, state: T_State):
raise IgnoredException("功能总开关关闭状态")
except AttributeError:
pass
if module in admin_manager.keys() and matcher.priority not in [1, 9]:
if module in admin_manager.keys() and matcher.priority not in [1, 999]:
if isinstance(event, GroupMessageEvent):
# 个人权限
if (
@ -146,7 +146,7 @@ async def _(matcher: Matcher, bot: Bot, event: Event, state: T_State):
if event.is_tome():
status_message_manager.add(event.user_id)
raise IgnoredException("权限不足")
if module in plugins2info_dict.keys() and matcher.priority not in [1, 9]:
if module in plugins2info_dict.keys() and matcher.priority not in [1, 999]:
# 戳一戳单独判断
if isinstance(event, GroupMessageEvent) or isinstance(event, PokeNotifyEvent) or matcher.plugin_name in other_limit_plugins:
if status_message_manager.get(event.group_id) is None:

6
basic_plugins/hooks/ban_hook.py
View File

@ -31,7 +31,7 @@ _flmt = FreqLimiter(300)
async def _(matcher: Matcher, bot: Bot, event: Event, state: T_State):
if (
(isinstance(event, MessageEvent) or isinstance(event, PokeNotifyEvent))
and matcher.priority not in [1, 9]
and matcher.priority not in [1, 999]
) or matcher.plugin_name in other_limit_plugins:
try:
if (
@ -58,7 +58,7 @@ async def _(matcher: Matcher, bot: Bot, event: Event, state: T_State):
if not static_flmt.check(event.user_id):
raise IgnoredException("用户处于黑名单中")
static_flmt.start_cd(event.user_id)
if matcher.priority != 9:
if matcher.priority != 999:
try:
ban_result = Config.get_config("hook", "BAN_RESULT")
if (
@ -79,7 +79,7 @@ async def _(matcher: Matcher, bot: Bot, event: Event, state: T_State):
if not static_flmt.check(event.user_id):
raise IgnoredException("用户处于黑名单中")
static_flmt.start_cd(event.user_id)
if matcher.priority != 9:
if matcher.priority != 999:
try:
ban_result = Config.get_config("hook", "BAN_RESULT")
if ban_result and matcher.plugin_name not in ignore_rst_module:

2
basic_plugins/hooks/chkdsk_hook.py
View File

@ -25,7 +25,7 @@ _blmt = BanCheckLimiter(
async def _(matcher: Matcher, bot: Bot, event: GroupMessageEvent, state: T_State):
if not isinstance(event, MessageEvent):
return
if matcher.type == "message" and matcher.priority not in [1, 9]:
if matcher.type == "message" and matcher.priority not in [1, 999]:
if state["_prefix"]["raw_command"]:
if _blmt.check(f'{event.user_id}{state["_prefix"]["raw_command"]}'):
if await BanUser.ban(

4
basic_plugins/invite_manager/__init__.py
View File

@ -1,4 +1,4 @@
from nonebot import on_request, on_message
from nonebot import on_request, on_command
from nonebot.adapters.onebot.v11 import (
Bot,
FriendRequestEvent,
@ -25,7 +25,7 @@ __plugin_configs__ = {
friend_req = on_request(priority=5, block=True)
group_req = on_request(priority=5, block=True)
x = on_message(priority=9, block=False)
x = on_command("[[test_invite_manager]]", priority=999, block=False)
exists_data = {"private": {}, "group": {}}

2
basic_plugins/shop/shop_handle/data_source.py
View File

@ -154,7 +154,7 @@ async def register_goods(
"""
if not await GoodsInfo.get_goods_info(name):
limit_time = float(limit_time) if limit_time else limit_time
discount = discount if discount is None else 1
discount = discount if discount is not None else 1
limit_time = (
int(time.time() + limit_time * 60 * 60)
if limit_time is not None and limit_time != 0

3
models/goods_info.py
View File

@ -114,8 +114,7 @@ class GoodsInfo(db.Model):
参数：
:param goods_name: 商品名称
"""
query = await cls.query.where(cls.goods_name == goods_name).gino.first()
return query
return await cls.query.where(cls.goods_name == goods_name).gino.first()
@classmethod
async def get_all_goods(cls) -> List["GoodsInfo"]:

9
plugins/bilibili_sub/data_source.py
View File

@ -1,15 +1,10 @@
# from bilibili_api.exceptions.ResponseCodeException import ResponseCodeException
from bilireq.exceptions import ResponseCodeError
from utils.manager import resources_manager
from asyncio.exceptions import TimeoutError
from .model import BilibiliSub
# from bilibili_api.live import LiveRoom
from bilireq.live import get_room_info_by_id
# from bilibili_api import bangumi
from .utils import get_meta
from utils.message_builder import image
# from bilibili_api import user
# from bilibili_api.user import User
from bilireq.user import get_user_info
from bilireq import dynamic
from .utils import get_videos
@ -60,7 +55,7 @@ async def add_live_sub(live_id: int, sub_user: str) -> str:
live_status=live_status,
):
await _get_up_status(live_id)
uname = (await BilibiliSub.get_sub(uid)).uname
uname = (await BilibiliSub.get_sub(live_id)).uname
return (
"已成功订阅主播：\n"
f"\ttitle：{title}\n"
@ -248,7 +243,7 @@ async def _get_up_status(id_: int) -> Optional[str]:
"""
_user = await BilibiliSub.get_sub(id_)
"""bilibili_api.user库中User类的get_user_info改为bilireq.user库的get_user_info方法"""
user_info = await get_user_info(id_)
user_info = await get_user_info(_user.uid)
uname = user_info["name"]
"""bilibili_api.user库中User类的get_videos改为bilireq.user库的get_videos方法"""
video_info = await get_videos(id_)

2
plugins/black_word/data_source.py
View File

@ -77,7 +77,7 @@ async def show_black_text_image(
(punish_level_str, "等级"),
(create_time_str, "记录日期"),
]:
img = await text2image(s, color="#f9f6f2", _add_height=3.32)
img = await text2image(s, color="#f9f6f2", _add_height=2.1)
w = _tmp_img.getsize(type_)[0] if _tmp_img.getsize(type_)[0] > img.w else img.w
A = BuildImage(w + 11, img.h + 50, color="#f9f6f2", font_size=35, font="CJGaoDeGuo.otf")
await A.atext((10, 10), type_)

7
plugins/fudu.py
View File

@ -9,6 +9,7 @@ from nonebot.adapters.onebot.v11 import GroupMessageEvent
from configs.config import Config
from utils.http_utils import AsyncHttpx
from services.log import logger
from configs.config import NICKNAME
__zx_plugin_name__ = "复读"
@ -74,15 +75,15 @@ class Fudu:
_fudu_list = Fudu()
fudu = on_message(permission=GROUP, priority=9)
fudu = on_message(permission=GROUP, priority=999)
@fudu.handle()
async def _(event: GroupMessageEvent):
if event.is_tome():
return
if get_message_text(event.json()):
if get_message_text(event.json()).find("@可爱的小真寻") != -1:
if msg := get_message_text(event.json()):
if msg.startswith(f"@可爱的{NICKNAME}"):
await fudu.finish("复制粘贴的虚空艾特？", at_sender=True)
img = get_message_img(event.json())
msg = get_message_text(event.json())

2
plugins/genshin/almanac/__init__.py
View File

@ -35,7 +35,7 @@ Config.add_plugin_config(
default_value=True,
)
almanac = on_command("原神黄历", priority=5, block=True)
almanac = on_command("原神黄历", priority=15, block=True)
ALC_PATH = IMAGE_PATH / "genshin" / "alc"

5
plugins/genshin/query_resource_points/query_resource.py
View File

@ -1,6 +1,5 @@
from typing import Tuple, Optional, List
from configs.path_config import IMAGE_PATH, TEXT_PATH
from PIL.Image import UnidentifiedImageError
from utils.message_builder import image
from services.log import logger
from utils.image_utils import BuildImage
@ -261,8 +260,8 @@ async def download_image(
gen_icon(path)
else:
logger.info(f"下载原神资源图标：{img_url} 失败，等待下次更新...")
except UnidentifiedImageError:
logger.warning(f"原神图片打开错误..已删除，等待下次更新... file: {path}")
except Exception as e:
logger.warning(f"原神图片错误..已删除，等待下次更新... file: {path} {type(e)}：{e}")
if os.path.exists(path):
os.remove(path)

2
plugins/genshin/query_user/query_role/draw_image.py
View File

@ -300,6 +300,8 @@ def get_country_data_image(world_data_dict: Dict) -> BuildImage:
region = BuildImage(790, 267 * (len(world_data_dict) - 1), color="#F9F6F2")
height = 0
for country in ["蒙德", "龙脊雪山", "璃月", "层岩巨渊", "稻妻", "渊下宫"]:
if not world_data_dict.get(country):
continue
x = BuildImage(790, 250, color="#3A4467")
logo = BuildImage(180, 180, background=image_path / "logo" / f"{country}.png")
tmp_bk = BuildImage(770, 230, color="#606779")

13
plugins/genshin/query_user/resin_remind/init_task.py
View File

@ -1,3 +1,5 @@
from nonebot.adapters.onebot.v11 import ActionFailed
from utils.utils import get_bot, scheduler
from utils.message_builder import at
from models.group_member_info import GroupInfoUser
@ -178,10 +180,13 @@ async def _remind(user_id: int, uid: str):
group_list = await GroupInfoUser.get_user_all_group(user_id)
if group_list:
group_id = group_list[0]
await bot.send_group_msg(
group_id=group_id,
message=at(user_id) + msg
)
try:
await bot.send_group_msg(
group_id=group_id,
message=at(user_id) + msg
)
except ActionFailed as e:
logger.error(f"树脂提醒推送发生错误 {type(e)}：{e}")
if not next_time:
if user_manager.check(uid) and Config.get_config("resin_remind", "AUTO_CLOSE_QUERY_FAIL_RESIN_REMIND"):
await Genshin.set_resin_remind(int(uid), False)

4
plugins/mute.py
View File

@ -145,7 +145,7 @@ async def _(event: GroupMessageEvent, cmd: Tuple[str, ...] = Command(), arg: Mes
await mute_setting.finish(
f'最大次数：{mute_data[group_id]["count"]} 次\n'
f'规定时间：{mute_data[group_id]["time"]} 秒\n'
f'禁言时长：{mute_data[group_id]["duration"] / 60} 分钟\n'
f'禁言时长：{mute_data[group_id]["duration"]:.2f} 分钟\n'
f"【在规定时间内发送相同消息超过最大次数则禁言\n当禁言时长为0时关闭此功能】"
)
if not is_number(msg):
@ -157,7 +157,7 @@ async def _(event: GroupMessageEvent, cmd: Tuple[str, ...] = Command(), arg: Mes
mute_data[group_id]["count"] = int(msg)
msg += " 次"
if cmd[0] == "设置禁言时长":
mute_data[group_id]["duration"] = int(msg) * 60
mute_data[group_id]["duration"] = int(msg)
msg += " 分钟"
await mute_setting.send(f'刷屏检测：{cmd[0]}为 {msg}')
logger.info(

73
plugins/pid_search.py
View File

@ -52,8 +52,8 @@ headers = {
async def _g(event: MessageEvent, state: T_State, pid: str = Arg("pid")):
if pid in ["取消", "算了"]:
await pid_search.finish("已取消操作...")
# if is_number(pid):
# await pid_search.reject_arg("pid", "笨蛋，重新输入数！字！")
if not is_number(pid):
await pid_search.reject_arg("pid", "笨蛋，重新输入数！字！")
for _ in range(3):
try:
data = (
@ -68,39 +68,46 @@ async def _g(event: MessageEvent, state: T_State, pid: str = Arg("pid")):
except Exception as e:
await pid_search.finish(f"发生了一些错误..{type(e)}：{e}")
else:
if not data["illust"]["width"] and not data["illust"]["height"]:
data = data["illust"]
if not data["width"] and not data["height"]:
await pid_search.finish(f"没有搜索到 PID：{pid} 的图片", at_sender=True)
pid = data["illust"]["id"]
title = data["illust"]["title"]
author = data["illust"]["user"]["name"]
author_id = data["illust"]["user"]["id"]
img_url = data["illust"]["meta_single_page"]["original_image_url"]
if not await AsyncHttpx.download_file(
img_url,
IMAGE_PATH / "temp" / f"pid_search_{event.user_id}.png",
headers=headers,
):
await pid_search.finish("图片下载失败了....", at_sender=True)
tmp = ""
if isinstance(event, GroupMessageEvent):
tmp = "\n【注】将在30后撤回......"
msg_id = await pid_search.send(
Message(
f"title：{title}\n"
f"pid：{pid}\n"
f"author：{author}\n"
f"author_id：{author_id}\n"
f'{image(f"pid_search_{event.user_id}.png", "temp")}'
f"{tmp}"
pid = data["id"]
title = data["title"]
author = data["user"]["name"]
author_id = data["user"]["id"]
image_list = []
try:
image_list.append(data["meta_single_page"]["original_image_url"])
except KeyError:
for image_url in data["meta_pages"]:
image_list.append(image_url["image_urls"]["original"])
for i, img_url in enumerate(image_list):
if not await AsyncHttpx.download_file(
img_url,
IMAGE_PATH / "temp" / f"pid_search_{event.user_id}_{i}.png",
headers=headers,
):
await pid_search.send("图片下载失败了....", at_sender=True)
tmp = ""
if isinstance(event, GroupMessageEvent):
tmp = "\n【注】将在30后撤回......"
msg_id = await pid_search.send(
Message(
f"title：{title}\n"
f"pid：{pid}\n"
f"author：{author}\n"
f"author_id：{author_id}\n"
f'{image(f"pid_search_{event.user_id}_{i}.png", "temp")}'
f"{tmp}"
)
)
)
logger.info(
f"(USER {event.user_id}, "
f"GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 查询图片 PID：{pid}"
)
if isinstance(event, GroupMessageEvent):
withdraw_message_manager.append((msg_id, 30))
logger.info(
f"(USER {event.user_id}, "
f"GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
f" 查询图片 PID：{pid}"
)
if isinstance(event, GroupMessageEvent):
withdraw_message_manager.append((msg_id, 30))
break
else:
await pid_search.finish("图片下载失败了....", at_sender=True)

2
plugins/search_anime/__init__.py
View File

@ -49,6 +49,8 @@ async def _(bot: Bot, event: MessageEvent, state: T_State, key_word: str = ArgSt
Config.get_config("search_anime", "SEARCH_ANIME_MAX_INFO"),
)
if anime_report:
if isinstance(anime_report, str):
await search_anime.finish(anime_report)
if isinstance(event, GroupMessageEvent):
mes_list = custom_forward_msg(anime_report, bot.self_id)
await bot.send_group_forward_msg(group_id=event.group_id, messages=mes_list)

8
plugins/search_anime/data_source.py
View File

@ -3,18 +3,18 @@ import feedparser
from urllib import parse
from services.log import logger
from utils.http_utils import AsyncHttpx
from typing import List
from typing import List, Union
import time
async def from_anime_get_info(key_word: str, max_: int) -> List[str]:
async def from_anime_get_info(key_word: str, max_: int) -> Union[str, List[str]]:
s_time = time.time()
repass = ""
url = "https://share.dmhy.org/topics/rss/rss.xml?keyword=" + parse.quote(key_word)
try:
repass = await get_repass(url, max_)
except Exception as e:
logger.error("Timeout! {}".format(e))
logger.error(f"发生了一些错误 {type(e)}：{e}")
return "发生了一些错误！"
repass.insert(0, f"搜索 {key_word} 结果（耗时 {int(time.time() - s_time)} 秒）：\n")
return repass

2
plugins/statistics/statistics_hook.py
View File

@ -96,7 +96,7 @@ async def _(
global _prefix_count_dict
if (
matcher.type == "message"
and matcher.priority not in [1, 9]
and matcher.priority not in [1, 999]
and matcher.plugin_name not in ["update_info", "statistics_handle"]
):
module = matcher.plugin_name

28
plugins/word_clouds/__init__.py
View File

@ -17,26 +17,22 @@ __plugin_usage__ = """
usage：
词云
指令：
获取今天的词云
今日词云
获取昨天的词云
昨日词云
获取本周词云
本周词云
获取本月词云
本月词云
获取年度词云
年度词云
今日词云：获取今天的词云
昨日词云：获取昨天的词云
本周词云：获取本周词云
本月词云：获取本月词云
年度词云：获取年度词云
历史词云(支持 ISO8601 格式的日期与时间，如 2022-02-22T22:22:22)
获取某日的词云
历史词云 2022-01-01
获取指定时间段的词云
历史词云
历史词云 2022-01-01~2022-02-22
历史词云 2022-02-22T11:11:11~2022-02-22T22:22:22
示例：历史词云 2022-01-01~2022-02-22
示例：历史词云 2022-02-22T11:11:11~2022-02-22T22:22:22
如果想要获取自己的发言，可在命令前添加 我的
我的今日词云
示例：我的今日词云
""".strip()
__plugin_des__ = "词云"
__plugin_cmd__ = ["今日词云", "昨日词云", "本周词云"]
@ -80,12 +76,12 @@ def parse_datetime(key: str):
async def _key_parser(
matcher: Matcher,
state: T_State,
input: Union[datetime, Message] = Arg(key),
input_: Union[datetime, Message] = Arg(key),
):
if isinstance(input, datetime):
if isinstance(input_, datetime):
return
plaintext = input.extract_plain_text()
plaintext = input_.extract_plain_text()
try:
state[key] = get_datetime_fromisoformat_with_timezone(plaintext)
except ValueError:

2
utils/browser.py
View File

@ -15,8 +15,6 @@ _browser: Optional[Browser] = None
async def init(**kwargs) -> Optional[Browser]:
global _browser
if platform.system() == "Windows":
return None
try:
browser = await async_playwright().start()
_browser = await browser.chromium.launch(**kwargs)

6
utils/decorator/limit.py Normal file
View File

@ -0,0 +1,6 @@

28
utils/decorator/shop.py
View File

@ -17,13 +17,15 @@ class ShopRegister(dict):
name: Tuple[str, ...],
price: Tuple[float, ...],
des: Tuple[str, ...],
discount: Tuple[float, ...],
limit_time: Tuple[int, ...],
load_status: Tuple[bool, ...],
**kwargs,
):
def add_register_item(func: Callable):
if name in self._data.keys():
raise ValueError("该商品已注册，请替换其他名称！")
for n, p, d, s in zip(name, price, des, load_status):
for n, p, d, dd, l, s in zip(name, price, des, discount, limit_time, load_status):
if s:
_temp_kwargs = {}
for key, value in kwargs.items():
@ -32,6 +34,8 @@ class ShopRegister(dict):
self._data[n] = {
"price": p,
"des": d,
"discount": dd,
"limit_time": l,
"func": func,
"kwargs": _temp_kwargs,
}
@ -46,7 +50,7 @@ class ShopRegister(dict):
self._flag = False
for name in self._data.keys():
await shop.register_goods(
name, self._data[name]["price"], self._data[name]["des"]
name, self._data[name]["price"], self._data[name]["des"], self._data[name]["discount"], self._data[name]["limit_time"]
)
use.register_use(
name, self._data[name]["func"], **self._data[name]["kwargs"]
@ -57,24 +61,36 @@ class ShopRegister(dict):
name: Union[str, Tuple[str, ...]],
price: Union[float, Tuple[float, ...]],
des: Union[str, Tuple[str, ...]],
discount: Union[float, Tuple[float, ...]] = 1,
limit_time: Union[int, Tuple[int, ...]] = 0,
load_status: Union[bool, Tuple[bool, ...]] = True,
**kwargs,
):
_tuple_list = []
_current_len = -1
for x in [name, price, des, load_status]:
for x in [name, price, des, discount, limit_time, load_status]:
if isinstance(x, tuple):
if _current_len == -1:
_current_len = len(x)
if _current_len != len(x):
raise ValueError(f"注册商品 {name} 中 name，price，des，load_status 数量不符！")
raise ValueError(f"注册商品 {name} 中 name，price，des，discount，limit_time，load_status 数量不符！")
_current_len = _current_len if _current_len > -1 else 1
_name = name if isinstance(name, tuple) else tuple(name)
_name = name if isinstance(name, tuple) else (name,)
_price = (
price
if isinstance(price, tuple)
else tuple([price for _ in range(_current_len)])
)
_discount = (
discount
if isinstance(discount, tuple)
else tuple([discount for _ in range(_current_len)])
)
_limit_time = (
limit_time
if isinstance(limit_time, tuple)
else tuple([limit_time for _ in range(_current_len)])
)
_des = (
des if isinstance(des, tuple) else tuple([des for _ in range(_current_len)])
)
@ -83,7 +99,7 @@ class ShopRegister(dict):
if isinstance(load_status, tuple)
else tuple([load_status for _ in range(_current_len)])
)
return self.register(_name, _price, _des, _load_status, **kwargs)
return self.register(_name, _price, _des, _discount, _limit_time, _load_status, **kwargs)
def __setitem__(self, key, value):
self._data[key] = value

9
utils/image_utils.py
View File

@ -1493,10 +1493,11 @@ async def text2image(
height = 0
_tmp = BuildImage(0, 0, font=font, font_size=font_size)
for x in text.split("\n"):
x = x if x.strip() else "正"
w, h = _tmp.getsize(x)
height += h + _add_height
width = width if width > w else w
if x:
w, _ = _tmp.getsize(x)
_, h = _tmp.getsize("正")
height += h + _add_height
width = width if width > w else w
width += pw
height += ph
A = BuildImage(

35
utils/utils.py
View File

@ -16,6 +16,7 @@ try:
except ModuleNotFoundError:
import json
scheduler = require("nonebot_plugin_apscheduler").scheduler
@ -107,8 +108,8 @@ class BanCheckLimiter:
self.mint[key] = 0
return False
if (
self.mint[key] >= self.default_count
and time.time() - self.mtime[key] < self.default_check_time
self.mint[key] >= self.default_count
and time.time() - self.mtime[key] < self.default_check_time
):
self.mtime[key] = time.time()
self.mint[key] = 0
@ -202,7 +203,10 @@ def get_message_at(data: Union[str, Message]) -> List[int]:
for msg in data["message"]:
if msg["type"] == "at":
qq_list.append(int(msg["data"]["qq"]))
else:
for seg in data:
if seg.type == "at":
qq_list.append(seg.data["qq"])
return qq_list
@ -225,25 +229,6 @@ def get_message_img(data: Union[str, Message]) -> List[str]:
return img_list
def get_message_img_file(data: Union[str, Message]) -> List[str]:
"""
说明：
获取消息中所有的 图片file
参数：
:param data: event.json()
"""
file_list = []
if isinstance(data, str):
data = json.loads(data)
for msg in data["message"]:
if msg["type"] == "image":
file_list.append(msg["data"]["file"])
else:
for seg in data["image"]:
file_list.append(seg.data["file"])
return file_list
def get_message_text(data: Union[str, Message]) -> str:
"""
说明：
@ -370,7 +355,7 @@ def cn2py(word: str) -> str:
def change_pixiv_image_links(
url: str, size: Optional[str] = None, nginx_url: Optional[str] = None
url: str, size: Optional[str] = None, nginx_url: Optional[str] = None
):
"""
说明：
@ -388,7 +373,7 @@ def change_pixiv_image_links(
if nginx_url:
url = (
url.replace("i.pximg.net", nginx_url)
.replace("i.pixiv.cat", nginx_url)
.replace("_webp", "")
.replace("i.pixiv.cat", nginx_url)
.replace("_webp", "")
)
return url