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

Update pid_search.py

Browse Source
This commit is contained in:
AkashiCoin 2022-05-10 18:47:39 +08:00
parent b3b09ed666
commit c035b1b19c
1 changed files with 106 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

206
plugins/pid_search.py
View File

@ -1,100 +1,106 @@
from asyncio.exceptions import TimeoutError from nonebot import on_command
from nonebot import on_command from nonebot.adapters.onebot.v11 import Bot, MessageEvent, Message, GroupMessageEvent
from nonebot.adapters.onebot.v11 import MessageEvent, Message, GroupMessageEvent from nonebot.typing import T_State
from nonebot.typing import T_State from utils.utils import is_number
from configs.path_config import TEMP_PATH from utils.message_builder import image
from services.log import logger from services.log import logger
from utils.message_builder import image from asyncio.exceptions import TimeoutError
from utils.utils import is_number from configs.path_config import IMAGE_PATH
from utils.manager import withdraw_message_manager from utils.manager import withdraw_message_manager
from utils.http_utils import AsyncHttpx from utils.http_utils import AsyncHttpx
from nonebot.params import CommandArg, Arg from nonebot.params import CommandArg, Arg
try:
import ujson as json __zx_plugin_name__ = "pid搜索"
except ModuleNotFoundError: __plugin_usage__ = """
import json usage：
通过 pid 搜索图片
指令：
__zx_plugin_name__ = "pid搜索" p搜 [pid]
__plugin_usage__ = """ """.strip()
usage： __plugin_des__ = "通过 pid 搜索图片"
通过 pid 搜索图片 __plugin_cmd__ = ["p搜 [pid]"]
指令： __plugin_version__ = 0.1
p搜 [pid] __plugin_author__ = "HibiKier"
""".strip() __plugin_settings__ = {
__plugin_des__ = "通过 pid 搜索图片" "level": 5,
__plugin_cmd__ = ["p搜 [pid]"] "default_status": True,
__plugin_version__ = 0.1 "limit_superuser": False,
__plugin_author__ = "HibiKier" "cmd": ["p搜"],
__plugin_settings__ = { }
"level": 5,
"default_status": True, pid_search = on_command("p搜", aliases={"pixiv搜", "P搜"}, priority=5, block=True)
"limit_superuser": False,
"cmd": ["p搜"], url = "https://api.obfs.dev/api/pixiv/"
}
pid_search = on_command("p搜", aliases={"pixiv搜", "P搜"}, priority=5, block=True) @pid_search.handle()
async def _h(event: MessageEvent, state: T_State, arg: Message = CommandArg()):
url = "https://api.fantasyzone.cc/tu/search.php" pid = arg.extract_plain_text().strip()
if pid:
state["pid"] = pid
@pid_search.handle()
async def _(state: T_State, arg: Message = CommandArg()):
pid = arg.extract_plain_text().strip() headers = {
if pid: "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10.6;"
state["pid"] = pid " rv:2.0.1) Gecko/20100101 Firefox/4.0.1",
"Referer": "https://www.pixiv.net",
}
@pid_search.got("pid", prompt="需要查询的图片PID是？")
async def _(event: MessageEvent, state: T_State, pid: str = Arg("pid")):
if pid in ["取消", "算了"]: @pid_search.got("pid", prompt="需要查询的图片PID是？")
await pid_search.finish("已取消操作...") async def _g(event: MessageEvent, state: T_State, pid: str = Arg("pid")):
if is_number(pid): if pid in ["取消", "算了"]:
await pid_search.reject_arg("pid", "笨蛋，重新输入数！字！") await pid_search.finish("已取消操作...")
params = { # if is_number(pid):
"id": pid, # await pid_search.reject_arg("pid", "笨蛋，重新输入数！字！")
"p": 1, for _ in range(3):
} try:
for _ in range(3): data = (
try: await AsyncHttpx.get(
data = (await AsyncHttpx.get(url, params=params, timeout=5)).json() url,
except TimeoutError: params={"id": pid},
pass timeout=5,
except Exception as e: )
await pid_search.finish(f"发生了一些错误..{type(e)}：{e}") ).json()
else: except TimeoutError:
if not data["width"] and not data["height"]: pass
await pid_search.finish(f"没有搜索到 PID：{pid} 的图片", at_sender=True) except Exception as e:
pid = data["id"] await pid_search.finish(f"发生了一些错误..{type(e)}：{e}")
title = data["title"] else:
author = data["userName"] if not data["illust"]["width"] and not data["illust"]["height"]:
author_id = data["userId"] await pid_search.finish(f"没有搜索到 PID：{pid} 的图片", at_sender=True)
img_url = data["url"] pid = data["illust"]["id"]
if not await AsyncHttpx.download_file( title = data["illust"]["title"]
img_url, TEMP_PATH / f"pid_search_{event.user_id}.png" author = data["illust"]["user"]["name"]
): author_id = data["illust"]["user"]["id"]
await pid_search.finish("图片下载失败了....", at_sender=True) img_url = data["illust"]["meta_single_page"]["original_image_url"]
tmp = "" if not await AsyncHttpx.download_file(
if isinstance(event, GroupMessageEvent): img_url,
tmp = "\n【注】将在30后撤回......" IMAGE_PATH / "temp" / f"pid_search_{event.user_id}.png",
msg_id = await pid_search.send( headers=headers,
Message( ):
f"title：{title}\n" await pid_search.finish("图片下载失败了....", at_sender=True)
f"pid：{pid}\n" tmp = ""
f"author：{author}\n" if isinstance(event, GroupMessageEvent):
f"author_id：{author_id}\n" tmp = "\n【注】将在30后撤回......"
f'{image(TEMP_PATH / f"pid_search_{event.user_id}.png")}' msg_id = await pid_search.send(
f"{tmp}" Message(
) f"title：{title}\n"
) f"pid：{pid}\n"
logger.info( f"author：{author}\n"
f"(USER {event.user_id}, " f"author_id：{author_id}\n"
f"GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'})" f'{image(f"pid_search_{event.user_id}.png", "temp")}'
f" 查询图片 PID：{pid}" f"{tmp}"
) )
if isinstance(event, GroupMessageEvent): )
withdraw_message_manager.append((msg_id, 30)) logger.info(
break f"(USER {event.user_id}, "
else: f"GROUP {event.group_id if isinstance(event, GroupMessageEvent) else 'private'})"
await pid_search.finish("图片下载失败了....", at_sender=True) f" 查询图片 PID：{pid}"
)
if isinstance(event, GroupMessageEvent):
withdraw_message_manager.append((msg_id, 30))
break
else:
await pid_search.finish("图片下载失败了....", at_sender=True)