Diuse/zhenxun_bot
1
0
Fork 0
mirror of https://github.com/zhenxun-org/zhenxun_bot.git synced 2026-09-28 16:20:56 +08:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

update alapi

Browse Source
This commit is contained in:
hibiki
2021-06-24 15:32:06 +08:00
parent 779e5594d3
commit 7ae2c4c8dc
24 changed files with 662 additions and 317 deletions
Download Patch File Download Diff File Expand all files Collapse all files
bot.py
+1
View File
@@ -12,6 +12,7 @@ nonebot.load_builtin_plugins()
nonebot.load_plugins("plugins")
nonebot.load_plugins("plugins/shop")
nonebot.load_plugins("plugins/genshin")
nonebot.load_plugins("plugins/alapi")
if __name__ == "__main__":
configs/config.py
+6 -3
View File
@@ -1,6 +1,6 @@
from .utils.util import get_config_data
from typing import List
from services.service_config import TL_M_KEY, SYSTEM_PROXY
from services.service_config import TL_M_KEY, SYSTEM_PROXY, ALAPI_TOKEN
try:
import ujson as json
except ModuleNotFoundError:
@@ -14,14 +14,15 @@ USE_CONFIG_FILE = False
# API KEY（必要）
LOLICON_KEY: str = "" # lolicon
RSSHUBAPP: str = "https://rsshub.app/" # rsshub
ALAPI_TOKEN: str = "" # ALAPI
# 图灵
TL_KEY: List[str] = []
# 数据库（必要）
# 如果填写了bind就不需要再填写后面的字段了#）
# bind示例：postgresql://user:password@127.0.0.1:5432/databases
# 示例："bind": "postgresql://user:password@127.0.0.1:5432/database"
bind: str = ''
sql_name: str = 'postgresql'
sql_name: str = ''
user: str = ''
password: str = ''
address: str = ''
@@ -194,6 +195,8 @@ if TL_M_KEY:
TL_KEY = TL_M_KEY
if SYSTEM_PROXY:
system_proxy = SYSTEM_PROXY
if ALAPI_TOKEN:
ALAPI_TOKEN = ALAPI_TOKEN
# 配置文件应用
plugins/ai/data_source.py
+24 -1
View File
@@ -1,8 +1,9 @@
from configs.config import TL_KEY
from configs.config import TL_KEY, ALAPI_TOKEN
import aiohttp
import random
import os
from configs.path_config import IMAGE_PATH, DATA_PATH
from services.log import logger
from util.init_result import image
from util.utils import get_bot
@@ -13,6 +14,8 @@ except ModuleNotFoundError:
url = "http://openapi.tuling123.com/openapi/api/v2"
check_url = "https://v2.alapi.cn/api/censor/text"
index = 0
anime_data = json.load(open(DATA_PATH + 'anime.json', 'r', encoding='utf8'))
@@ -97,6 +100,7 @@ async def get_qqbot_chat_result(text: str, img_url: str, user_id: int, user_name
content = content.replace('{br}', '\n')
if content.find('提示') != -1:
content = content[:content.find('提示')]
return content
if resp_payload['results']:
for result in resp_payload['results']:
@@ -140,3 +144,22 @@ def no_result() -> str:
]) + image(
random.choice(os.listdir(IMAGE_PATH + "noresult/")
), "noresult")
async def check_text(text: str):
if not ALAPI_TOKEN:
return text
params = {
'token': ALAPI_TOKEN,
'text': text
}
async with aiohttp.ClientSession() as session:
try:
async with session.get(url, timeout=2, params=params) as response:
data = await response.json()
if data['code'] == 200:
if data['conclusion_type'] == 2:
return no_result()
except Exception as e:
logger.error(f'检测违规文本错误...e：{e}')
return text
plugins/alapi/__init__.py
View File
plugins/alapi/comments_163.py
+36
View File
@@ -0,0 +1,36 @@
from nonebot import on_command
from nonebot.adapters.cqhttp import Bot, MessageEvent
from nonebot.typing import T_State
from configs.config import ALAPI_TOKEN
from .util import get_data
from services.log import logger
__plugin_name__ = '网易云热评'
__plugin_usage__ = '用法： 生了个人，我很抱歉'
comments_163 = on_command("网易云热评", aliases={'网易云评论', '到点了', '12点了'}, priority=5, block=True)
comments_163_url = 'https://v2.alapi.cn/api/comment'
@comments_163.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
params = {
'token': ALAPI_TOKEN
}
data, code = await get_data(comments_163_url, params)
if code != 200:
await comments_163.finish(data, at_sender=True)
data = data['data']
comment = data['comment_content']
song_name = data['title']
await comments_163.send(f'{comment}\n\t——《{song_name}》')
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if event.message_type != 'private' else 'private'})"
f" 发送网易云热评: {comment} \n\t\t————{song_name}")
plugins/alapi/cover.py
+40
View File
@@ -0,0 +1,40 @@
from nonebot import on_command
from nonebot.adapters.cqhttp import Bot, MessageEvent, Message
from nonebot.typing import T_State
from configs.config import ALAPI_TOKEN
from util.init_result import image
from util.utils import get_message_text
from .util import get_data
from services.log import logger
__plugin_name__ = 'b封面'
__plugin_usage__ = '用法： b封面 (链接，av，bv，cv，直播id)\n\t' \
'示例：b封面 av86863038'
cover = on_command('b封面', priority=5, block=True)
cover_url = 'https://v2.alapi.cn/api/bilibili/cover'
@cover.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
msg = get_message_text(event.json())
params = {
'token': ALAPI_TOKEN,
'c': msg
}
data, code = await get_data(cover_url, params)
if code != 200:
await cover.finish(data, at_sender=True)
data = data['data']
title = data['title']
img = data['cover']
await cover.send(Message(f'title：{title}\n{image(img)}'))
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if event.message_type != 'private' else 'private'})"
f" 获取b站封面: {title} url：{img}")
plugins/alapi/poetry.py
+36
View File
@@ -0,0 +1,36 @@
from nonebot import on_command
from nonebot.adapters.cqhttp import Bot, MessageEvent
from nonebot.typing import T_State
from configs.config import ALAPI_TOKEN
from services.log import logger
from .util import get_data
__plugin_name__ = '古诗'
__plugin_usage__ = '用法： 无'
poetry = on_command("念诗", aliases={'来首诗', '念首诗'}, priority=5, block=True)
poetry_url = 'https://v2.alapi.cn/api/shici'
@poetry.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
params = {
'token': ALAPI_TOKEN
}
data, code = await get_data(poetry_url, params)
if code != 200:
await poetry.finish(data, at_sender=True)
data = data['data']
content = data['content']
title = data['origin']
author = data['author']
await poetry.send(f'{content}\n\t——{author}《{title}》')
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if event.message_type != 'private' else 'private'})"
f" 发送古诗: f'{content}\n\t--{author}《{title}》'")
plugins/alapi/util.py
+17
View File
@@ -0,0 +1,17 @@
import aiohttp
from util.utils import get_local_proxy
async def get_data(url: str, params: dict):
async with aiohttp.ClientSession() as session:
try:
async with session.get(url, proxy=get_local_proxy(), timeout=2, params=params) as response:
data = await response.json()
if data['code'] == 200:
if not data['data']:
return '没有搜索到...', 997
return data, 200
else:
return f'发生了错误...code：{data["code"]}', 999
except TimeoutError:
return '超时了....', 998
plugins/draw_card/announcement.py
+120 -119
View File
@@ -18,7 +18,7 @@ genshin_up_char = Path(DRAW_PATH + "/draw_card_up/genshin_up_char.json")
pretty_up_char = Path(DRAW_PATH + "/draw_card_up/pretty_up_char.json")
guardian_up_char = Path(DRAW_PATH + "/draw_card_up/guardian_up_char.json")
prts_url = "https://wiki.biligame.com/arknights/%E6%96%B0%E9%97%BB%E5%85%AC%E5%91%8A"
prts_url = "https://ak.hypergryph.com/news.html"
genshin_url = "https://wiki.biligame.com/ys/%E7%A5%88%E6%84%BF"
pretty_url = "https://wiki.biligame.com/umamusume/%E5%85%AC%E5%91%8A"
guardian_url = "https://wiki.biligame.com/gt/%E9%A6%96%E9%A1%B5"
@@ -36,19 +36,10 @@ def is_expired(data: dict):
# 检查写入
def check_write(data: dict, up_char_file, game_name: str = ''):
tmp = data
if game_name in ['genshin', 'pretty', 'guardian']:
tmp = data['char']
if not is_expired(tmp):
if game_name in ['genshin', 'guardian']:
data['char']['title'] = ''
data['arms']['title'] = ''
elif game_name in ['pretty']:
data['char']['title'] = ''
data['card']['title'] = ''
else:
data['title'] = ''
def check_write(data: dict, up_char_file):
if not is_expired(data['char']):
for x in list(data.keys()):
data[x]['title'] = ''
else:
with open(up_char_file, 'w', encoding='utf8') as f:
json.dump(data, f, indent=4, ensure_ascii=False)
@@ -58,10 +49,7 @@ def check_write(data: dict, up_char_file, game_name: str = ''):
else:
with open(up_char_file, 'r', encoding='utf8') as f:
old_data = json.load(f)
tmp = old_data
if game_name in ['genshin', 'pretty', 'guardian']:
tmp = old_data['char']
if is_expired(tmp):
if is_expired(data['char']):
return old_data
else:
with open(up_char_file, 'w', encoding='utf8') as f:
@@ -69,100 +57,104 @@ def check_write(data: dict, up_char_file, game_name: str = ''):
return data
def _get_up_char(r: str, text: str):
pr = re.search(r, text)
chars = pr.group(1)
probability = pr.group(2)
chars = chars.replace('[限定]', '').replace('[', '').replace(']', '')
probability = probability.replace('【', '')
return chars, probability
class PrtsAnnouncement:
@staticmethod
async def get_announcement_text():
def __init__(self):
self.game_name = '明日方舟'
async def _get_announcement_text(self):
async with aiohttp.ClientSession(headers=headers) as session:
async with session.get(prts_url, timeout=7) as res:
soup = BeautifulSoup(await res.text(), 'lxml')
trs = soup.find('table').find('tbody').find_all('tr')
for tr in trs:
a = tr.find_all('td')[-1].find('a')
if a.text.find('寻访') != -1:
url = a.get('href')
break
async with session.get(f'https://wiki.biligame.com/{url}', timeout=7) as res:
return await res.text(), a.text[:-4]
ol = soup.find('ol', {'class': 'articleList active', 'data-category-key': 'LATEST'})
for li in ol:
itype = li.find('span', {'class': 'articleItemCate'}).text
if itype == '活动':
a = li.find('a')['href']
async with session.get(f'https://ak.hypergryph.com{a}', headers=headers, timeout=7) as res:
return await res.text()
@staticmethod
async def update_up_char():
async def update_up_char(self):
prts_up_char.parent.mkdir(parents=True, exist_ok=True)
data = {'up_char': {'6': {}, '5': {}, '4': {}}, 'title': '', 'time': '', 'pool_img': ''}
try:
text, title = await PrtsAnnouncement.get_announcement_text()
data = {'char': {'up_char': {'6': {}, '5': {}, '4': {}}, 'title': '', 'time': '', 'pool_img': ''}}
text = await self._get_announcement_text()
soup = BeautifulSoup(text, 'lxml')
data['title'] = title
context = soup.find('div', {'id': 'mw-content-text'}).find('div')
data['pool_img'] = str(context.find('div', {'class': 'center'}).find('div').find('a').
find('img').get('srcset')).split(' ')[-2]
# print(context.find_all('p'))
for p in context.find_all('p')[1:]:
if p.text.find('活动时间') != -1:
pr = re.search(r'.*?活动时间：(.*)', p.text)
data['time'] = pr.group(1)
elif p.text.find('★★★★★★') != -1:
chars, probability = _get_up_char(r'.*?★★★★★★：(.*?)（.*?出率的?(.*?)%.*?）.*?', p.text)
slt = '/'
if chars.find('\\') != -1:
slt = '\\'
for char in chars.split(slt):
data['up_char']['6'][char.strip()] = probability.strip()
elif p.text.find('★★★★★') != -1:
chars, probability = _get_up_char(r'.*?★★★★★：(.*?)（.*?出率的?(.*?)%.*?）.*?', p.text)
slt = '/'
if chars.find('\\') != -1:
slt = '\\'
for char in chars.split(slt):
data['up_char']['5'][char.strip()] = probability.strip()
elif p.text.find('★★★★') != -1:
chars, probability = _get_up_char(r'.*?★★★★：(.*?)（.*?出率的?(.*?)%.*?）.*?', p.text)
slt = '/'
if chars.find('\\') != -1:
slt = '\\'
for char in chars.split(slt):
data['up_char']['4'][char.strip()] = probability.strip()
break
pr = re.search(r'.*?★：(.*?)（在(.*?)★.*?以(.*?)倍权值.*?）.*?', p.text)
if pr:
char = pr.group(1)
star = pr.group(2)
weight = pr.group(3)
char = char.replace('[限定]', '').replace('[', '').replace(']', '')
data['up_char'][star][char.strip()] = f'权{weight}'
# data['time'] = '03月09日16:00 - 05月23日03:5
content = soup.find('div', {'class': 'article-content'})
contents = [x for x in content.contents if x.text or str(x).find('img') != -1]
start_index = -1
end_index = -1
for i in range(len(contents)):
if str(contents[i]).startswith('<p>'):
r = re.search('(.*)(寻访|复刻).*?开启', contents[i].text)
if r:
if str(contents[i+3].text).find('★') != -1:
img = contents[i-1].find('img')
if img:
data['char']['pool_img'] = img['src']
start_index = i
for j in range(i, len(contents)):
if str(contents[j]).find('注意') != -1:
end_index = j
break
break
contents = contents[start_index: end_index]
title = contents[0].text
data['char']['title'] = title[title.find('【'): title.find('】') + 1]
data['char']['time'] = str(contents[1].text).split('：', maxsplit=1)[1]
for p in contents[2:]:
p = str(p.text)
r = None
if p.find('★') != -1:
if p.find('权值') == -1:
r = re.search(r'.*?：(.*)（占(.*)★.*?的(.*)%）', p)
else:
r = re.search(r'.*?：(.*)（在(.*)★.*?以(.*)倍权值.*?）', p)
star = r.group(2)
if r:
chars = r.group(1)
if chars.find('/') != -1:
chars = chars.strip().split('/')
elif chars.find('\\') != -1:
chars = chars.strip().split('\\')
else:
chars = chars.split('\n')
chars = [x.replace('[限定]', '').strip() for x in chars]
probability = r.group(3)
probability = probability if int(probability) > 10 else f'权{probability}'
for char in chars:
if char.strip():
data['char']['up_char'][star][char.strip()] = probability
except TimeoutError:
print(f'更新明日方舟UP池信息超时...')
with open(prts_up_char, 'r', encoding='utf8') as f:
data = json.load(f)
logger.warning(f'更新明日方舟UP池信息超时...')
if prts_up_char.exists():
with open(prts_up_char, 'r', encoding='utf8') as f:
data = json.load(f)
except Exception as e:
logger.error(f'更新明日方舟未知错误 e：{e}')
if prts_up_char.exists():
with open(prts_up_char, 'r', encoding='utf8') as f:
data = json.load(f)
return check_write(data, prts_up_char)
class GenshinAnnouncement:
@staticmethod
async def get_announcement_text():
def __init__(self):
self.game_name = '原神'
async def _get_announcement_text(self):
async with aiohttp.ClientSession(headers=headers) as session:
async with session.get(genshin_url, timeout=7) as res:
return await res.text()
@staticmethod
async def update_up_char():
async def update_up_char(self):
genshin_up_char.parent.mkdir(exist_ok=True, parents=True)
data = {
'char': {'up_char': {'5': {}, '4': {}}, 'title': '', 'time': '', 'pool_img': ''},
'arms': {'up_char': {'5': {}, '4': {}}, 'title': '', 'time': '', 'pool_img': ''}
}
text = await GenshinAnnouncement.get_announcement_text()
text = await self._get_announcement_text()
soup = BeautifulSoup(text, 'lxml')
try:
div = soup.find_all('div', {'class': 'row'})[1]
@@ -194,25 +186,30 @@ class GenshinAnnouncement:
for a in trs[3].find('td').find_all('a'):
char_name = a['title']
data[itype]['up_char']['4'][char_name] = "50"
except TimeoutError as e:
except TimeoutError:
logger.warning(f'更新原神UP池信息超时...')
with open(genshin_up_char, 'r', encoding='utf8') as f:
data = json.load(f)
if genshin_up_char.exists():
with open(genshin_up_char, 'r', encoding='utf8') as f:
data = json.load(f)
except Exception as e:
print(f'更新原神UP失败，疑似UP池已结束， e：{e}')
with open(genshin_up_char, 'r', encoding='utf8') as f:
data = json.load(f)
data['char']['title'] = ''
data['arms']['title'] = ''
with open(genshin_up_char, 'w', encoding='utf8') as wf:
json.dump(data, wf, ensure_ascii=False, indent=4)
return data
return check_write(data, genshin_up_char, 'genshin')
logger.error(f'更新原神UP失败，疑似UP池已结束， e：{e}')
if genshin_up_char.exists():
with open(genshin_up_char, 'r', encoding='utf8') as f:
data = json.load(f)
data['char']['title'] = ''
data['arms']['title'] = ''
with open(genshin_up_char, 'w', encoding='utf8') as wf:
json.dump(data, wf, ensure_ascii=False, indent=4)
return data
return check_write(data, genshin_up_char)
class PrettyAnnouncement:
@staticmethod
async def get_announcement_text():
def __init__(self):
self.game_name = '赛马娘'
async def _get_announcement_text(self):
async with aiohttp.ClientSession(headers=headers) as session:
async with session.get(pretty_url, timeout=7) as res:
soup = BeautifulSoup(await res.text(), 'lxml')
@@ -229,14 +226,13 @@ class PrettyAnnouncement:
async with session.get(f'https://wiki.biligame.com/{url}', timeout=7) as res:
return await res.text(), title[:-2]
@staticmethod
async def update_up_char():
async def update_up_char(self):
data = {
'char': {'up_char': {'3': {}, '2': {}, '1': {}}, 'title': '', 'time': '', 'pool_img': ''},
'card': {'up_char': {'3': {}, '2': {}, '1': {}}, 'title': '', 'time': '', 'pool_img': ''}
}
try:
text, title = await PrettyAnnouncement.get_announcement_text()
text, title = await self._get_announcement_text()
soup = BeautifulSoup(text, 'lxml')
context = soup.find('div', {'class': 'toc-sticky'})
if not context:
@@ -299,26 +295,30 @@ class PrettyAnnouncement:
with open(pretty_up_char, 'r', encoding='utf8') as f:
data = json.load(f)
except Exception as e:
logger.error(f'赛马娘up更新失败 {type(e)}：{e}')
return check_write(data, pretty_up_char, 'pretty')
logger.error(f'赛马娘up更新未知错误 {type(e)}：{e}')
if pretty_up_char.exists():
with open(pretty_up_char, 'r', encoding='utf8') as f:
data = json.load(f)
return check_write(data, pretty_up_char)
class GuardianAnnouncement:
@staticmethod
async def get_announcement_text():
def __init__(self):
self.game_name = '坎公骑冠剑'
async def _get_announcement_text(self):
async with aiohttp.ClientSession(headers=headers) as session:
async with session.get(guardian_url, timeout=7) as res:
return await res.text()
@staticmethod
async def update_up_char():
async def update_up_char(self):
data = {
'char': {'up_char': {'3': {}}, 'title': '', 'time': '', 'pool_img': ''},
'arms': {'up_char': {'5': {}}, 'title': '', 'time': '', 'pool_img': ''}
}
try:
text = await GuardianAnnouncement.get_announcement_text()
text = await self._get_announcement_text()
soup = BeautifulSoup(text, 'lxml')
context = soup.select('div.col-sm-3:nth-child(3) > div:nth-child(2) > div:nth-child(1) '
'> div:nth-child(2) > div:nth-child(3) > font:nth-child(1)')[0]
@@ -346,16 +346,17 @@ class GuardianAnnouncement:
end_idx = index
break
index += 1
for x in divs[start_idx+1: end_idx]:
for x in divs[start_idx + 1: end_idx]:
name = x.find('p').find_all('a')[-1].text
data['char']['up_char']['3'][name] = '0'
for x in divs[end_idx+1:]:
for x in divs[end_idx + 1:]:
name = x.find('p').find_all('a')[-1].text
data['arms']['up_char']['5'][name] = '0'
except TimeoutError:
print(f'更新坎公骑冠剑UP池信息超时...')
with open(pretty_up_char, 'r', encoding='utf8') as f:
data = json.load(f)
logger.warning(f'更新坎公骑冠剑UP池信息超时...')
if guardian_up_char.exists():
with open(guardian_up_char, 'r', encoding='utf8') as f:
data = json.load(f)
except Exception as e:
print(f'坎公骑冠剑up更新失败 {type(e)}：{e}')
return check_write(data, guardian_up_char, 'guardian')
logger.error(f'坎公骑冠剑up更新未知错误 {type(e)}：{e}')
return check_write(data, guardian_up_char)
plugins/draw_card/azur_handle.py
+1 -1
View File
@@ -24,7 +24,7 @@ async def azur_draw(count: int, pool_name: str):
cnlist = ['金', '紫', '蓝', '白']
star_list = [0, 0, 0, 0]
char_list, char_dict, max_star_list, star_list, max_star_index_list = \
format_card_information(count, star_list, _get_azur_card, pool_name)
format_card_information(count, star_list, _get_azur_card, pool_name, guaranteed=False)
rst = init_star_rst(star_list, cnlist, max_star_list, max_star_index_list)
if count > 90:
char_list = set_list(char_list)
plugins/draw_card/genshin_handle.py
+10 -36
View File
@@ -1,17 +1,14 @@
import os
from nonebot.adapters.cqhttp import MessageSegment
from nonebot.adapters.cqhttp import MessageSegment, Message
import nonebot
import random
from configs.path_config import IMAGE_PATH
from .update_game_info import update_info
from .util import generate_img, init_star_rst, BaseData, set_list, get_star, UpEvent, download_img
from .util import generate_img, init_star_rst, BaseData, set_list, get_star, init_up_char
from .config import GENSHIN_FIVE_P, GENSHIN_FOUR_P, GENSHIN_G_FIVE_P, GENSHIN_G_FOUR_P, GENSHIN_THREE_P, I72_ADD, \
DRAW_PATH, GENSHIN_FLAG
from dataclasses import dataclass
from .init_card_pool import init_game_pool
from .announcement import GenshinAnnouncement
from services.log import logger
from util.init_result import image
try:
import ujson as json
@@ -20,6 +17,8 @@ except ModuleNotFoundError:
driver: nonebot.Driver = nonebot.get_driver()
announcement = GenshinAnnouncement()
genshin_five = {}
genshin_count = {}
genshin_pl_count = {}
@@ -85,7 +84,7 @@ async def update_genshin_info():
'攻击力（MAX）', '副属性（MAX）', '技能'])
if code == 200:
ALL_ARMS = init_game_pool('genshin_arms', data, GenshinChar)
await _init_up_char()
await _genshin_init_up_char()
async def init_genshin_data():
@@ -100,7 +99,7 @@ async def init_genshin_data():
genshin_ARMS_dict = json.load(f)
ALL_CHAR = init_game_pool('genshin', genshin_dict, GenshinChar)
ALL_ARMS = init_game_pool('genshin_arms', genshin_ARMS_dict, GenshinChar)
await _init_up_char()
await _genshin_init_up_char()
# 抽取卡池
@@ -201,36 +200,11 @@ def reset_count(user_id: int):
# 获取up和概率
async def _init_up_char():
async def _genshin_init_up_char():
global _CURRENT_CHAR_POOL_TITLE, _CURRENT_ARMS_POOL_TITLE, UP_CHAR, UP_ARMS, POOL_IMG
UP_CHAR = []
UP_ARMS = []
up_char_dict = await GenshinAnnouncement.update_up_char()
_CURRENT_CHAR_POOL_TITLE = up_char_dict['char']['title']
_CURRENT_ARMS_POOL_TITLE = up_char_dict['arms']['title']
if _CURRENT_CHAR_POOL_TITLE and _CURRENT_ARMS_POOL_TITLE:
await download_img(up_char_dict['char']['pool_img'], 'genshin', 'up_char_pool_img')
await download_img(up_char_dict['arms']['pool_img'], 'genshin', 'up_arms_pool_img')
POOL_IMG = image('up_char_pool_img.png', '/draw_card/genshin/') + \
image('up_arms_pool_img.png', '/draw_card/genshin/')
else:
if os.path.exists(IMAGE_PATH + '/draw_card/genshin/up_char_pool_img.png'):
os.remove(IMAGE_PATH + '/draw_card/genshin/up_char_pool_img.png')
if os.path.exists(IMAGE_PATH + '/draw_card/genshin/up_arms_pool_img.png'):
os.remove(IMAGE_PATH + '/draw_card/genshin/up_arms_pool_img.png')
logger.info(f'成功获取原神当前up信息...当前up池: {_CURRENT_CHAR_POOL_TITLE} & {_CURRENT_ARMS_POOL_TITLE}')
for key in up_char_dict.keys():
for star in up_char_dict[key]['up_char'].keys():
up_char_lst = []
for char in up_char_dict[key]['up_char'][star].keys():
up_char_lst.append(char)
if key == 'char':
UP_CHAR.append(UpEvent(star=int(star), operators=up_char_lst, zoom=0))
else:
UP_ARMS.append(UpEvent(star=int(star), operators=up_char_lst, zoom=0))
_CURRENT_CHAR_POOL_TITLE, _CURRENT_ARMS_POOL_TITLE, POOL_IMG, UP_CHAR, UP_ARMS = await init_up_char(announcement)
async def reload_genshin_pool():
await _init_up_char()
return f'当前UP池子：{_CURRENT_CHAR_POOL_TITLE} & {_CURRENT_ARMS_POOL_TITLE} {POOL_IMG}'
await _genshin_init_up_char()
return Message(f'当前UP池子：{_CURRENT_CHAR_POOL_TITLE} & {_CURRENT_ARMS_POOL_TITLE} {POOL_IMG}')
plugins/draw_card/guardian_handle.py
+20 -29
View File
@@ -1,12 +1,11 @@
import os
import nonebot
from nonebot.adapters.cqhttp import MessageSegment
from nonebot.adapters.cqhttp import MessageSegment, Message
from .update_game_info import update_info
from services.log import logger
from .announcement import GuardianAnnouncement
from .util import init_star_rst, generate_img, max_card, BaseData,\
set_list, get_star, format_card_information, UpEvent
set_list, get_star, format_card_information, init_up_char
import random
from .config import DRAW_PATH, GUARDIAN_ONE_CHAR_P, GUARDIAN_TWO_CHAR_P, GUARDIAN_THREE_CHAR_P, \
GUARDIAN_THREE_CHAR_UP_P, GUARDIAN_TWO_ARMS_P, GUARDIAN_FIVE_ARMS_P, GUARDIAN_THREE_CHAR_OTHER_P, \
@@ -21,6 +20,8 @@ except ModuleNotFoundError:
driver: nonebot.Driver = nonebot.get_driver()
announcement = GuardianAnnouncement()
ALL_CHAR = []
ALL_ARMS = []
@@ -92,6 +93,7 @@ async def update_guardian_info():
if code_1 == 200 and code_2 == 200:
data.update(tmp)
ALL_ARMS = init_game_pool('guardian_arms', data, GuardianArms)
await _guardian_init_up_char()
async def init_guardian_data():
@@ -106,21 +108,27 @@ async def init_guardian_data():
guardian_arms_dict = json.load(f)
ALL_CHAR = init_game_pool('guardian', guardian_char_dict, GuardianChar)
ALL_ARMS = init_game_pool('guardian_arms', guardian_arms_dict, GuardianArms)
await _init_up_char()
await _guardian_init_up_char()
# 抽取卡池
def _get_guardian_card(pool_name: str):
def _get_guardian_card(pool_name: str = '', mode: int = 1):
global ALL_ARMS, ALL_CHAR, UP_ARMS, UP_CHAR, _CURRENT_ARMS_POOL_TITLE, _CURRENT_CHAR_POOL_TITLE
if pool_name == 'char':
star = get_star([3, 2, 1], [GUARDIAN_THREE_CHAR_P, GUARDIAN_TWO_CHAR_P, GUARDIAN_ONE_CHAR_P])
if mode == 1:
star = get_star([3, 2, 1], [GUARDIAN_THREE_CHAR_P, GUARDIAN_TWO_CHAR_P, GUARDIAN_ONE_CHAR_P])
else:
star = get_star([3, 2], [GUARDIAN_THREE_CHAR_P, GUARDIAN_TWO_CHAR_P])
up_lst = UP_CHAR
flag = _CURRENT_CHAR_POOL_TITLE
_max_star = 3
all_data = ALL_CHAR
else:
star = get_star([5, 4, 3, 2], [GUARDIAN_FIVE_ARMS_P, GUARDIAN_FOUR_ARMS_P,
GUARDIAN_THREE_ARMS_P, GUARDIAN_TWO_ARMS_P])
if mode == 1:
star = get_star([5, 4, 3, 2], [GUARDIAN_FIVE_ARMS_P, GUARDIAN_FOUR_ARMS_P,
GUARDIAN_THREE_ARMS_P, GUARDIAN_TWO_ARMS_P])
else:
star = get_star([5, 4], [GUARDIAN_FIVE_ARMS_P, GUARDIAN_FOUR_ARMS_P])
up_lst = UP_ARMS
flag = _CURRENT_ARMS_POOL_TITLE
_max_star = 5
@@ -144,28 +152,11 @@ def _get_guardian_card(pool_name: str):
# 获取up和概率
async def _init_up_char():
async def _guardian_init_up_char():
global _CURRENT_CHAR_POOL_TITLE, _CURRENT_ARMS_POOL_TITLE, UP_CHAR, UP_ARMS, POOL_IMG
UP_CHAR = []
UP_ARMS = []
up_char_dict = await GuardianAnnouncement.update_up_char()
_CURRENT_CHAR_POOL_TITLE = up_char_dict['char']['title']
_CURRENT_ARMS_POOL_TITLE = up_char_dict['arms']['title']
if _CURRENT_CHAR_POOL_TITLE and _CURRENT_ARMS_POOL_TITLE:
POOL_IMG = MessageSegment.image(up_char_dict['char']['pool_img']) + \
MessageSegment.image(up_char_dict['arms']['pool_img'])
logger.info(f'成功获取坎公骑冠剑当前up信息...当前up池: {_CURRENT_CHAR_POOL_TITLE} & {_CURRENT_ARMS_POOL_TITLE}')
for key in up_char_dict.keys():
for star in up_char_dict[key]['up_char'].keys():
up_char_lst = []
for char in up_char_dict[key]['up_char'][star].keys():
up_char_lst.append(char)
if key == 'char':
UP_CHAR.append(UpEvent(star=int(star), operators=up_char_lst, zoom=0))
else:
UP_ARMS.append(UpEvent(star=int(star), operators=up_char_lst, zoom=0))
_CURRENT_CHAR_POOL_TITLE, _CURRENT_ARMS_POOL_TITLE, POOL_IMG, UP_CHAR, UP_ARMS = await init_up_char(announcement)
async def reload_guardian_pool():
await _init_up_char()
return f'当前UP池子：{_CURRENT_CHAR_POOL_TITLE} & {_CURRENT_ARMS_POOL_TITLE} {POOL_IMG}'
await _guardian_init_up_char()
return Message(f'当前UP池子：{_CURRENT_CHAR_POOL_TITLE} & {_CURRENT_ARMS_POOL_TITLE}')
plugins/draw_card/pretty_handle.py
+19 -41
View File
@@ -1,16 +1,14 @@
import os
import nonebot
from .update_game_info import update_info
from .announcement import PrettyAnnouncement
from .util import download_img, init_star_rst, generate_img, max_card, BaseData, \
set_list, get_star, format_card_information, UpEvent
from .util import init_star_rst, generate_img, max_card, BaseData, \
set_list, get_star, format_card_information, init_up_char
import random
from .config import PRETTY_THREE_P, PRETTY_TWO_P, DRAW_PATH, PRETTY_ONE_P, PRETTY_FLAG
from dataclasses import dataclass
from .init_card_pool import init_game_pool
from util.init_result import image
from configs.path_config import IMAGE_PATH
from services.log import logger
from nonebot.adapters.cqhttp import MessageSegment, Message
try:
import ujson as json
@@ -19,6 +17,8 @@ except ModuleNotFoundError:
driver: nonebot.Driver = nonebot.get_driver()
announcement = PrettyAnnouncement()
ALL_CHAR = []
ALL_CARD = []
@@ -76,7 +76,7 @@ async def pretty_draw(count: int, pool_name):
rst = init_star_rst(star_list, cnlist, three_list, three_olist, up_list)
if count > 90:
obj_list = set_list(obj_list)
return pool_info + image(b64=await generate_img(obj_list, 'pretty', star_list)) \
return pool_info + MessageSegment.image("base64://" + await generate_img(obj_list, 'pretty', star_list)) \
+ '\n' + rst[:-1] + '\n' + max_card(obj_dict)
@@ -90,7 +90,7 @@ async def update_pretty_info():
data, code = await update_info(url, 'pretty_card')
if code == 200:
ALL_CARD = init_game_pool('pretty_card', data, PrettyChar)
await _init_up_char()
await _pretty_init_up_char()
async def init_pretty_data():
@@ -102,13 +102,16 @@ async def init_pretty_data():
pretty_card_dict = json.load(f)
ALL_CHAR = init_game_pool('pretty', pretty_char_dict, PrettyChar)
ALL_CARD = init_game_pool('pretty_card', pretty_card_dict, PrettyChar)
await _init_up_char()
await _pretty_init_up_char()
# 抽取卡池
def _get_pretty_card(pool_name: str):
def _get_pretty_card(pool_name: str, mode: int = 1):
global ALL_CHAR, ALL_CARD, _CURRENT_CHAR_POOL_TITLE, _CURRENT_CARD_POOL_TITLE
star = get_star([3, 2, 1], [PRETTY_THREE_P, PRETTY_TWO_P, PRETTY_ONE_P])
if mode == 1:
star = get_star([3, 2, 1], [PRETTY_THREE_P, PRETTY_TWO_P, PRETTY_ONE_P])
else:
star = get_star([3, 2], [PRETTY_THREE_P, PRETTY_TWO_P])
if pool_name == 'card':
title = _CURRENT_CARD_POOL_TITLE
up_data = UP_CARD
@@ -126,6 +129,7 @@ def _get_pretty_card(pool_name: str):
acquire_operator = random.choice(all_up_star)
if pool_name == 'char':
acquire_operator = acquire_operator.split(']')[1]
print(acquire_operator)
acquire_operator = [x for x in data if x.name == acquire_operator][0]
else:
acquire_operator = random.choice([x for x in data if x.star == star and not x.limited])
@@ -135,39 +139,13 @@ def _get_pretty_card(pool_name: str):
# 获取up和概率
async def _init_up_char():
async def _pretty_init_up_char():
global _CURRENT_CHAR_POOL_TITLE, _CURRENT_CARD_POOL_TITLE, UP_CHAR, UP_CARD, POOL_IMG
UP_CHAR = []
UP_CARD = []
up_char_dict = await PrettyAnnouncement.update_up_char()
_CURRENT_CHAR_POOL_TITLE = up_char_dict['char']['title']
_CURRENT_CARD_POOL_TITLE = up_char_dict['card']['title']
if _CURRENT_CHAR_POOL_TITLE and _CURRENT_CARD_POOL_TITLE:
await download_img(up_char_dict['char']['pool_img'], 'pretty', 'up_char_pool_img')
await download_img(up_char_dict['card']['pool_img'], 'pretty', 'up_card_pool_img')
POOL_IMG = image('up_char_pool_img.png', '/draw_card/pretty/') + \
image('up_card_pool_img.png', '/draw_card/pretty/')
else:
if os.path.exists(IMAGE_PATH + '/draw_card/genshin/up_char_pool_img.png'):
os.remove(IMAGE_PATH + '/draw_card/pretty/up_char_pool_img.png')
if os.path.exists(IMAGE_PATH + '/draw_card/genshin/up_arms_pool_img.png'):
os.remove(IMAGE_PATH + '/draw_card/pretty/up_card_pool_img.png')
logger.info(f'成功获取赛马娘当前up信息...当前up池: {_CURRENT_CHAR_POOL_TITLE} & {_CURRENT_CARD_POOL_TITLE}')
for key in up_char_dict.keys():
for star in up_char_dict[key]['up_char'].keys():
up_lst = []
for char in up_char_dict[key]['up_char'][star].keys():
up_lst.append(char)
if key == 'char':
if up_lst:
UP_CHAR.append(UpEvent(star=int(star), operators=up_lst, zoom=0))
else:
if up_lst:
UP_CARD.append(UpEvent(star=int(star), operators=up_lst, zoom=0))
_CURRENT_CHAR_POOL_TITLE, _CURRENT_CARD_POOL_TITLE, POOL_IMG, UP_CHAR, UP_CARD = await init_up_char(announcement)
async def reload_pretty_pool():
await _init_up_char()
return f'当前UP池子：{_CURRENT_CHAR_POOL_TITLE} & {_CURRENT_CARD_POOL_TITLE} {POOL_IMG}'
await _pretty_init_up_char()
return Message(f'当前UP池子：{_CURRENT_CHAR_POOL_TITLE} & {_CURRENT_CARD_POOL_TITLE} {POOL_IMG}')
plugins/draw_card/prts_handle.py
+17 -8
View File
@@ -1,5 +1,5 @@
from nonebot.adapters.cqhttp import MessageSegment
from nonebot.adapters.cqhttp import MessageSegment, Message
import nonebot
import random
from .config import PRTS_FIVE_P, PRTS_FOUR_P, PRTS_SIX_P, PRTS_THREE_P, DRAW_PATH, PRTS_FLAG
@@ -17,6 +17,8 @@ except ModuleNotFoundError:
driver: nonebot.Driver = nonebot.get_driver()
announcement = PrtsAnnouncement()
up_char_file = Path() / "data" / "draw_card" / "draw_card_up" / "prts_up_char.json"
prts_dict = {}
@@ -38,15 +40,22 @@ async def prts_draw(count: int = 300):
star_list = [0, 0, 0, 0]
operator_list, operator_dict, six_list, star_list, six_index_list = format_card_information(count, star_list)
up_list = []
tmp = ''
if _CURRENT_POOL_TITLE:
for x in UP_OPERATOR:
for operator in x.operators:
up_list.append(operator)
if x.star == 6:
tmp += f'六星UP：{" ".join(x.operators)} \n'
elif x.star == 5:
tmp += f'五星UP：{" ".join(x.operators)} \n'
elif x.star == 4:
tmp += f'四星UP：{" ".join(x.operators)}'
rst = init_star_rst(star_list, cnlist, six_list, six_index_list, up_list)
if count > 90:
operator_list = set_list(operator_list)
pool_info = "当前up池: " if _CURRENT_POOL_TITLE else ""
return pool_info + _CURRENT_POOL_TITLE + MessageSegment.image(
pool_info = f"当前up池: {_CURRENT_POOL_TITLE}\n{tmp}" if _CURRENT_POOL_TITLE else ""
return pool_info + MessageSegment.image(
"base64://" + await generate_img(operator_list, 'prts', star_list)) \
+ '\n' + rst[:-1] + '\n' + max_card(operator_dict)
@@ -137,11 +146,11 @@ def format_card_information(count: int, star_list: list):
async def _init_up_char():
global _CURRENT_POOL_TITLE, POOL_IMG, UP_OPERATOR
UP_OPERATOR = []
up_char_dict = await PrtsAnnouncement.update_up_char()
_CURRENT_POOL_TITLE = up_char_dict['title']
up_char_dict = await announcement.update_up_char()
_CURRENT_POOL_TITLE = up_char_dict['char']['title']
if _CURRENT_POOL_TITLE:
POOL_IMG = MessageSegment.image(up_char_dict['pool_img'])
up_char_dict = up_char_dict['up_char']
POOL_IMG = MessageSegment.image(up_char_dict['char']['pool_img'])
up_char_dict = up_char_dict['char']['up_char']
logger.info(f'成功获取明日方舟当前up信息...当前up池: {_CURRENT_POOL_TITLE}')
average_dict = {'6': {}, '5': {}, '4': {}}
for star in up_char_dict.keys():
@@ -161,4 +170,4 @@ async def _init_up_char():
async def reload_prts_pool():
await _init_up_char()
return f'当前UP池：{_CURRENT_POOL_TITLE} {POOL_IMG}'
return Message(f'当前UP池：{_CURRENT_POOL_TITLE} {POOL_IMG}')
plugins/draw_card/util.py
+59 -3
View File
@@ -3,6 +3,7 @@ import aiohttp
import aiofiles
from asyncio.exceptions import TimeoutError
from aiohttp.client_exceptions import InvalidURL
from nonebot.adapters.cqhttp import MessageSegment
from typing import List, Union, Set
from pathlib import Path
from .config import path_dict
@@ -218,21 +219,76 @@ def get_star(star_lst: List[int], probability_lst: List[float]) -> int:
return star_lst[i]
# 获取up和概率
async def init_up_char(announcement):
UP_CHAR = []
UP_ARMS = []
tmp = ''
up_char_dict = await announcement.update_up_char()
for x in list(up_char_dict.keys()):
tmp += up_char_dict[x]['title'] + '[\n]'
tmp = tmp.split('[\n]')
_CURRENT_CHAR_POOL_TITLE = tmp[0]
if len(up_char_dict) > 1:
_CURRENT_ARMS_POOL_TITLE = tmp[1]
else:
_CURRENT_ARMS_POOL_TITLE = ''
POOL_IMG = ''
x = [x for x in list(up_char_dict.keys())]
if _CURRENT_CHAR_POOL_TITLE:
POOL_IMG += MessageSegment.image(up_char_dict[x[0]]['pool_img'])
try:
if _CURRENT_ARMS_POOL_TITLE:
POOL_IMG += MessageSegment.image(up_char_dict[x[1]]['pool_img'])
except (IndexError, KeyError):
pass
logger.info(f'成功获取{announcement.game_name}当前up信息...当前up池: {_CURRENT_CHAR_POOL_TITLE} & {_CURRENT_ARMS_POOL_TITLE}')
for key in up_char_dict.keys():
for star in up_char_dict[key]['up_char'].keys():
up_char_lst = []
for char in up_char_dict[key]['up_char'][star].keys():
up_char_lst.append(char)
if up_char_lst:
if key == 'char':
UP_CHAR.append(UpEvent(star=int(star), operators=up_char_lst, zoom=0))
else:
UP_ARMS.append(UpEvent(star=int(star), operators=up_char_lst, zoom=0))
return _CURRENT_CHAR_POOL_TITLE, _CURRENT_ARMS_POOL_TITLE, POOL_IMG, UP_CHAR, UP_ARMS
# 整理数据
def format_card_information(count: int, star_list: List[int], func, pool_name: str = ''):
def format_card_information(count: int, star_list: List[int], func, pool_name: str = '', guaranteed: bool = True):
max_star_lst = [] # 获取的最高星级角色列表
max_index_lst = [] # 获取最高星级角色的次数
obj_list = [] # 获取所有角色
obj_dict = {} # 获取角色次数字典
_count = -1
if guaranteed:
_count = 0
for i in range(count):
if guaranteed:
_count += 1
if pool_name:
obj, code = func(pool_name)
if _count == 10:
obj, code = func(pool_name, 2)
_count = 0
else:
obj, code = func(pool_name)
else:
obj, code = func()
if _count == 10:
obj, code = func(mode=2)
_count = 0
else:
obj, code = func()
star_list[code] += 1
if code == 0:
max_star_lst.append(obj.name)
max_index_lst.append(i)
if guaranteed:
_count = 0
if code == 1:
if guaranteed:
_count = 0
try:
obj_dict[obj.name] += 1
except KeyError:
plugins/help/config.py
+5 -2
View File
@@ -14,7 +14,8 @@ utility_help = {
'epic': 'epic速速白嫖 --> 指令:epic',
'pixiv_r': 'P站排行榜直接冲 --> 指令:p站排行(可含参数)',
'pixiv_s': 'P站的图随便搜搜 --> 指令:搜图(可含参数)',
'translate': '出国旅游助手(狗头) --> 指令:英翻/翻英/日翻/翻日/韩翻/翻韩'
'translate': '出国旅游助手(狗头) --> 指令:英翻/翻英/日翻/翻日/韩翻/翻韩',
'cover': '快捷的b站封面获取方式 --> 指令:b封面 [连接/av/bv/cv/直播id]'
}
# 娱乐
entertainment_help = {
@@ -45,7 +46,9 @@ entertainment_help = {
'qiu_qiu_translation': '这家伙到底在说什么？ --> 指令:丘丘翻译/丘丘一下/丘丘语翻译',
'query_resource_points': '地图资源速速查看 --> 指令:原神资源查询xx/原神资源列表/哪里有xx/xx在哪(xx=资源名称)',
'russian': '紧张刺激的俄罗斯轮盘 --> 指令:俄罗斯轮盘帮助',
'gold_redbag': '运气项目又来啦！ --> 指令:塞红包/ 抢红包指令: 开/抢/戳一戳真寻/ 退回:退回未抢完的红包'
'gold_redbag': '运气项目又来啦！ --> 指令:塞红包/ 抢红包指令: 开/抢/戳一戳真寻/ 退回:退回未抢完的红包',
'poetry': '突然文艺的真寻是否搞错了什么 --> 指令:念诗/来首诗/念首诗',
'comments_163': '生了个人，我很抱歉 --> 指令:到点了/12点了/网易云热评/网易云评论',
}
# 其他
other_help = [
plugins/hook.py
+40
View File
@@ -108,5 +108,45 @@ async def _(matcher: Matcher, bot: Bot, event: MessageEvent, state: T_State):
raise IgnoredException('有命令就别说话了')
# _ulmt = UserExistLimiter()
#
#
# # 是否有命令正在处理
# @run_preprocessor
# async def _(matcher: Matcher, bot: Bot, event: Event, state: T_State):
# if matcher.module == 'admin_bot_manage':
# return
# if event.user_id == bot.self_id:
# raise IgnoredException('自己和自己聊天？')
# print(state["_prefix"]["raw_command"])
# print(matcher.type)
# if (event.is_tome() and (state["_prefix"]["raw_command"] and matcher.type == 'message')
# and matcher.module == 'ai') or not static_group_dict[event.group_id]['总开关']:
# raise IgnoredException('Ai给爷爬？')
# # if matcher.module in ['ai']:
# # raise IgnoredException('Ai给爷爬？')
# if matcher.type == 'message' and matcher.priority not in [1, 8, 9]:
# if _ulmt.check(event.user_id):
# if event.message_type == 'group':
# await bot.send_group_msg(group_id=event.group_id, message=at(event.user_id) + '您有命令正在处理，等一哈！')
# else:
# await bot.send_private_msg(user_id=event.user_id, message='您有命令正在处理，等一哈！')
# raise IgnoredException('有命令正在处理')
# _ulmt.set_True(event.user_id)
#
#
# # 结束正在处理的命令
# @run_postprocessor
# async def _(matcher: Matcher, bot: Bot, event: Event, state: T_State):
# _ulmt.set_False(event.user_id)
#
#
# # 结束正在处理的命令
# @event_preprocessor
# async def _(bot: Bot, event: Event, state: T_State):
# try:
# _ulmt.set_False(event.user_id)
# except AttributeError:
# pass
plugins/reimu/__init__.py
+1 -1
View File
@@ -64,7 +64,7 @@ async def _(bot: Bot, event: Event, state: T_State):
page = state['page']
print(keyword, page)
await UserCount.add_count(event.user_id, 'reimu')
await reimu.send('已经帮你关好车门了', at_sender=True)
await reimu.send('已经帮你关好车门了，请等待发车（不加真寻好友的话是欣赏不到旅途的风景的）', at_sender=True)
reimu_report = await from_reimu_get_info(keyword, page)
if reimu_report:
await reimu.send(reimu_report)
plugins/send_setu/__init__.py
+32 -9
View File
@@ -79,7 +79,7 @@ async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
f"USER {event.user_id} GROUP {event.group_id} 发送色图 {index}.jpg 成功")
else:
msg = img_id
if list(bot.config.nickname)[0].find(msg) != -1:
if msg.find('真寻') != -1:
_ulmt.set_False(event.user_id)
await setu.finish('咳咳咳，虽然我很可爱，但是我木有自己的色图~~~有的话记得发我一份呀')
keyword, r18, num = await check_r18_and_keyword(msg, event.user_id)
@@ -100,13 +100,19 @@ async def _(bot: Bot, event: GroupMessageEvent, state: T_State):
for i in range(num):
try:
setu_img, index = await search_online_setu(urls[i])
await setu.send(text_list[i] + setu_img)
if index == -1:
setu_img, index = get_setu('', setu_data, keyword)
await setu.send(setu_img)
logger.error(f'色图下载失败 本地发送 {index}.jpg')
else:
await setu.send(text_list[i] + setu_img)
logger.info(
f"USER {event.user_id} GROUP {event.group_id}"
f" 发送在线色图 {keyword}.jpg 成功")
except Exception as e:
logger.error(f'色图发送错误 e：{e}')
await setu.send('图片下载惜败！', at_sender=True)
setu_img, index = get_setu('', setu_data, keyword)
await setu.send(setu_img)
logger.error(f'色图发送错误 本地发送 {index}.jpg e：{e}')
else:
await setu.send(urls)
_ulmt.set_False(event.user_id)
@@ -155,13 +161,21 @@ async def _(bot: Bot, event: PrivateMessageEvent, state: T_State):
for i in range(num):
try:
setu_img, index = await search_online_setu(urls[i])
await setu.send(text_list[i] + setu_img)
if index == -1 and img_id != 'r':
setu_img, index = get_setu('', setu_data, keyword)
await setu.send(setu_img)
logger.error(f'色图下载失败 本地发送 {index}.jpg')
else:
await setu.send(text_list[i] + setu_img)
logger.info(
f"USER {event.user_id} GROUP private"
f" 发送{'r18' if img_id == 'r' else ''}色图 {index}.jpg 成功")
except Exception as e:
logger.error(f'色图发送错误 e：{e}')
await setu.send('图片下载惜败！', at_sender=True)
if img_id != 'r':
img_setu, index = get_setu('', setu_data, keyword)
logger.error(f'色图发送错误 本地发送 {index}.jpg e：{e}')
else:
await setu.send('图片下载惜败！', at_sender=True)
count += 1
if count > 6:
await setu.send('检测到下载惜败的图片过多，这次就不算数了，果咩..', at_sender=True)
@@ -255,16 +269,25 @@ async def _(bot: Bot, event: MessageEvent, state: T_State):
else:
_ulmt.set_False(event.user_id)
return
if list(bot.config.nickname)[0].find(keyword) != -1:
if keyword.find('真寻') != -1:
await setu.finish('咳咳咳，虽然我很可爱，但是我木有自己的色图~~~有的话记得发我一份呀')
urls, text_list, code = await get_setu_urls(keyword, num)
if code == 200:
for i in range(len(urls)):
try:
setu_img, index = await search_online_setu(urls[i])
await setu_reg.send(text_list[i] + '\n' + setu_img)
if index == -1:
setu_img, index = get_setu('', setu_data, keyword)
await setu_reg.send(setu_img)
logger.error(f'色图下载失败 本地发送 {index}.jpg')
else:
await setu_reg.send(text_list[i] + '\n' + setu_img)
except ActionFailed as e:
await setu_reg.send('这图太色了，会教坏小孩子的，不给看..')
except Exception as e:
setu_img, index = get_setu('', setu_data, keyword)
await setu_reg.send(setu_img)
logger.error(f'色图发送错误 e：{e}')
else:
logger.info(
f"USER {event.user_id} GROUP {event.group_id if event.message_type != 'private' else 'private'}"
plugins/send_setu/data_source.py
+8 -3
View File
@@ -46,7 +46,7 @@ async def get_setu_urls(keyword: str, num: int = 1, r18: int = 0):
print(f'get_setu_url: count --> {count}')
async with aiohttp.ClientSession() as session:
try:
async with session.get(url, proxy=get_local_proxy(), timeout=5, params=params) as response:
async with session.get(url, proxy=get_local_proxy(), timeout=2, params=params) as response:
if response.status == 429:
return '调用达到上限，明日赶早呀~', '', 429
if response.status == 404:
@@ -96,7 +96,7 @@ async def search_online_setu(url: str):
for i in range(3):
print(f'search_online_setu --> {i}')
try:
async with session.get(url, proxy=get_local_proxy(), timeout=7) as res:
async with session.get(url, proxy=get_local_proxy(), timeout=2) as res:
if res.status == 200:
index = str(random.randint(1, 100000))
async with aiofiles.open(IMAGE_PATH + 'temp/' + index + "_temp_setu.jpg", 'wb') as f:
@@ -115,12 +115,17 @@ async def search_online_setu(url: str):
return '\n图片被小怪兽恰掉啦..!QAQ', -1
def get_setu(index: str, setu_data: dict):
def get_setu(index: str, setu_data: dict, tag: str = None):
length = len(os.listdir(IMAGE_PATH + path))
if length == 0:
return None, None
if not index:
index = random.randint(0, length - 1)
if tag:
all_tag_setu = [x for x in setu_data.keys() if tag in setu_data[x]['tags']]
print(all_tag_setu)
if all_tag_setu:
index = random.choice(all_tag_setu)
if is_number(index):
if int(index) > length or int(index) < 0:
return f"超过当前上下限！({length - 1})", 999
plugins/songpicker2/__init__.py
+9 -44
View File
@@ -1,4 +1,4 @@
from .data_source import dataGet, dataProcess
from .music_163 import dataGet, dataProcess, get_music_id
from nonebot.adapters import Bot, Event
from nonebot.typing import T_State
from nonebot import on_command
@@ -15,9 +15,6 @@ songpicker = on_command("点歌", priority=5, block=True)
@songpicker.handle()
async def handle_first_receive(bot: Bot, event: Event, state: T_State):
args = str(event.get_message()).strip()
# if args.isdigit():
# if "songName" in state:
# state["songNum"] = int(args)
if args:
state["songName"] = args
@@ -27,34 +24,20 @@ async def handle_songName(bot: Bot, event: Event, state: T_State):
songName = state["songName"]
songIdList = await dataget.songIds(songName=songName)
if not songIdList:
await songpicker.reject("没有找到这首歌，请发送其它歌名！")
songInfoList = list()
for songId in songIdList:
songInfoDict = await dataget.songInfo(songId)
if songInfoDict == '网易云网络繁忙！':
await songpicker.finish('网易云网络繁忙！')
songInfoList.append(songInfoDict)
# songInfoMessage = await dataProcess.mergeSongInfo(songInfoList)
# await songpicker.send(songInfoMessage)
await songpicker.finish("没有找到这首歌！", at_sender=True)
for _ in range(3):
songInfoDict = await dataget.songInfo(songIdList[0])
if songInfoDict != '网易云网络繁忙！':
break
else:
await songpicker.finish("网易云繁忙...")
state["songIdList"] = songIdList
@songpicker.got("songName")
async def handle_songNum(bot: Bot, event: Event, state: T_State):
songIdList = state["songIdList"]
# songNum = state["songNum"]
songNum = 0
# 处理重选
# if not songNum.isdigit():
# await songpicker.finish()
# else:
# songNum = int(songNum)
#
# if songNum >= len(songIdList) or songNum < 0:
# await songpicker.reject("数字序号错误")
selectedSongId = songIdList[int(songNum)]
selectedSongId = songIdList[0]
songContent = [
{
"type": "music",
@@ -66,21 +49,3 @@ async def handle_songNum(bot: Bot, event: Event, state: T_State):
]
await songpicker.send(songContent)
# songCommentsDict = await dataget.songComments(songId=selectedSongId)
# songCommentsMessage = await dataProcess.mergeSongComments(songCommentsDict)
# commentContent = [
# {
# "type": "text",
# "data": {
# "text": "下面为您播送热评：\n"
# }
# },
# {
# "type": "text",
# "data": {
# "text": songCommentsMessage
# }
# }
# ]
#
# await songpicker.send(commentContent)
plugins/songpicker2/music_163.py
+144
View File
@@ -0,0 +1,144 @@
import aiohttp
import json
from util.utils import get_local_proxy
from configs.config import ALAPI_TOKEN
from asyncio.exceptions import TimeoutError
class dataApi():
'''
从网易云音乐接口直接获取数据（实验性）
'''
headers = {"referer": "http://music.163.com"}
cookies = {"appver": "2.0.2"}
async def search(self, songName: str):
'''
搜索接口，用于由歌曲名查找id
'''
async with aiohttp.ClientSession(headers=self.headers, cookies=self.cookies) as session:
async with session.post(f"http://music.163.com/api/search/get/", data={"s": songName, "limit": 5, "type": 1, "offset": 0},) as r:
if r.status != 200:
return None
r = await r.text()
return json.loads(r)
async def getSongInfo(self, songId: int):
'''
获取歌曲信息
'''
async with aiohttp.ClientSession(headers=self.headers, cookies=self.cookies) as session:
async with session.post(f"http://music.163.com/api/song/detail/?id={songId}&ids=%5B{songId}%5D",) as r:
if r.status != 200:
return None
r = await r.text()
return json.loads(r)
class dataGet(dataApi):
'''
从dataApi获取数据，并做简单处理
'''
api = dataApi()
async def songIds(self, songName: str, amount=5) -> list:
'''
根据用户输入的songName 获取候选songId列表 [默认songId数量：5]
'''
songIds = list()
r = await self.api.search(songName=songName)
if r is None:
raise WrongDataError
idRange = amount if amount < len(
r["result"]["songs"]) else len(r["result"]["songs"])
for i in range(idRange):
songIds.append(r["result"]["songs"][i]["id"])
return songIds
async def songInfo(self, songId: int):
'''
根据传递的songId，获取歌曲名、歌手、专辑等信息，作为dict返回
'''
songInfo = dict()
r = await self.api.getSongInfo(songId)
if r is None:
raise WrongDataError
if r['code'] == -460:
return '网易云网络繁忙！'
songInfo["songName"] = r["songs"][0]["name"]
songArtists = list()
for ars in r["songs"][0]["artists"]:
songArtists.append(ars["name"])
songArtistsStr = "、".join(songArtists)
songInfo["songArtists"] = songArtistsStr
songInfo["songAlbum"] = r["songs"][0]["album"]["name"]
return songInfo
class dataProcess():
'''
将获取的数据处理为用户能看懂的形式
'''
@staticmethod
async def mergeSongInfo(songInfos: list) -> str:
'''
将歌曲信息list处理为字符串，供用户点歌
传递进的歌曲信息list含有多个歌曲信息dict
'''
songInfoMessage = "请输入欲点播歌曲的序号：\n"
numId = 0
for songInfo in songInfos:
songInfoMessage += f"{numId}："
songInfoMessage += songInfo["songName"]
songInfoMessage += "-"
songInfoMessage += songInfo["songArtists"]
songInfoMessage += " 专辑："
songInfoMessage += songInfo["songAlbum"]
songInfoMessage += "\n"
numId += 1
return songInfoMessage
@staticmethod
async def mergeSongComments(songComments: dict) -> str:
songCommentsMessage = '\n'.join(
['%s： %s' % (key, value) for (key, value) in songComments.items()])
return songCommentsMessage
class Error(Exception):
'''
谁知道网易的接口会出什么幺蛾子
'''
pass
class WrongDataError(Error):
def __init__(self, expression, message):
self.expression = expression
self.message = message
self.message += "\n未从网易接口获取到有效的数据！"
async def get_music_id(keyword: str):
url = f'https://v2.alapi.cn/api/music/search'
params = {
'token': ALAPI_TOKEN,
'keyword': keyword
}
async with aiohttp.ClientSession() as session:
try:
async with session.get(url, proxy=get_local_proxy(), timeout=2, params=params) as response:
data = await response.json()
if data['code'] == 200:
data = data['data']['songs']
return data['id'], 200
else:
return f'访问失败...code：{data["code"]}', 999
except TimeoutError:
return '超时了...', 999
plugins/test.py
+15 -17
View File
@@ -20,21 +20,24 @@ from models.group_member_info import GroupInfoUser
# erm = on_command('异世相遇，尽享美味', aliases={'异世相遇 尽享美味', '异世相遇,尽享美味'}, priority=5, block=True)
matcher = on_keyword({"test"})
# matcher = on_keyword({"test"})
#
#
# @matcher.handle()
# async def _(bot: Bot, event: MessageEvent, state: T_State):
# await matcher.send(json(
# {"app": "com.tencent.gxhServiceIntelligentTip", "desc": "", "view": "gxhServiceIntelligentTip",
# "ver": "", "prompt": "[QQ红包]", "appID": "", "sourceName": "", "actionData": "", "actionData_A": "",
# "sourceUrl": "", "meta": {"gxhServiceIntelligentTip": {
# "bgImg": "http:\/\/ptlogin2.qq.com\/ho_cross_domain?tourl=https:\/\/gxh.vip.qq.com\/\/qqshow"
# "\/admindata\/comdata\/vipActTpl_mobile_zbltyxn\/dddb247a4a9c6d34757c160f9e0b6669.gif",
# "appid": "gxhServiceIntelligentTip", "reportParams": 'null', "action": ""}},
# "config": {"forward": 1, "height": 240, "type": "normal", "autoSize": 0, "width": 180}, "text": "",
# "sourceAd": ""}
# ))
# songContent = [
# {
# "type": "music",
# "data": {
# "type": "custom",
# "subtype": "kugou",
# "url": "https://webfs.yun.kugou.com/",
# "audio": "https://webfs.yun.kugou.com/202106231924/77ad13a9c9c091f74aa10098281433ab/G128/M02/02/00/wA0DAFplVJ6AIcIiADK1ZUYugV4043.mp3",
# "title": "音乐标题"
# }
# }
# ]
# await matcher.send(songContent)
# @erm.handle()
@@ -43,10 +46,5 @@ matcher = on_keyword({"test"})
# await matcher.send(record('erm'))
@matcher.args_parser
async def _(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.get_message())
if len(msg) > 5:
return
await matcher.reject()
services/service_config.py
+2
View File
@@ -6,4 +6,6 @@ TL_M_KEY = os.environ.get('TL_KEY')
SYSTEM_PROXY = os.environ.get('SYSTEM_PROXY')
ALAPI_TOKEN = os.environ.get('ALAPI_TOKEN')