mirror of
https://github.com/zhenxun-org/zhenxun_bot.git
synced 2026-10-03 19:00:00 +08:00
Delete plugins directory
This commit is contained in:
HibiKier
@@ -1,149 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import traceback
|
||||
from typing import Dict
|
||||
|
||||
from aiohttp.client_exceptions import ClientError
|
||||
from nonebot.plugin import on_command, on_message
|
||||
from nonebot.adapters.cqhttp import Bot, MessageEvent, GroupMessageEvent
|
||||
from nonebot.typing import T_State
|
||||
from services.log import logger
|
||||
from util.utils import get_message_text, get_message_imgs
|
||||
from configs.config import MAX_FIND_IMG_COUNT
|
||||
|
||||
from .ex import get_des as get_des_ex
|
||||
from .iqdb import get_des as get_des_iqdb
|
||||
from .saucenao import get_des as get_des_sau
|
||||
from .ascii2d import get_des as get_des_asc
|
||||
from .trace import get_des as get_des_trace
|
||||
from .yandex import get_des as get_des_yandex
|
||||
|
||||
|
||||
async def get_des(url: str, mode: str, user_id: int):
|
||||
"""
|
||||
:param url: 图片链接
|
||||
:param mode: 图源
|
||||
:return:
|
||||
"""
|
||||
if mode == "iqdb":
|
||||
async for msg in get_des_iqdb(url):
|
||||
yield msg
|
||||
elif mode == "ex":
|
||||
async for msg in get_des_ex(url):
|
||||
yield msg
|
||||
elif mode == "trace":
|
||||
async for msg in get_des_trace(url):
|
||||
yield msg
|
||||
elif mode == "yandex":
|
||||
async for msg in get_des_yandex(url):
|
||||
yield msg
|
||||
elif mode.startswith("asc"):
|
||||
async for msg in get_des_asc(url, user_id):
|
||||
yield msg
|
||||
else:
|
||||
async for msg in get_des_sau(url, user_id):
|
||||
yield msg
|
||||
|
||||
|
||||
setu = on_command("识图", aliases={"search"}, block=True, priority=5)
|
||||
|
||||
|
||||
@setu.handle()
|
||||
async def handle_first_receive(bot: Bot, event: MessageEvent, state: T_State):
|
||||
msg = get_message_text(event.json())
|
||||
imgs = get_message_imgs(event.json())
|
||||
if msg in ['帮助']:
|
||||
await setu.finish('示例：\n\t识图 (图片)\n\t识图asc (图片)')
|
||||
if imgs:
|
||||
state["setu"] = imgs[0]
|
||||
if msg:
|
||||
state["mod"] = msg
|
||||
|
||||
|
||||
# ex/nao/trace/iqdb/ascii2d
|
||||
# @setu.got("mod", prompt="从哪里查找呢? ex/nao/trace/iqdb/ascii2d")
|
||||
# async def get_func(bot: Bot, event: MessageEvent, state: dict):
|
||||
# pass
|
||||
|
||||
@setu.args_parser
|
||||
async def get_setu(bot: Bot, event: MessageEvent, state: T_State):
|
||||
imgs = get_message_imgs(event.json())
|
||||
msg = get_message_text(event.json())
|
||||
if not imgs:
|
||||
await setu.reject()
|
||||
if msg:
|
||||
state['mod'] = msg
|
||||
state["setu"] = imgs[0]
|
||||
|
||||
|
||||
@setu.got("setu", prompt="图呢？")
|
||||
async def get_setu(bot: Bot, event: MessageEvent, state: T_State):
|
||||
"""
|
||||
发现没有的时候要发问
|
||||
:return:
|
||||
"""
|
||||
url: str = state["setu"]
|
||||
mod: str = state["mod"] if state.get("mod") else "nao" # 模式
|
||||
try:
|
||||
await bot.send(event=event, message="正在处理图片")
|
||||
idx = 1
|
||||
async for msg in get_des(url, mod, event.user_id):
|
||||
await bot.send(event=event, message=msg)
|
||||
if idx == MAX_FIND_IMG_COUNT:
|
||||
break
|
||||
idx += 1
|
||||
logger.info(f"(USER {event.user_id}, GROUP "
|
||||
f"{event.group_id if event.message_type != 'private' else 'private'}) 识图:{url}")
|
||||
# image_data: List[Tuple] = await get_pic_from_url(url)
|
||||
# await setu.finish("hso")
|
||||
except IndexError:
|
||||
# await bot.send(event, traceback.format_exc())
|
||||
await setu.finish("参数错误")
|
||||
except ClientError:
|
||||
await setu.finish("连接失败")
|
||||
|
||||
|
||||
pic_map: Dict[str, str] = {} # 保存这个群的其阿金一张色图 {"123456":http://xxx"}
|
||||
|
||||
|
||||
async def check_pic(bot: Bot, event: MessageEvent, state: T_State) -> bool:
|
||||
if isinstance(event, MessageEvent):
|
||||
for msg in event.message:
|
||||
if msg.type == "image":
|
||||
url: str = msg.data["url"]
|
||||
state["url"] = url
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
notice_pic = on_message(check_pic, block=False, priority=1)
|
||||
|
||||
|
||||
@notice_pic.handle()
|
||||
async def handle_pic(bot: Bot, event: GroupMessageEvent, state: T_State):
|
||||
try:
|
||||
group_id: str = str(event.group_id)
|
||||
pic_map.update({group_id: state["url"]})
|
||||
except AttributeError:
|
||||
pass
|
||||
|
||||
|
||||
previous = on_command("上一张图是什么", aliases={"上一张", "这是什么"})
|
||||
|
||||
|
||||
@previous.handle()
|
||||
async def handle_previous(bot: Bot, event: GroupMessageEvent, state: T_State):
|
||||
await bot.send(event=event, message="processing...")
|
||||
try:
|
||||
url: str = pic_map[str(event.group_id)]
|
||||
idx = 1
|
||||
async for msg in get_des(url, "nao", event.user_id):
|
||||
await bot.send(event=event, message=msg)
|
||||
if idx == MAX_FIND_IMG_COUNT:
|
||||
break
|
||||
idx += 1
|
||||
except IndexError:
|
||||
await previous.finish("参数错误")
|
||||
except ClientError:
|
||||
await previous.finish("连接错误")
|
||||
except KeyError:
|
||||
await previous.finish("没有图啊QAQ")
|
||||
@@ -1,63 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from typing import List, Tuple
|
||||
from urllib.parse import urljoin
|
||||
import aiofiles
|
||||
from util.utils import get_local_proxy
|
||||
from util.user_agent import get_user_agent
|
||||
from configs.path_config import IMAGE_PATH
|
||||
from asyncio.exceptions import TimeoutError
|
||||
from util.init_result import image
|
||||
|
||||
from lxml.html import fromstring
|
||||
import aiohttp
|
||||
|
||||
|
||||
def parse_html(html: str):
|
||||
selector = fromstring(html)
|
||||
for tag in selector.xpath('//div[@class="container"]/div[@class="row"]/div/div[@class="row item-box"]')[1:5]:
|
||||
if pic_url := tag.xpath('./div/img[@loading="lazy"]/@src'): # 缩略图url
|
||||
pic_url = urljoin("https://ascii2d.net/", pic_url[0])
|
||||
if description := tag.xpath('./div/div/h6/a[1]/text()'): # 名字
|
||||
description = description[0]
|
||||
if author := tag.xpath('./div/div/h6/a[2]/text()'): # 作者
|
||||
author = author[0]
|
||||
if origin_url := tag.xpath('./div/div/h6/a[1]/@href'): # 原图地址
|
||||
origin_url = origin_url[0]
|
||||
if author_url := tag.xpath('./div/div/h6/a[2]/@href'): # 作者地址
|
||||
author_url = author_url[0]
|
||||
yield pic_url, description, author, origin_url, author_url
|
||||
|
||||
pass
|
||||
|
||||
|
||||
async def get_pic_from_url(url: str):
|
||||
real_url = "https://ascii2d.net/search/url/" + url
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(real_url) as resp:
|
||||
html: str = await resp.text()
|
||||
return [i for i in parse_html(html)]
|
||||
|
||||
|
||||
async def get_des(url: str, user_id):
|
||||
image_data: List[Tuple] = await get_pic_from_url(url)
|
||||
if not image_data:
|
||||
msg: str = "找不到高相似度的"
|
||||
yield msg
|
||||
return
|
||||
for pic in image_data:
|
||||
msg = await download_img(pic[0], user_id) + "\n"
|
||||
for i in pic[1:]:
|
||||
msg = msg + f"{i}\n"
|
||||
yield msg
|
||||
|
||||
|
||||
async def download_img(url, user_id):
|
||||
try:
|
||||
async with aiohttp.ClientSession(headers=get_user_agent()) as session:
|
||||
async with session.get(url, proxy=get_local_proxy(), timeout=7) as response:
|
||||
async with aiofiles.open(IMAGE_PATH + f'temp/{user_id}_pic_find.png', 'wb') as f:
|
||||
await f.write(await response.read())
|
||||
return image(f'{user_id}_pic_find.png', 'temp')
|
||||
except TimeoutError:
|
||||
return image(url)
|
||||
|
||||
@@ -1,107 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from base64 import b64encode
|
||||
from typing import List, Tuple
|
||||
import io
|
||||
|
||||
from lxml.html import fromstring
|
||||
import aiohttp
|
||||
import nonebot
|
||||
from aiohttp.client_exceptions import InvalidURL
|
||||
from nonebot.adapters.cqhttp import MessageSegment
|
||||
|
||||
from .formdata import FormData
|
||||
|
||||
driver = nonebot.get_driver()
|
||||
cookie: str = driver.config.ex_cookie
|
||||
proxy: str = driver.config.proxy
|
||||
target: str = "https://exhentai.org/upload/image_lookup.php"
|
||||
|
||||
headers = {
|
||||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
|
||||
'Accept-Encoding': 'gzip, deflate',
|
||||
'Accept-Language': 'zh-CN,zh;q=0.9',
|
||||
'Cache-Control': 'max-age=0',
|
||||
'Connection': 'keep-alive',
|
||||
'Content-Type': 'multipart/form-data; boundary=----WebKitFormBoundaryB0NrMSYMfjY5r0l1',
|
||||
'Host': 'exhentai.org',
|
||||
'Origin': 'https://exhentai.org',
|
||||
'Referer': 'https://exhentai.org/?filesearch=1',
|
||||
'Sec-Fetch-Dest': 'document',
|
||||
'Sec-Fetch-Mode': 'navigate',
|
||||
'Sec-Fetch-Site': 'same-origin',
|
||||
'Sec-Fetch-User': '?1',
|
||||
'Upgrade-Insecure-Requests': '1',
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.105 Safari/537.36'}
|
||||
|
||||
if cookie:
|
||||
headers['Cookie'] = cookie
|
||||
else:
|
||||
headers['Host'] = 'e-hentai.org'
|
||||
headers['Origin'] = 'https://e-hentai.org'
|
||||
headers['Referer'] = 'https://e-hentai.org/?filesearch=1'
|
||||
target: str = "https://e-hentai.org/upload/image_lookup.php"
|
||||
|
||||
|
||||
def parse_html(html: str):
|
||||
"""
|
||||
解析exhentai返回的数据
|
||||
:param html:
|
||||
:return:
|
||||
"""
|
||||
selector = fromstring(html)
|
||||
hrefs = selector.xpath('//td[@class="gl3c glname"]/a/@href')
|
||||
names = selector.xpath('//td[@class="gl3c glname"]/a/div[1]/text()')
|
||||
pics = selector.xpath('//tr/td[@class="gl2c"]/div[@class="glthumb"]/div[1]/img/@src') # 缩略图
|
||||
for name, href, pic in zip(names, hrefs, pics):
|
||||
yield name, href, pic
|
||||
|
||||
|
||||
async def get_pic_from_url(url: str):
|
||||
"""
|
||||
从接受到的picurl获取图片信息
|
||||
:param url:
|
||||
:return:
|
||||
"""
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as resp:
|
||||
content = io.BytesIO(await resp.read())
|
||||
# Content_Length = resp.content_length
|
||||
data = FormData(boundary="----WebKitFormBoundaryB0NrMSYMfjY5r0l1")
|
||||
data.add_field(name="sfile", value=content, content_type="image/jpeg",
|
||||
filename="0.jpg")
|
||||
data.add_field(name="f_sfile", value="search")
|
||||
data.add_field(name="fs_similar", value="on")
|
||||
async with session.post(target, data=data, headers=headers, proxy=proxy) as res:
|
||||
html = await res.text()
|
||||
return [i for i in parse_html(html)]
|
||||
|
||||
|
||||
async def get_content_from_url(url: str):
|
||||
"""
|
||||
从url 获得b64 encode
|
||||
:param url:
|
||||
:return:
|
||||
"""
|
||||
try:
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url, headers=headers) as resp:
|
||||
return "base64://" + b64encode(await resp.read()).decode()
|
||||
except aiohttp.client_exceptions.InvalidURL:
|
||||
return url
|
||||
|
||||
|
||||
async def get_des(url: str):
|
||||
"""
|
||||
迭代要发送的信息
|
||||
:param url:
|
||||
:return:
|
||||
"""
|
||||
image_data: List[Tuple] = await get_pic_from_url(url)
|
||||
if not image_data:
|
||||
msg: str = "找不到高相似度的"
|
||||
yield msg
|
||||
return
|
||||
for name, href, pic_url in image_data:
|
||||
content = await get_content_from_url(pic_url)
|
||||
msg = MessageSegment.image(file=content) + f"\n本子名称：{name}\n" + f"链接{href}\n"
|
||||
yield msg
|
||||
@@ -1,27 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from typing import Any, Iterable, List, Optional
|
||||
|
||||
from aiohttp import FormData as _FormData
|
||||
import aiohttp.multipart as multipart
|
||||
|
||||
|
||||
class FormData(_FormData):
|
||||
def __init__(
|
||||
self,
|
||||
fields: Iterable[Any] = (),
|
||||
quote_fields: bool = True,
|
||||
charset: Optional[str] = None,
|
||||
boundary: Optional[str] = None
|
||||
) -> None:
|
||||
self._writer = multipart.MultipartWriter("form-data", boundary=boundary)
|
||||
self._fields = [] # type: List[Any]
|
||||
self._is_multipart = False
|
||||
self._is_processed = False
|
||||
self._quote_fields = quote_fields
|
||||
self._charset = charset
|
||||
|
||||
if isinstance(fields, dict):
|
||||
fields = list(fields.items())
|
||||
elif not isinstance(fields, (list, tuple)):
|
||||
fields = (fields,)
|
||||
self.add_fields(*fields)
|
||||
@@ -1,79 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import asyncio
|
||||
from typing import List, Tuple
|
||||
import io
|
||||
|
||||
from urllib.parse import urljoin
|
||||
from lxml.html import fromstring
|
||||
import aiohttp
|
||||
from nonebot.adapters.cqhttp import MessageSegment
|
||||
|
||||
from .formdata import FormData
|
||||
|
||||
headers = {
|
||||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
|
||||
'Accept-Encoding': 'gzip, deflate', 'Accept-Language': 'zh-CN,zh;q=0.9', 'Cache-Control': 'max-age=0',
|
||||
'Connection': 'keep-alive',
|
||||
'Content-Type': 'multipart/form-data; boundary=----WebKitFormBoundaryuwjSiBcpPag4k159',
|
||||
'Cookie': 'Hm_lvt_765ecde8c11b85f1ac5f168fa6e6821f=1602471368; Hm_lpvt_765ecde8c11b85f1ac5f168fa6e6821f=1602472300',
|
||||
'Host': 'iqdb.org', 'Origin': 'http://iqdb.org', 'Referer': 'http://iqdb.org/', 'Upgrade-Insecure-Requests': '1',
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/84.0.4147.105 Safari/537.36'}
|
||||
|
||||
|
||||
def parse_html(html: str):
|
||||
selector = fromstring(html)
|
||||
for tag in selector.xpath('//div[@id="pages"]/div[position()>1]/table'):
|
||||
# 第一个是bestmatch
|
||||
if pic_url := tag.xpath('./tr[2]/td/a/img/@src'):
|
||||
pic_url = urljoin("http://iqdb.org/", pic_url[0]) # 缩略图
|
||||
else:
|
||||
pic_url = "没有最相似的"
|
||||
similarity = tag.xpath('./tr[last()]/td/text()')[0] # 相似度
|
||||
href: List[str] = tag.xpath('./tr/td/a/@href') # 第一个href
|
||||
href.extend(tag.xpath('./tr/td/span/a/@href')) # 第二个 可能是空
|
||||
href = list(map(lambda x: "https:" + x if not x.startswith("https") else x, href))
|
||||
yield pic_url, similarity, href
|
||||
|
||||
pass
|
||||
|
||||
|
||||
async def get_pic_from_url(url: str):
|
||||
"""
|
||||
返回信息元祖
|
||||
:param url:
|
||||
:return:
|
||||
"""
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as resp:
|
||||
content = io.BytesIO(await resp.read())
|
||||
data = FormData(boundary="----WebKitFormBoundaryuwjSiBcpPag4k159")
|
||||
data.add_field(name="MAX_FILE_SIZE", value="")
|
||||
for i in range(1, 7):
|
||||
data.add_field(name="service[]", value=str(i))
|
||||
data.add_field(name="service[]", value="11")
|
||||
data.add_field(name="service[]", value="13")
|
||||
data.add_field(name="file", value=content, content_type="application/octet-stream", filename="0.jpg")
|
||||
data.add_field(name="url", value="")
|
||||
async with session.post("http://iqdb.org/", data=data, headers=headers) as res:
|
||||
html = await res.text()
|
||||
return [i for i in parse_html(html)]
|
||||
pass
|
||||
|
||||
|
||||
async def get_des(url: str):
|
||||
"""
|
||||
返回详细简介 cq码转义
|
||||
:param url:
|
||||
:return:
|
||||
"""
|
||||
image_data: List[Tuple] = await get_pic_from_url(url)
|
||||
if not image_data:
|
||||
msg: str = "找不到高相似度的"
|
||||
yield msg
|
||||
return
|
||||
for pic in image_data:
|
||||
msg = MessageSegment.image(file=pic[0]) + f"\n{pic[1]}\n"
|
||||
for i in pic[2]:
|
||||
msg = msg + f"{i}\n"
|
||||
yield msg
|
||||
|
||||
@@ -1,116 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import io
|
||||
from typing import List, Tuple, Union
|
||||
import aiofiles
|
||||
from util.utils import get_local_proxy
|
||||
from util.user_agent import get_user_agent
|
||||
from configs.path_config import IMAGE_PATH
|
||||
from asyncio.exceptions import TimeoutError
|
||||
from util.init_result import image
|
||||
|
||||
import aiohttp
|
||||
from lxml.html import fromstring
|
||||
from nonebot.adapters.cqhttp import Message
|
||||
|
||||
from .formdata import FormData
|
||||
|
||||
header = {
|
||||
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.9',
|
||||
'Accept-Encoding': 'gzip, deflate', 'Accept-Language': 'zh-CN,zh;q=0.9',
|
||||
'Cache-Control': 'max-age=0',
|
||||
"Content-Type": "multipart/form-data; boundary=----WebKitFormBoundaryPpuR3EZ1Ap2pXv8W",
|
||||
'Connection': 'keep-alive',
|
||||
'Host': 'saucenao.com', 'Origin': 'https://saucenao.com', 'Referer': 'https://saucenao.com/index.php',
|
||||
'Sec-Fetch-Dest': 'document', 'Sec-Fetch-Mode': 'navigate', 'Sec-Fetch-Site': 'same-origin', 'Sec-Fetch-User': '?1',
|
||||
'Upgrade-Insecure-Requests': '1',
|
||||
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.163 Safari/537.36'}
|
||||
|
||||
|
||||
def parse_html(html: str):
|
||||
"""
|
||||
解析nao返回的html
|
||||
:param html:
|
||||
:return:
|
||||
"""
|
||||
selector = fromstring(html)
|
||||
for tag in selector.xpath('//div[@class="result"]/table'):
|
||||
pic_url = tag.xpath('./tr/td/div/a/img/@src')
|
||||
if pic_url:
|
||||
pic_url = pic_url[0]
|
||||
else:
|
||||
pic_url = None # 相似度
|
||||
xsd: List[str] = tag.xpath(
|
||||
'./tr/td[@class="resulttablecontent"]/div[@class="resultmatchinfo"]/div[@class="resultsimilarityinfo"]/text()')
|
||||
if xsd:
|
||||
xsd = xsd[0]
|
||||
else:
|
||||
xsd = "没有写" # 相似度
|
||||
title: List[str] = tag.xpath(
|
||||
'./tr/td[@class="resulttablecontent"]/div[@class="resultcontent"]/div[@class="resulttitle"]/strong/text()')
|
||||
if title:
|
||||
title = title[0]
|
||||
else:
|
||||
title = "没有写" # 标题
|
||||
# pixiv id
|
||||
pixiv_id: List[str] = tag.xpath(
|
||||
'./tr/td[@class="resulttablecontent"]/div[@class="resultcontent"]/div[@class="resultcontentcolumn"]/a[1]/@href')
|
||||
if pixiv_id:
|
||||
pixiv_id = pixiv_id[0]
|
||||
else:
|
||||
pixiv_id = "没有说"
|
||||
member: List[str] = tag.xpath(
|
||||
'./tr/td[@class="resulttablecontent"]/div[@class="resultcontent"]/div[@class="resultcontentcolumn"]/a[2]/@href')
|
||||
if member:
|
||||
member = member[0]
|
||||
else:
|
||||
member = "没有说"
|
||||
yield pic_url, xsd, title, pixiv_id, member
|
||||
|
||||
|
||||
async def get_pic_from_url(url: str):
|
||||
"""
|
||||
从url搜图
|
||||
:param url:
|
||||
:return:
|
||||
"""
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as resp:
|
||||
content = io.BytesIO(await resp.read())
|
||||
data = FormData(boundary="----WebKitFormBoundaryPpuR3EZ1Ap2pXv8W")
|
||||
data.add_field(name="file", value=content, content_type="image/jpeg",
|
||||
filename="blob")
|
||||
async with session.post("https://saucenao.com/search.php", data=data, headers=header) as res:
|
||||
html = await res.text()
|
||||
image_data = [each for each in parse_html(html)]
|
||||
return image_data
|
||||
|
||||
|
||||
async def get_des(url: str, user_id: int):
|
||||
image_data: List[Tuple] = await get_pic_from_url(url)
|
||||
if not image_data:
|
||||
msg: Union[str, Message] = "找不到高相似度的"
|
||||
yield msg
|
||||
return
|
||||
for pic in image_data:
|
||||
# print(pic)
|
||||
msg = await download_img(pic[0], user_id) \
|
||||
+ f"\n相似度:{pic[1]}" \
|
||||
f"\n标题:{pic[2] if (str(pic[2]).strip() != 'Creator:' and len(str(pic[2]).split('-')) < 3) else '未知'}" \
|
||||
f"\nPID:{pic[3]}" \
|
||||
f"\nmember:{pic[4]}\n"
|
||||
yield msg
|
||||
pass
|
||||
|
||||
|
||||
async def download_img(url, user_id):
|
||||
try:
|
||||
async with aiohttp.ClientSession(headers=get_user_agent()) as session:
|
||||
async with session.get(url, proxy=get_local_proxy(), timeout=7) as response:
|
||||
async with aiofiles.open(IMAGE_PATH + f'temp/{user_id}_pic_find.png', 'wb') as f:
|
||||
await f.write(await response.read())
|
||||
return image(f'{user_id}_pic_find.png', 'temp')
|
||||
except TimeoutError:
|
||||
return image(url)
|
||||
|
||||
|
||||
|
||||
@@ -1,89 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
import io
|
||||
from copy import deepcopy
|
||||
|
||||
from base64 import b64encode
|
||||
from typing import List, Tuple
|
||||
|
||||
import aiohttp
|
||||
from nonebot.adapters.cqhttp import MessageSegment
|
||||
|
||||
header = {':authority': 'api.trace.moe',
|
||||
'accept': '*/*',
|
||||
'accept-encoding': 'gzip, deflate, br',
|
||||
'accept-language': 'zh-CN,zh;q=0.9',
|
||||
'content-type': 'multipart/form-data; boundary=----WebKitFormBoundary9cyjY8YBBN8SGdG4',
|
||||
'origin': 'https://trace.moe',
|
||||
'sec-fetch-dest': 'empty', 'sec-fetch-mode': 'cors', 'sec-fetch-site': 'same-site',
|
||||
'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
|
||||
'Chrome/84.0.4147.105 Safari/537.36'}
|
||||
|
||||
|
||||
async def parse_json(session: aiohttp.ClientSession, data: dict):
|
||||
count = 0
|
||||
for i in data["result"]:
|
||||
title: dict = i["anilist"]["title"]
|
||||
similarity = i["similarity"]
|
||||
from_ = i["from"]
|
||||
to = i["to"]
|
||||
file = i["filename"] # 名字
|
||||
is_adult = i["anilist"]["isAdult"]
|
||||
episode = i["episode"] # 集
|
||||
header_new = deepcopy(header)
|
||||
del header_new["content-type"]
|
||||
header_new[":method"] = 'GET'
|
||||
header_new["accept"] = "image/webp,image/apng,image/*,*/*;q=0.8"
|
||||
header_new["sec-fetch-dest"] = "image"
|
||||
header_new["sec-fetch-mode"] = "no-cors"
|
||||
async with session.get(i["image"], headers=header_new) as resp:
|
||||
pic = "base64://" + b64encode(await resp.read()).decode()
|
||||
yield pic, similarity, file, is_adult, from_, to, title, episode
|
||||
count += 1
|
||||
if count > 4:
|
||||
break
|
||||
|
||||
|
||||
# POST https://api.trace.moe/search?cutBorders=1&anilistID=
|
||||
async def get_pic_from_url(url: str):
|
||||
"""
|
||||
从url搜图
|
||||
:param url:
|
||||
:return:
|
||||
"""
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(url) as resp:
|
||||
content = io.BytesIO(await resp.read())
|
||||
# with open("F:\elu.PNG", "rb") as f:
|
||||
# content = io.BytesIO(f.read())
|
||||
data = aiohttp.FormData(boundary="----WebKitFormBoundary9cyjY8YBBN8SGdG4")
|
||||
data.add_field(name="image", value=content, content_type="image/jpeg",
|
||||
filename="blob")
|
||||
# data.add_field(name="filter", value="")
|
||||
# data.add_field(name="trial", value="0")
|
||||
async with session.post("https://api.trace.moe/search?cutBorders=1&anilistID=", data=data,
|
||||
headers=header) as res:
|
||||
data: dict = await res.json()
|
||||
image_data = [each async for each in parse_json(session, data)]
|
||||
return image_data
|
||||
|
||||
|
||||
async def get_des(url: str):
|
||||
image_data: List[Tuple] = await get_pic_from_url(url)
|
||||
if not image_data:
|
||||
msg: str = "找不到高相似度的"
|
||||
yield msg
|
||||
return
|
||||
for pic in image_data:
|
||||
msg = MessageSegment.image(
|
||||
file=pic[
|
||||
0]) + f"\n相似度:{pic[1]}%\n标题:{pic[6]['native'] + ' ' + pic[6]['chinese']}\n第{pic[7]}集\nR18:{pic[3]}\n开始时间:{pic[4]}s\n结束时间{pic[5]}s"
|
||||
yield msg
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
import asyncio
|
||||
|
||||
data = asyncio.run(get_pic_from_url(
|
||||
"https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1606681978562&di=6d6c90aef5ff1f9f8915bbc2e18e3c98&imgtype=0&src=http%3A%2F%2Fc-ssl.duitang.com%2Fuploads%2Fblog%2F202011%2F15%2F20201115190356_c5b95.thumb.1000_0.jpg"))
|
||||
pass
|
||||
@@ -1,53 +0,0 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
from typing import List, Tuple
|
||||
|
||||
import nonebot
|
||||
from nonebot.adapters.cqhttp import MessageSegment
|
||||
from lxml.html import fromstring
|
||||
import aiohttp
|
||||
|
||||
"""
|
||||
http://yandex.com/clck/jsredir?from=yandex.com%3Bimages%2Fsearch%3Bimages%3B%3B&text=&etext=9185.K4iyzsNBG9xrJrSJCUTF4i-XPMAfmBQYR_Igss1ESRc.65568e796f3375fae39da91273ae8a1a82410929&uuid=&state=iric5OQ0sS2054x1_o8yG9mmGMT8WeQxqpuwa4Ft4KVzd9aE_Y4Dfw,,&data=eEwyM2lDYU9Gd1VROE1ZMXhZYkJTYW5fZC1TWjIzaFh5TmR1Z09fQm5DdDB3bFJSSUpVdUxfZmUzcVhfaXhTN1BCU2dINGxmdkY4NFVNcHYyUmw0emFKT2pnOWJoVmlPVzAzX1FIbWh6aXVFV3F0YWFaMGdxeGFtY2dxTzFZZl9VY1huZmlLaGVGOFZleUthZXBlM1pxUGM2elVDLXdvZEo3OGJwdVFqYmVkTDJxWElHSzFZR2NVQUhVcTdzelJwSXlrTjhlS0txdHpYY1RMMHRLOU5HSTYtT0VDb0hpdll6YjVYRXNVcUhCRFJaeDExNTQwZlhMdjh4M2YtTVFUbVJ5ZzBxMTVJcG9DNW51UWhvRzE0WjlFS19uS0VUZWhNRGxOZWlPUkFlRUUs&sign=7ba9ee25d3716868ec8464fb766c9e25&keyno=IMGS_0&b64e=2&l10n=en
|
||||
"""
|
||||
|
||||
driver = nonebot.get_driver()
|
||||
proxy: str = driver.config.proxy
|
||||
|
||||
|
||||
def parse_html(html: str):
|
||||
selector = fromstring(html)
|
||||
for item in selector.xpath('//li[@class="other-sites__item"]'):
|
||||
pic_url = item.xpath('./a[@class="other-sites__preview-link"]/img/@src')[0].lstrip("//") # 图床
|
||||
des = item.xpath(
|
||||
'./div[@class="other-sites__snippet"]/div[@class="other-sites__snippet-title"]/a/text()')[0] # 简介
|
||||
url = item.xpath(
|
||||
'./div[@class="other-sites__snippet"]/div[@class="other-sites__snippet-site"]/a/@href')[0] # 链接
|
||||
yield pic_url, des, url
|
||||
|
||||
|
||||
async def get_pic_from_url(url: str):
|
||||
real_url = f"https://yandex.com/images/search?rpt=imageview&url={url}"
|
||||
async with aiohttp.ClientSession() as session:
|
||||
async with session.get(real_url, proxy=proxy) as resp:
|
||||
html: str = await resp.text()
|
||||
return [i for i in parse_html(html)]
|
||||
|
||||
|
||||
async def get_des(url: str):
|
||||
image_data: List[Tuple] = await get_pic_from_url(url)
|
||||
if not image_data:
|
||||
msg: str = "找不到高相似度的"
|
||||
yield msg
|
||||
return
|
||||
for pic in image_data:
|
||||
msg = MessageSegment.image(file=pic[0]) + "\n"
|
||||
for i in pic[1:]:
|
||||
msg = msg + f"{i}\n"
|
||||
yield msg
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
with open("yandex.html", "r", encoding="utf-8") as f:
|
||||
data = f.read()
|
||||
for item in parse_html(data):
|
||||
print(item)
|
||||
Reference in New Issue
Block a user