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

update wbtop

Browse Source
This commit is contained in:
HibiKier 2022-05-21 23:34:48 +08:00
parent 82902e0cde
commit c511ef50a2
5 changed files with 174 additions and 178 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -254,6 +254,7 @@ __Docker 最新版本由 [Sakuracio](https://github.com/Sakuracio) 提供__
* 修复了多连开箱无法指定武器箱 * 修复了多连开箱无法指定武器箱
* 修复识番链接无法正确获取 * 修复识番链接无法正确获取
* 新增真寻入群时即刻刷新权限 * 新增真寻入群时即刻刷新权限
* 提高了微博热搜截图的等待时间
### 2022/5/19 ### 2022/5/19

4
plugins/alapi/wbtop.py
View File

@ -60,8 +60,8 @@ async def _(event: MessageEvent, arg: Message = CommandArg()):
img = await AsyncPlaywright.screenshot( img = await AsyncPlaywright.screenshot(
url, url,
f"{IMAGE_PATH}/temp/wbtop_{event.user_id}.png", f"{IMAGE_PATH}/temp/wbtop_{event.user_id}.png",
"#pl_feedlist_index", "#pl_feed_main",
wait_time=5 wait_time=12
) )
if img: if img:
await wbtop.send(img) await wbtop.send(img)

337
plugins/parse_bilibili_json.py
View File

@ -1,171 +1,166 @@
from nonebot import on_message # from nonebot import on_message
from services.log import logger # from services.log import logger
from nonebot.adapters.onebot.v11 import GroupMessageEvent # from nonebot.adapters.onebot.v11 import GroupMessageEvent
from utils.utils import get_message_json, get_local_proxy, is_number, get_message_text # from utils.utils import get_message_json, get_local_proxy, is_number, get_message_text
from nonebot.adapters.onebot.v11.permission import GROUP # from nonebot.adapters.onebot.v11.permission import GROUP
from bilibili_api import video # from utils.message_builder import image
from utils.message_builder import image # from nonebot.adapters.onebot.v11.exception import ActionFailed
from nonebot.adapters.onebot.v11.exception import ActionFailed # from utils.image_utils import BuildImage
from utils.image_utils import BuildImage # from utils.browser import get_browser
from utils.browser import get_browser # from configs.path_config import IMAGE_PATH
from configs.path_config import IMAGE_PATH # from utils.http_utils import AsyncHttpx
from utils.http_utils import AsyncHttpx # from configs.config import Config
from configs.config import Config # from utils.user_agent import get_user_agent
from utils.user_agent import get_user_agent # import aiohttp
import aiohttp # import asyncio
import asyncio # import time
import time # import ujson as json
from bilibili_api import settings #
import ujson as json #
# __zx_plugin_name__ = "B站转发解析"
# __plugin_usage__ = """
__zx_plugin_name__ = "B站转发解析" # usage：
__plugin_usage__ = """ # B站转发解析，解析b站分享信息，支持bv，bilibili链接，b站手机端转发卡片，cv，b23.tv，且5分钟内不解析相同url
usage： # """.strip()
B站转发解析，解析b站分享信息，支持bv，bilibili链接，b站手机端转发卡片，cv，b23.tv，且5分钟内不解析相同url # __plugin_des__ = "B站转发解析"
""".strip() # __plugin_type__ = ("其他",)
__plugin_des__ = "B站转发解析" # __plugin_version__ = 0.1
__plugin_type__ = ("其他",) # __plugin_author__ = "HibiKier"
__plugin_version__ = 0.1 # __plugin_task__ = {"bilibili_parse": "b站转发解析"}
__plugin_author__ = "HibiKier" # Config.add_plugin_config(
__plugin_task__ = {"bilibili_parse": "b站转发解析"} # "_task",
Config.add_plugin_config( # "DEFAULT_BILIBILI_PARSE",
"_task", # True,
"DEFAULT_BILIBILI_PARSE", # help_="被动 B站转发解析 进群默认开关状态",
True, # default_value=True,
help_="被动 B站转发解析 进群默认开关状态", # )
default_value=True, #
) #
# parse_bilibili_json = on_message(priority=1, permission=GROUP, block=False)
#
if get_local_proxy(): # _tmp = {}
settings.proxy = get_local_proxy() #
#
parse_bilibili_json = on_message(priority=1, permission=GROUP, block=False) # @parse_bilibili_json.handle()
# async def _(event: GroupMessageEvent):
_tmp = {} # vd_info = None
# url = None
# if get_message_json(event.json()):
@parse_bilibili_json.handle() # try:
async def _(event: GroupMessageEvent): # data = json.loads(get_message_json(event.json())[0]["data"])
vd_info = None # except (IndexError, KeyError):
url = None # data = None
if get_message_json(event.json()): # if data:
try: # # 转发视频
data = json.loads(get_message_json(event.json())[0]["data"]) # if data.get("desc") == "哔哩哔哩":
except (IndexError, KeyError): # async with aiohttp.ClientSession(
data = None # headers=get_user_agent()
if data: # ) as session:
# 转发视频 # async with session.get(
if data.get("desc") == "哔哩哔哩": # data["meta"]["detail_1"]["qqdocurl"],
async with aiohttp.ClientSession( # proxy=get_local_proxy(),
headers=get_user_agent() # timeout=7,
) as session: # ) as response:
async with session.get( # url = str(response.url).split("?")[0]
data["meta"]["detail_1"]["qqdocurl"], # bvid = url.split("/")[-1]
proxy=get_local_proxy(), # vd_info = await video.Video(bvid=bvid).get_info()
timeout=7, # # response = await AsyncHttpx.get(
) as response: # # data["meta"]["detail_1"]["qqdocurl"], timeout=7
url = str(response.url).split("?")[0] # # )
bvid = url.split("/")[-1] # # url = str(response.url).split("?")[0]
vd_info = await video.Video(bvid=bvid).get_info() # # bvid = url.split("/")[-1]
# response = await AsyncHttpx.get( # # vd_info = await video.Video(bvid=bvid).get_info()
# data["meta"]["detail_1"]["qqdocurl"], timeout=7 # # 转发专栏
# ) # if (
# url = str(response.url).split("?")[0] # data.get("meta")
# bvid = url.split("/")[-1] # and data["meta"].get("news")
# vd_info = await video.Video(bvid=bvid).get_info() # and data["meta"]["news"].get("desc") == "哔哩哔哩专栏"
# 转发专栏 # ):
if ( # url = data["meta"]["news"]["jumpUrl"]
data.get("meta") # page = None
and data["meta"].get("news") # try:
and data["meta"]["news"].get("desc") == "哔哩哔哩专栏" # browser = await get_browser()
): # if not browser:
url = data["meta"]["news"]["jumpUrl"] # return
page = None # page = await browser.new_page(
try: # user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
browser = await get_browser() # " (KHTML, like Gecko) Chrome/93.0.4530.0 Safari/537.36"
if not browser: # )
return # await page.goto(url, wait_until="networkidle", timeout=10000)
page = await browser.new_page( # await page.set_viewport_size({"width": 2560, "height": 1080})
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36" # await page.click("#app > div")
" (KHTML, like Gecko) Chrome/93.0.4530.0 Safari/537.36" # div = await page.query_selector("#app > div")
) # await div.screenshot(
await page.goto(url, wait_until="networkidle", timeout=10000) # path=f"{IMAGE_PATH}/temp/cv_{event.user_id}.png",
await page.set_viewport_size({"width": 2560, "height": 1080}) # timeout=100000,
await page.click("#app > div") # )
div = await page.query_selector("#app > div") # await asyncio.get_event_loop().run_in_executor(
await div.screenshot( # None, resize, f"{IMAGE_PATH}/temp/cv_{event.user_id}.png"
path=f"{IMAGE_PATH}/temp/cv_{event.user_id}.png", # )
timeout=100000, # await parse_bilibili_json.send(
) # "[[_task|bilibili_parse]]" + image(f"cv_{event.user_id}.png", "temp")
await asyncio.get_event_loop().run_in_executor( # )
None, resize, f"{IMAGE_PATH}/temp/cv_{event.user_id}.png" # await page.close()
) # logger.info(
await parse_bilibili_json.send( # f"USER {event.user_id} GROUP {event.group_id} 解析bilibili转发 {url}"
"[[_task|bilibili_parse]]" + image(f"cv_{event.user_id}.png", "temp") # )
) # except Exception as e:
await page.close() # logger.error(f"尝试解析bilibili专栏 {url} 失败 {type(e)}：{e}")
logger.info( # if page:
f"USER {event.user_id} GROUP {event.group_id} 解析bilibili转发 {url}" # await page.close()
) # return
except Exception as e: # # BV
logger.error(f"尝试解析bilibili专栏 {url} 失败 {type(e)}：{e}") # if msg := get_message_text(event.json()):
if page: # if "BV" in msg:
await page.close() # index = msg.find("BV")
return # if len(msg[index + 2 :]) >= 10:
# BV # msg = msg[index : index + 12]
if msg := get_message_text(event.json()): # url = f"https://www.bilibili.com/video/{msg}"
if "BV" in msg: # vd_info = await video.Video(bvid=msg).get_info()
index = msg.find("BV") # elif "av" in msg:
if len(msg[index + 2 :]) >= 10: # index = msg.find("av")
msg = msg[index : index + 12] # if len(msg[index + 2 :]) >= 9:
url = f"https://www.bilibili.com/video/{msg}" # msg = msg[index + 2 : index + 11]
vd_info = await video.Video(bvid=msg).get_info() # if is_number(msg):
elif "av" in msg: # url = f"https://www.bilibili.com/video/{msg}"
index = msg.find("av") # vd_info = await video.Video(aid=int(msg)).get_info()
if len(msg[index + 2 :]) >= 9: # elif "https://b23.tv" in msg:
msg = msg[index + 2 : index + 11] # url = "https://" + msg[msg.find("b23.tv") : msg.find("b23.tv") + 13]
if is_number(msg): # res = await AsyncHttpx.get(url, timeout=7)
url = f"https://www.bilibili.com/video/{msg}" # url = str(res.url).split("?")[0]
vd_info = await video.Video(aid=int(msg)).get_info() # bvid = url.split("/")[-1]
elif "https://b23.tv" in msg: # vd_info = await video.Video(bvid=bvid).get_info()
url = "https://" + msg[msg.find("b23.tv") : msg.find("b23.tv") + 13] # if vd_info:
res = await AsyncHttpx.get(url, timeout=7) # if (
url = str(res.url).split("?")[0] # url in _tmp.keys() and time.time() - _tmp[url] > 30
bvid = url.split("/")[-1] # ) or url not in _tmp.keys():
vd_info = await video.Video(bvid=bvid).get_info() # _tmp[url] = time.time()
if vd_info: # aid = vd_info["aid"]
if ( # title = vd_info["title"]
url in _tmp.keys() and time.time() - _tmp[url] > 30 # author = vd_info["owner"]["name"]
) or url not in _tmp.keys(): # reply = vd_info["stat"]["reply"] # 回复
_tmp[url] = time.time() # favorite = vd_info["stat"]["favorite"] # 收藏
aid = vd_info["aid"] # coin = vd_info["stat"]["coin"] # 投币
title = vd_info["title"] # # like = vd_info['stat']['like'] # 点赞
author = vd_info["owner"]["name"] # # danmu = vd_info['stat']['danmaku'] # 弹幕
reply = vd_info["stat"]["reply"] # 回复 # date = time.strftime("%Y-%m-%d", time.localtime(vd_info["ctime"]))
favorite = vd_info["stat"]["favorite"] # 收藏 # try:
coin = vd_info["stat"]["coin"] # 投币 # await parse_bilibili_json.send(
# like = vd_info['stat']['like'] # 点赞 # "[[_task|bilibili_parse]]" +
# danmu = vd_info['stat']['danmaku'] # 弹幕 # image(vd_info["pic"]) + f"\nav{aid}\n标题：{title}\n"
date = time.strftime("%Y-%m-%d", time.localtime(vd_info["ctime"])) # f"UP：{author}\n"
try: # f"上传日期：{date}\n"
await parse_bilibili_json.send( # f"回复：{reply}，收藏：{favorite}，投币：{coin}\n"
"[[_task|bilibili_parse]]" + # f"{url}"
image(vd_info["pic"]) + f"\nav{aid}\n标题：{title}\n" # )
f"UP：{author}\n" # except ActionFailed:
f"上传日期：{date}\n" # logger.warning(f"{event.group_id} 发送bilibili解析失败")
f"回复：{reply}，收藏：{favorite}，投币：{coin}\n" # else:
f"{url}" # logger.info(
) # f"USER {event.user_id} GROUP {event.group_id} 解析bilibili转发 {url}"
except ActionFailed: # )
logger.warning(f"{event.group_id} 发送bilibili解析失败") #
else: #
logger.info( # def resize(path: str):
f"USER {event.user_id} GROUP {event.group_id} 解析bilibili转发 {url}" # A = BuildImage(0, 0, background=path, ratio=0.5)
) # A.save(path)
def resize(path: str):
A = BuildImage(0, 0, background=path, ratio=0.5)
A.save(path)

6
plugins/weather/__init__.py
View File

@ -12,7 +12,7 @@ __plugin_usage__ = """
usage： usage：
普普通通的查天气吧 普普通通的查天气吧
指令： 指令：
[城市]天气/天气[城市] [城市]天气
""".strip() """.strip()
__plugin_des__ = "出门要看看天气，不要忘了带伞" __plugin_des__ = "出门要看看天气，不要忘了带伞"
__plugin_cmd__ = ["[城市]天气/天气[城市]"] __plugin_cmd__ = ["[城市]天气/天气[城市]"]
@ -27,13 +27,13 @@ __plugin_settings__ = {
} }
weather = on_regex(r".{0,10}?(.*)的?天气.*?.{0,10}", priority=5, block=True) weather = on_regex(r".{0,10}?(.*)的?天气.{0,10}", priority=5, block=True)
@weather.handle() @weather.handle()
async def _(event: MessageEvent, reg_group: Tuple[Any, ...] = RegexGroup()): async def _(event: MessageEvent, reg_group: Tuple[Any, ...] = RegexGroup()):
msg = reg_group[0] msg = reg_group[0]
if msg[-1] != "市": if msg and msg[-1] != "市":
msg += "市" msg += "市"
city = "" city = ""
if msg: if msg:

4
utils/http_utils.py
View File

@ -327,14 +327,14 @@ class AsyncPlaywright:
await page.set_viewport_size(viewport_size) await page.set_viewport_size(viewport_size)
if isinstance(element, str): if isinstance(element, str):
if wait_time: if wait_time:
card = await page.wait_for_selector(element, timeout=wait_time) card = await page.wait_for_selector(element, timeout=wait_time * 1000)
else: else:
card = await page.query_selector(element) card = await page.query_selector(element)
else: else:
card = page card = page
for e in element: for e in element:
if wait_time: if wait_time:
card = await card.wait_for_selector(e, timeout=wait_time) card = await card.wait_for_selector(e, timeout=wait_time * 1000)
else: else:
card = await card.query_selector(e) card = await card.query_selector(e)
await card.screenshot(path=path, timeout=timeout, type=type_) await card.screenshot(path=path, timeout=timeout, type=type_)