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-02-21 15:59:17 +08:00
parent 87f0189abd
commit 0d3bd8a89c
13 changed files with 230 additions and 208 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
README.md
View File

@ -236,6 +236,15 @@ __Docker 最新版本由 [Sakuracio](https://github.com/Sakuracio) 提供__
## 更新
### 2022/2/21 \[v0.1.3.2]
* 群权限为-1时，超级用户发送的命令依旧生效
* 当群权限为-1时，被动技能也将不会发送
* 修复功能开关，b站转发解析，复读 ignore无法使用
* 修复色图下载文件名与路径错误
* 修复被动技能提醒有时无法删除控制文本
### 2022/2/20 \[v0.1.3.1]
* 修复pix下载临时文件目录错误
@ -247,7 +256,6 @@ __Docker 最新版本由 [Sakuracio](https://github.com/Sakuracio) 提供__
* 由于nonebot升级版本，提供更新建议（__该次升级将会导致nonebot.beta1以下的插件无法使用__
* 保证services，utils，configs，plugins，basic_plugins，文件夹均为最新
* 根目录有pyproject.toml与poetry.lock
* gocq配置文件地址更改为 ws://127.0.0.1:8080/onebot/v11/ws/
* 执行命令：
* pip3 install poetry
* poetry install

2
__version__
View File

@ -1 +1 @@
__version__: v0.1.3.1
__version__: v0.1.3.2

2
basic_plugins/admin_bot_manage/rule.py
View File

@ -31,7 +31,7 @@ def switch_rule(event: Event) -> bool:
cmd.append(f"关闭 {x}")
except KeyError:
pass
msg = get_message_text(event.json())
msg = get_message_text(event.json()).split()
msg = msg[0] if msg else ""
return msg in cmd
except Exception as e:

16
basic_plugins/admin_bot_manage/switch_rule.py
View File

@ -1,7 +1,6 @@
from nonebot.adapters.onebot.v11 import Bot, GroupMessageEvent, MessageEvent, GROUP, Message
from nonebot.adapters.onebot.v11 import Bot, GroupMessageEvent, MessageEvent, GROUP
from nonebot import on_command, on_message, on_regex
from nonebot.typing import T_State
from nonebot.params import CommandArg
from nonebot.params import RegexGroup
from ._data_source import (
change_group_switch,
set_plugin_status,
@ -13,6 +12,7 @@ from services.log import logger
from configs.config import NICKNAME, Config
from utils.utils import get_message_text, is_number
from nonebot.permission import SUPERUSER
from typing import Tuple, Any
from .rule import switch_rule
@ -65,14 +65,14 @@ group_status = on_regex("^(休息吧|醒来)$", permission=GROUP, priority=5, bl
@switch_rule_matcher.handle()
async def _(bot: Bot, event: MessageEvent, arg: Message = CommandArg()):
_cmd = arg.extract_plain_text().strip().split()[0]
async def _(bot: Bot, event: MessageEvent):
_cmd = get_message_text(event.json()).split()[0]
if isinstance(event, GroupMessageEvent):
await switch_rule_matcher.send(await change_group_switch(_cmd, event.group_id))
logger.info(f"USER {event.user_id} GROUP {event.group_id} 使用群功能管理命令 {_cmd}")
else:
if str(event.user_id) in bot.config.superusers:
block_type = " ".join(arg.extract_plain_text().strip().split()[1:])
block_type = " ".join(get_message_text(event.json()).split()[1:])
block_type = block_type if block_type else "a"
if is_number(block_type):
if not int(block_type) in [
@ -114,8 +114,8 @@ async def _(event: GroupMessageEvent):
@group_status.handle()
async def _(event: GroupMessageEvent, state: T_State):
cmd = state["_matched_groups"][0]
async def _(event: GroupMessageEvent, reg_group: Tuple[Any, ...] = RegexGroup()):
cmd = reg_group[0]
if cmd == "休息吧":
msg = set_group_bot_status(event.group_id, False)
else:

89
basic_plugins/hooks/auth_hook.py
View File

@ -11,7 +11,7 @@ from utils.manager import (
plugins_manager,
plugins2cd_manager,
plugins2block_manager,
plugins2count_manager
plugins2count_manager,
)
from ._utils import set_block_limit_false, status_message_manager
from nonebot.typing import T_State
@ -23,7 +23,7 @@ from nonebot.adapters.onebot.v11 import (
PokeNotifyEvent,
PrivateMessageEvent,
Message,
Event
Event,
)
from configs.config import Config
from models.ban_user import BanUser
@ -49,23 +49,28 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
# 功能的金币检测 #######################################
# 功能的金币检测 #######################################
cost_gold = 0
if isinstance(event, GroupMessageEvent) and plugins2settings_manager.get_plugin_data(module).get('cost_gold'):
cost_gold = plugins2settings_manager.get_plugin_data(module).get('cost_gold')
if isinstance(
event, GroupMessageEvent
) and plugins2settings_manager.get_plugin_data(module).get("cost_gold"):
cost_gold = plugins2settings_manager.get_plugin_data(module).get("cost_gold")
if await BagUser.get_gold(event.user_id, event.group_id) < cost_gold:
await send_msg(f"金币不足..该功能需要{cost_gold}金币..", bot, event)
raise IgnoredException(f"{module} 金币限制...")
# 当插件不阻塞超级用户时，超级用户提前扣除金币
if str(event.user_id) in bot.config.superusers and not plugins2info_dict[module]["limit_superuser"]:
if (
str(event.user_id) in bot.config.superusers
and not plugins2info_dict[module]["limit_superuser"]
):
await BagUser.spend_gold(event.user_id, event.group_id, cost_gold)
try:
if (
(not isinstance(event, MessageEvent) and module != "poke")
or await BanUser.is_ban(event.user_id)
and str(event.user_id) not in bot.config.superusers
(not isinstance(event, MessageEvent) and module != "poke")
or await BanUser.is_ban(event.user_id)
and str(event.user_id) not in bot.config.superusers
) or (
str(event.user_id) in bot.config.superusers
and plugins2info_dict.get(module)
and not plugins2info_dict[module]["limit_superuser"]
str(event.user_id) in bot.config.superusers
and plugins2info_dict.get(module)
and not plugins2info_dict[module]["limit_superuser"]
):
return
except AttributeError:
@ -76,8 +81,8 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
_module = _plugin.module
plugin_name = _module.__getattribute__("__zx_plugin_name__")
if (
"[superuser]" in plugin_name.lower()
and str(event.user_id) in bot.config.superusers
"[superuser]" in plugin_name.lower()
and str(event.user_id) in bot.config.superusers
):
return
except AttributeError:
@ -85,7 +90,10 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
# 群黑名单检测 群总开关检测
if isinstance(event, GroupMessageEvent) or matcher.plugin_name == "poke":
try:
if group_manager.get_group_level(event.group_id) < 0:
if (
group_manager.get_group_level(event.group_id) < 0
and str(event.user_id) not in bot.config.superusers
):
raise IgnoredException("群黑名单")
if not group_manager.check_group_bot_status(event.group_id):
try:
@ -99,12 +107,12 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
if isinstance(event, GroupMessageEvent):
# 个人权限
if (
not await LevelUser.check_level(
event.user_id,
event.group_id,
admin_manager.get_plugin_level(module),
)
and admin_manager.get_plugin_level(module) > 0
not await LevelUser.check_level(
event.user_id,
event.group_id,
admin_manager.get_plugin_level(module),
)
and admin_manager.get_plugin_level(module) > 0
):
try:
if _flmt.check(event.user_id):
@ -112,7 +120,7 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
await bot.send_group_msg(
group_id=event.group_id,
message=f"{at(event.user_id)}你的权限不足喔，该功能需要的权限等级："
f"{admin_manager.get_plugin_level(module)}",
f"{admin_manager.get_plugin_level(module)}",
)
except ActionFailed:
pass
@ -122,7 +130,7 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
raise IgnoredException("权限不足")
else:
if not await LevelUser.check_level(
event.user_id, 0, admin_manager.get_plugin_level(module)
event.user_id, 0, admin_manager.get_plugin_level(module)
):
try:
await bot.send_private_msg(
@ -138,12 +146,12 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
if module in plugins2info_dict.keys() and matcher.priority not in [1, 9]:
# 戳一戳单独判断
if isinstance(event, GroupMessageEvent) or (
isinstance(event, PokeNotifyEvent) and event.group_id
isinstance(event, PokeNotifyEvent) and event.group_id
):
if status_message_manager.get(event.group_id) is None:
status_message_manager.delete(event.group_id)
if plugins2info_dict[module]["level"] > group_manager.get_group_level(
event.group_id
event.group_id
):
try:
if _flmt_g.check(event.user_id) and module not in ignore_rst_module:
@ -161,7 +169,7 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
if not group_manager.get_plugin_status(module, event.group_id):
try:
if module not in ignore_rst_module and _flmt_s.check(
event.group_id
event.group_id
):
_flmt_s.start_cd(event.group_id)
await bot.send_group_msg(
@ -177,8 +185,8 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
if not group_manager.get_plugin_status(f"{module}:super", event.group_id):
try:
if (
_flmt_s.check(event.group_id)
and module not in ignore_rst_module
_flmt_s.check(event.group_id)
and module not in ignore_rst_module
):
_flmt_s.start_cd(event.group_id)
await bot.send_group_msg(
@ -194,8 +202,8 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
if not plugins_manager.get_plugin_status(module, block_type="group"):
try:
if (
_flmt_c.check(event.group_id)
and module not in ignore_rst_module
_flmt_c.check(event.group_id)
and module not in ignore_rst_module
):
_flmt_c.start_cd(event.group_id)
await bot.send_group_msg(
@ -225,14 +233,14 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
# 维护
if not plugins_manager.get_plugin_status(module, block_type="all"):
if isinstance(
event, GroupMessageEvent
event, GroupMessageEvent
) and group_manager.check_group_is_white(event.group_id):
return
try:
if isinstance(event, GroupMessageEvent):
if (
_flmt_c.check(event.group_id)
and module not in ignore_rst_module
_flmt_c.check(event.group_id)
and module not in ignore_rst_module
):
_flmt_c.start_cd(event.group_id)
await bot.send_group_msg(
@ -267,9 +275,9 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
limit_type = plugin_cd_data["limit_type"]
rst = plugin_cd_data["rst"]
if (
(isinstance(event, PrivateMessageEvent) and check_type == "private")
or (isinstance(event, GroupMessageEvent) and check_type == "group")
or plugins2cd_manager.get_plugin_data(module).get("check_type") == "all"
(isinstance(event, PrivateMessageEvent) and check_type == "private")
or (isinstance(event, GroupMessageEvent) and check_type == "group")
or plugins2cd_manager.get_plugin_data(module).get("check_type") == "all"
):
cd_type_ = event.user_id
if limit_type == "group" and isinstance(event, GroupMessageEvent):
@ -288,9 +296,9 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
limit_type = plugin_block_data["limit_type"]
rst = plugin_block_data["rst"]
if (
(isinstance(event, PrivateMessageEvent) and check_type == "private")
or (isinstance(event, GroupMessageEvent) and check_type == "group")
or check_type == "all"
(isinstance(event, PrivateMessageEvent) and check_type == "private")
or (isinstance(event, GroupMessageEvent) and check_type == "group")
or check_type == "all"
):
block_type_ = event.user_id
if limit_type == "group" and isinstance(event, GroupMessageEvent):
@ -304,8 +312,8 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
plugins2block_manager.set_true(block_type_, module)
# Count
if (
plugins2count_manager.check_plugin_count_status(module)
and event.user_id not in bot.config.superusers
plugins2count_manager.check_plugin_count_status(module)
and event.user_id not in bot.config.superusers
):
plugin_count_data = plugins2count_manager.get_plugin_count_data(module)
limit_type = plugin_count_data["limit_type"]
@ -375,4 +383,3 @@ async def init_rst(rst: str, event: MessageEvent):
if "[at]" in rst and isinstance(event, GroupMessageEvent):
rst = rst.replace("[at]", str(at(event.user_id)))
return rst

35
basic_plugins/hooks/task_hook.py
View File

@ -1,7 +1,6 @@
from nonebot.exception import MockApiException
from nonebot.adapters.onebot.v11 import Bot, MessageSegment
from nonebot.adapters.onebot.v11 import Bot, Message
from utils.manager import group_manager
from utils.utils import get_message_text
from typing import Dict, Any
import re
@ -10,27 +9,29 @@ import re
async def handle_api_call(bot: Bot, api: str, data: Dict[str, Any]):
r = None
if (
(api == "send_msg" and data["message"] == "group_id" or api == "send_group_msg")
(
(api == "send_msg" and data["message_type"] == "group")
or api == "send_group_msg"
)
and (
r := re.search(
"^\[\[_task\|(.*)]]",
data["message"]
if isinstance(data["message"], str)
else get_message_text(data["message"]),
(r := re.search("^\[\[_task\|(.*)]]", str(data["message"]).strip()))
or (
r := re.search(
"^&#91;&#91;_task\|(.*)&#93;&#93;", str(data["message"]).strip()
)
)
)
and r.group(1) in group_manager.get_task_data().keys()
):
task = r.group(1)
group_id = data["group_id"]
if not await group_manager.check_group_task_status(group_id, task):
if group_manager.get_group_level(
group_id
) < 0 or not await group_manager.check_group_task_status(group_id, task):
raise MockApiException(f"被动技能 {task} 处于关闭状态...")
else:
if isinstance(data["message"], str):
msg = data["message"]
msg = msg.replace(f"[[_task|{task}]]", "")
data["message"] = msg
else:
msg = str(data["message"][0])
msg = msg.replace(f"&#91;&#91;_task|{task}&#93;&#93;", "")
data["message"][0] = MessageSegment.text(msg)
msg = str(data["message"]).strip()
msg = msg.replace(f"&#91;&#91;_task|{task}&#93;&#93;", "").replace(
f"[[_task|{task}]]", ""
)
data["message"] = Message(msg)

1
plugins/draw_card/__init__.py
View File

@ -122,7 +122,6 @@ onmyoji_update = on_keyword({'更新阴阳师信息'}, permission=SUPERUSER, pri
@prts.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State, reg: Tuple[Any, ...] = RegexGroup()):
msg = str(event.get_message()).strip()
print(reg)
if msg in ['方舟一井', '方舟1井']:
num = 300
else:

1
plugins/draw_card/util.py
View File

@ -182,7 +182,6 @@ def init_rst(
rst += f"第 {index} 抽获取UP {name}\n"
else:
rst += f"第 {index} 抽获取 {name}\n"
print(rst)
return rst[:-1] if rst else ""

2
plugins/epic/__init__.py
View File

@ -72,6 +72,6 @@ async def _():
if msg_list and code == 200:
await bot.send_group_forward_msg(group_id=g, messages=msg_list)
else:
bot.send_group_msg(group_id=g, messages=msg_list)
await bot.send_group_msg(group_id=g)
except Exception as e:
logger.error(f"GROUP {g} epic免费游戏推送错误 {type(e)}: {e}")

25
plugins/fudu.py
View File

@ -4,14 +4,11 @@ from utils.image_utils import get_img_hash
import random
from utils.message_builder import image
from nonebot import on_message
from utils.utils import get_message_img
from nonebot.params import CommandArg, Command
from nonebot.adapters.onebot.v11 import GroupMessageEvent, Message
from utils.utils import get_message_img, get_message_text
from nonebot.adapters.onebot.v11 import GroupMessageEvent
from configs.config import Config
from utils.http_utils import AsyncHttpx
from utils.manager import group_manager
from services.log import logger
from typing import Tuple
__zx_plugin_name__ = "复读"
@ -81,18 +78,14 @@ fudu = on_message(permission=GROUP, priority=9)
@fudu.handle()
async def _(event: GroupMessageEvent, cmd: Tuple[str, ...] = Command(), arg: Message = CommandArg()):
if (
event.is_tome()
or cmd
or not await group_manager.check_group_task_status(event.group_id, "fudu")
):
async def _(event: GroupMessageEvent):
if event.is_tome():
return
if arg.extract_plain_text().strip():
if arg.extract_plain_text().strip().find("@可爱的小真寻") != -1:
if get_message_text(event.json()):
if get_message_text(event.json()).find("@可爱的小真寻") != -1:
await fudu.finish("复制粘贴的虚空艾特？", at_sender=True)
img = get_message_img(event.json())
msg = arg.extract_plain_text().strip()
msg = get_message_text(event.json())
if not img and not msg:
return
if img:
@ -112,7 +105,7 @@ async def _(event: GroupMessageEvent, cmd: Tuple[str, ...] = Command(), arg: Mes
"fudu", "FUDU_PROBABILITY"
) and not _fudu_list.is_repeater(event.group_id):
if random.random() < 0.2:
await fudu.finish("打断施法！")
await fudu.finish("[[_task|fudu]]打断施法！")
_fudu_list.set_repeater(event.group_id)
if img and msg:
rst = msg + image(f"compare_{event.group_id}_img.jpg", "temp")
@ -125,7 +118,7 @@ async def _(event: GroupMessageEvent, cmd: Tuple[str, ...] = Command(), arg: Mes
if rst:
if rst.endswith("打断施法！"):
rst = "打断" + rst
await fudu.send(rst)
await fudu.send("[[_task|fudu]]" + rst)
async def get_fudu_img_hash(url, group_id):

220
plugins/parse_bilibili_json.py
View File

@ -1,7 +1,7 @@
from nonebot import on_message
from services.log import logger
from nonebot.adapters.onebot.v11 import GroupMessageEvent, Message
from utils.utils import get_message_json, get_local_proxy, is_number
from nonebot.adapters.onebot.v11 import GroupMessageEvent
from utils.utils import get_message_json, get_local_proxy, is_number, get_message_text
from nonebot.adapters.onebot.v11.permission import GROUP
from bilibili_api import video
from utils.message_builder import image
@ -11,11 +11,11 @@ from utils.browser import get_browser
from configs.path_config import IMAGE_PATH
from utils.http_utils import AsyncHttpx
from configs.config import Config
from nonebot.params import CommandArg
from utils.user_agent import get_user_agent
import aiohttp
import asyncio
import time
from bilibili_api import settings
from utils.manager import group_manager
import ujson as json
@ -47,113 +47,123 @@ _tmp = {}
@parse_bilibili_json.handle()
async def _(event: GroupMessageEvent, arg: Message = CommandArg()):
if await group_manager.check_group_task_status(event.group_id, "bilibili_parse"):
vd_info = None
url = None
if get_message_json(event.json()):
try:
data = json.loads(get_message_json(event.json())[0]["data"])
except (IndexError, KeyError):
data = None
if data:
# 转发视频
if data.get("desc") == "哔哩哔哩":
response = await AsyncHttpx.get(
data["meta"]["detail_1"]["qqdocurl"], timeout=7
)
url = str(response.url).split("?")[0]
bvid = url.split("/")[-1]
vd_info = await video.Video(bvid=bvid).get_info()
# 转发专栏
if (
data.get("meta")
and data["meta"].get("news")
and data["meta"]["news"].get("desc") == "哔哩哔哩专栏"
):
url = data["meta"]["news"]["jumpUrl"]
page = None
try:
browser = await get_browser()
if not browser:
return
page = await browser.new_page(
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
" (KHTML, like Gecko) Chrome/93.0.4530.0 Safari/537.36"
)
await page.goto(url, wait_until="networkidle", timeout=10000)
await page.set_viewport_size({"width": 2560, "height": 1080})
await page.click("#app > div")
div = await page.query_selector("#app > div")
await div.screenshot(
path=f"{IMAGE_PATH}/temp/cv_{event.user_id}.png",
timeout=100000,
)
await asyncio.get_event_loop().run_in_executor(
None, resize, f"{IMAGE_PATH}/temp/cv_{event.user_id}.png"
)
await parse_bilibili_json.send(
image(f"cv_{event.user_id}.png", "temp")
)
await page.close()
logger.info(
f"USER {event.user_id} GROUP {event.group_id} 解析bilibili转发 {url}"
)
except Exception as e:
logger.error(f"尝试解析bilibili专栏 {url} 失败 {type(e)}：{e}")
if page:
await page.close()
return
# BV
if arg.extract_plain_text().strip():
msg = arg.extract_plain_text().strip()
if "BV" in msg:
index = msg.find("BV")
if len(msg[index + 2 :]) >= 10:
msg = msg[index : index + 12]
url = f"https://www.bilibili.com/video/{msg}"
vd_info = await video.Video(bvid=msg).get_info()
elif "av" in msg:
index = msg.find("av")
if len(msg[index + 2 :]) >= 9:
msg = msg[index + 2 : index + 11]
if is_number(msg):
url = f"https://www.bilibili.com/video/{msg}"
vd_info = await video.Video(aid=int(msg)).get_info()
elif "https://b23.tv" in msg:
url = "https://" + msg[msg.find("b23.tv") : msg.find("b23.tv") + 13]
res = await AsyncHttpx.get(url, timeout=7)
url = str(res.url).split("?")[0]
bvid = url.split("/")[-1]
vd_info = await video.Video(bvid=bvid).get_info()
if vd_info:
async def _(event: GroupMessageEvent):
vd_info = None
url = None
if get_message_json(event.json()):
try:
data = json.loads(get_message_json(event.json())[0]["data"])
except (IndexError, KeyError):
data = None
if data:
# 转发视频
if data.get("desc") == "哔哩哔哩":
async with aiohttp.ClientSession(
headers=get_user_agent()
) as session:
async with session.get(
data["meta"]["detail_1"]["qqdocurl"],
proxy=get_local_proxy(),
timeout=7,
) as response:
url = str(response.url).split("?")[0]
bvid = url.split("/")[-1]
vd_info = await video.Video(bvid=bvid).get_info()
# response = await AsyncHttpx.get(
# data["meta"]["detail_1"]["qqdocurl"], timeout=7
# )
# url = str(response.url).split("?")[0]
# bvid = url.split("/")[-1]
# vd_info = await video.Video(bvid=bvid).get_info()
# 转发专栏
if (
url in _tmp.keys() and time.time() - _tmp[url] > 30
) or url not in _tmp.keys():
_tmp[url] = time.time()
aid = vd_info["aid"]
title = vd_info["title"]
author = vd_info["owner"]["name"]
reply = vd_info["stat"]["reply"] # 回复
favorite = vd_info["stat"]["favorite"] # 收藏
coin = vd_info["stat"]["coin"] # 投币
# like = vd_info['stat']['like'] # 点赞
# danmu = vd_info['stat']['danmaku'] # 弹幕
date = time.strftime("%Y-%m-%d", time.localtime(vd_info["ctime"]))
data.get("meta")
and data["meta"].get("news")
and data["meta"]["news"].get("desc") == "哔哩哔哩专栏"
):
url = data["meta"]["news"]["jumpUrl"]
page = None
try:
await parse_bilibili_json.send(
image(vd_info["pic"]) + f"\nav{aid}\n标题：{title}\n"
f"UP：{author}\n"
f"上传日期：{date}\n"
f"回复：{reply}，收藏：{favorite}，投币：{coin}\n"
f"{url}"
browser = await get_browser()
if not browser:
return
page = await browser.new_page(
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
" (KHTML, like Gecko) Chrome/93.0.4530.0 Safari/537.36"
)
except ActionFailed:
logger.warning(f"{event.group_id} 发送bilibili解析失败")
else:
await page.goto(url, wait_until="networkidle", timeout=10000)
await page.set_viewport_size({"width": 2560, "height": 1080})
await page.click("#app > div")
div = await page.query_selector("#app > div")
await div.screenshot(
path=f"{IMAGE_PATH}/temp/cv_{event.user_id}.png",
timeout=100000,
)
await asyncio.get_event_loop().run_in_executor(
None, resize, f"{IMAGE_PATH}/temp/cv_{event.user_id}.png"
)
await parse_bilibili_json.send(
"[[_task|bilibili_parse]]" + image(f"cv_{event.user_id}.png", "temp")
)
await page.close()
logger.info(
f"USER {event.user_id} GROUP {event.group_id} 解析bilibili转发 {url}"
)
except Exception as e:
logger.error(f"尝试解析bilibili专栏 {url} 失败 {type(e)}：{e}")
if page:
await page.close()
return
# BV
if msg := get_message_text(event.json()):
if "BV" in msg:
index = msg.find("BV")
if len(msg[index + 2 :]) >= 10:
msg = msg[index : index + 12]
url = f"https://www.bilibili.com/video/{msg}"
vd_info = await video.Video(bvid=msg).get_info()
elif "av" in msg:
index = msg.find("av")
if len(msg[index + 2 :]) >= 9:
msg = msg[index + 2 : index + 11]
if is_number(msg):
url = f"https://www.bilibili.com/video/{msg}"
vd_info = await video.Video(aid=int(msg)).get_info()
elif "https://b23.tv" in msg:
url = "https://" + msg[msg.find("b23.tv") : msg.find("b23.tv") + 13]
res = await AsyncHttpx.get(url, timeout=7)
url = str(res.url).split("?")[0]
bvid = url.split("/")[-1]
vd_info = await video.Video(bvid=bvid).get_info()
if vd_info:
if (
url in _tmp.keys() and time.time() - _tmp[url] > 30
) or url not in _tmp.keys():
_tmp[url] = time.time()
aid = vd_info["aid"]
title = vd_info["title"]
author = vd_info["owner"]["name"]
reply = vd_info["stat"]["reply"] # 回复
favorite = vd_info["stat"]["favorite"] # 收藏
coin = vd_info["stat"]["coin"] # 投币
# like = vd_info['stat']['like'] # 点赞
# danmu = vd_info['stat']['danmaku'] # 弹幕
date = time.strftime("%Y-%m-%d", time.localtime(vd_info["ctime"]))
try:
await parse_bilibili_json.send(
"[[_task|bilibili_parse]]" +
image(vd_info["pic"]) + f"\nav{aid}\n标题：{title}\n"
f"UP：{author}\n"
f"上传日期：{date}\n"
f"回复：{reply}，收藏：{favorite}，投币：{coin}\n"
f"{url}"
)
except ActionFailed:
logger.warning(f"{event.group_id} 发送bilibili解析失败")
else:
logger.info(
f"USER {event.user_id} GROUP {event.group_id} 解析bilibili转发 {url}"
)
def resize(path: str):

14
plugins/send_setu_/send_setu/data_source.py
View File

@ -89,13 +89,13 @@ async def search_online_setu(
ws_url = Config.get_config("pixiv", "PIXIV_NGINX_URL")
if ws_url:
url_ = url_.replace("i.pximg.net", ws_url).replace("i.pixiv.cat", ws_url)
index = random.randint(1, 100000) if id_ is None else id_
path_ = IMAGE_PATH / path_ if path_ else TEMP_PATH
file_name = f"{index}_temp_setu.jpg" if path_ == TEMP_PATH else f"{index}.jpg"
path_.mkdir(parents=True, exist_ok=True)
for i in range(3):
logger.info(f"search_online_setu --> {i}")
try:
index = random.randint(1, 100000) if id_ is None else id_
path_ = IMAGE_PATH / path_ if path_ else TEMP_PATH
file_name = f"{index}_temp_setu.jpg" if not path_ == TEMP_PATH else f"{index}.jpg"
path_.mkdir(parents=True, exist_ok=True)
if not await AsyncHttpx.download_file(
url_,
path_ / file_name,
@ -104,14 +104,14 @@ async def search_online_setu(
continue
if id_ is not None:
if (
os.path.getsize(f"{IMAGE_PATH}/{path_}/{index}.jpg")
os.path.getsize(path_ / f"{index}.jpg")
> 1024 * 1024 * 1.5
):
compressed_image(
f"{IMAGE_PATH}/{path_}/{index}.jpg",
path_ / f"{index}.jpg",
)
logger.info(f"下载 lolicon 图片 {url_} 成功， id：{index}")
return image(abspath=path_ / file_name), index
return image(path_ / file_name), index
except TimeoutError:
pass
except Exception as e:

21
utils/manager/group_manager.py
View File

@ -261,15 +261,20 @@ class GroupManager(StaticData):
初始化群聊 被动技能 状态
"""
if not self._task:
_m = []
for matcher in get_matchers():
_plugin = nonebot.plugin.get_plugin(matcher.plugin_name)
try:
_module = _plugin.module
plugin_task = _module.__getattribute__("__plugin_task__")
for key in plugin_task.keys():
self._task[key] = plugin_task[key]
except AttributeError:
pass
if matcher.plugin_name not in _m:
_m.append(matcher.plugin_name)
_plugin = nonebot.plugin.get_plugin(matcher.plugin_name)
try:
_module = _plugin.module
plugin_task = _module.__getattribute__("__plugin_task__")
for key in plugin_task.keys():
if key in self._task.keys():
raise ValueError(f"plugin_task：{key} 已存在！")
self._task[key] = plugin_task[key]
except AttributeError:
pass
bot = get_bot()
if bot or group_id:
if group_id: