first_commit .gitignore

This commit is contained in:
hibiki
2021-05-20 18:37:51 +08:00
parent df4e3aea45
commit 155f62c08a
947 changed files with 26899 additions and 0 deletions
+174
View File
@@ -0,0 +1,174 @@
from nonebot import on_regex, on_keyword
from nonebot.adapters.cqhttp import Bot, MessageEvent
from nonebot.permission import SUPERUSER
from nonebot.typing import T_State
from .genshin_handle import genshin_draw, update_genshin_info, reset_count
from .prts_handle import update_prts_info, prts_draw, reload_pool
from .pretty_handle import update_pretty_info, pretty_draw
from .update_game_info import update_info
from util.utils import is_number, scheduler
from services.log import logger
import re
prts = on_regex(r'.*?方舟[1-9|一][0-9]{0,2}[抽|井]', priority=5, block=True)
prts_update = on_keyword({'更新方舟信息', '更新明日方舟信息'}, permission=SUPERUSER, priority=1, block=True)
prts_reload = on_keyword({'重载方舟卡池'}, priority=1, block=True)
genshin = on_regex('.*?原神[1-9|一][0-9]{0,2}[抽|井]', priority=5, block=True)
genshin_reset = on_keyword({'重置原神抽卡'}, priority=1, block=True)
genshin_update = on_keyword({'更新原神信息'}, permission=SUPERUSER, priority=1, block=True)
pretty = on_regex('.*?马娘卡?[1-9|一][0-9]{0,2}[抽|井]', priority=5, block=True)
pretty_update = on_keyword({'更新马娘信息', '更新赛马娘信息'}, permission=SUPERUSER, priority=1, block=True)
@prts.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.get_message()).strip()
if msg in ['方舟一井', '方舟1井']:
num = 300
else:
rmsg = re.search(r'.*?方舟(.*)抽', msg)
if rmsg and is_number(rmsg.group(1)):
try:
num = int(rmsg.group(1))
except ValueError:
await prts.finish('必!须!是!数!字!', at_sender=True)
if num > 300:
await prts.finish('一井都满不足不了你嘛!快爬开!', at_sender=True)
if num < 1:
await prts.finish('虚空抽卡???', at_sender=True)
else:
return
# print(num)
await prts.send(await prts_draw(num), at_sender=True)
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if event.message_type != 'private' else 'private'})"
f" 方舟{num}抽")
@prts_reload.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
await reload_pool()
await prts_reload.finish('重载完成!')
@genshin.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.get_message()).strip()
if msg in ['原神一井', '原神1井']:
num = 180
else:
rmsg = re.search(r'.*?原神(.*)抽', msg)
if rmsg and is_number(rmsg.group(1)):
try:
num = int(rmsg.group(1))
except ValueError:
await genshin.finish('必!须!是!数!字!', at_sender=True)
if num > 300:
await genshin.finish('一井都满不足不了你嘛!快爬开!', at_sender=True)
if num < 1:
await genshin.finish('虚空抽卡???', at_sender=True)
else:
return
await genshin.send(await genshin_draw(event.user_id, num), at_sender=True)
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if event.message_type != 'private' else 'private'})"
f" 原神{num}抽")
@genshin_reset.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
reset_count(event.user_id)
await genshin_reset.send('重置了原神抽卡次数', at_sender=True)
@pretty.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
msg = str(event.get_message()).strip()
if msg in ['赛马娘一井', '赛马娘1井', '马娘一井', '马娘1井', '赛马娘卡一井', '赛马娘卡1井', '马娘卡一井', '马娘卡1井']:
num = 200
if msg.find("卡") == -1:
pool_name = 'horse'
else:
pool_name = 'card'
else:
rmsg = re.search(r'.*?马娘(.*)抽', msg)
if rmsg:
num = rmsg.group(1)
if num[0] == '卡':
num = num[1:]
pool_name = 'card'
else:
pool_name = 'horse'
if is_number(num):
try:
num = int(num)
except ValueError:
await genshin.finish('必!须!是!数!字!', at_sender=True)
if num > 200:
await genshin.finish('一井都满不足不了你嘛!快爬开!', at_sender=True)
if num < 1:
await genshin.finish('虚空抽卡???', at_sender=True)
else:
return
await pretty.send(await pretty_draw(num, pool_name), at_sender=True)
logger.info(
f"(USER {event.user_id}, GROUP {event.group_id if event.message_type != 'private' else 'private'})"
f" 赛马娘{num}抽")
@prts_update.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
await update_prts_info()
await reload_pool()
await prts_update.finish('更新完成!')
@genshin_update.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
await update_genshin_info()
await genshin_update.finish('更新完成!')
@pretty_update.handle()
async def _(bot: Bot, event: MessageEvent, state: T_State):
await update_pretty_info()
await genshin_update.finish('更新完成!')
# 更新资源
@scheduler.scheduled_job(
'cron',
hour=4,
minute=1,
)
async def _():
try:
await update_prts_info()
logger.info('自动更新明日方舟信息')
except Exception as e:
logger.error(f'自动更新明日方舟信息出错 e:{e}')
try:
await update_genshin_info()
logger.info('自动更新原神信息')
except Exception as e:
logger.error(f'自动更新原神信息出错 e:{e}')
try:
await update_pretty_info()
logger.info('自动更新赛马娘信息')
except Exception as e:
logger.error(f'自动更新赛马娘信息出错 e:{e}')
# 每天四点重载up卡池
@scheduler.scheduled_job(
'cron',
hour=4,
minute=1,
)
async def _():
await reload_pool()
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
+119
View File
@@ -0,0 +1,119 @@
import aiohttp
from bs4 import BeautifulSoup
import re
from datetime import datetime
from pathlib import Path
from configs.path_config import DRAW_PATH
from util.user_agent import get_user_agent
try:
import ujson as json
except ModuleNotFoundError:
import json
up_char_file = Path(DRAW_PATH) / "draw_card_up" / "prts_up_char.json"
prts_url = "https://wiki.biligame.com/arknights/%E6%96%B0%E9%97%BB%E5%85%AC%E5%91%8A"
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():
async with aiohttp.ClientSession(headers=get_user_agent()) 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]
@staticmethod
async def update_up_char():
up_char_file.parent.mkdir(parents=True, exist_ok=True)
data = {'up_char': {'6': {}, '5': {}, '4': {}}, 'title': '', 'time': ''}
text, title = await PrtsAnnouncement.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:59'
if not is_expired(data):
data['title'] = ''
else:
with open(up_char_file, 'w', encoding='utf8') as f:
json.dump(data, f, indent=4, ensure_ascii=False)
if not up_char_file.exists():
with open(up_char_file, 'w', encoding='utf8') as f:
json.dump(data, f, indent=4, ensure_ascii=False)
else:
with open(up_char_file, 'r', encoding='utf8') as f:
old_data = json.load(f)
if is_expired(old_data):
return old_data
else:
with open(up_char_file, 'w', encoding='utf8') as f:
json.dump(data, f, indent=4, ensure_ascii=False)
return data
# 是否过时
def is_expired(data: dict):
times = data['time'].split('-')
for i in range(len(times)):
times[i] = str(datetime.now().year) + '-' + times[i].split('日')[0].strip().replace('月', '-')
start_date = datetime.strptime(times[0], '%Y-%m-%d').date()
end_date = datetime.strptime(times[1], '%Y-%m-%d').date()
now = datetime.now().date()
return start_date < now < end_date
# ad = Announcement('https://wiki.biligame.com/arknights/%E6%96%B0%E9%97%BB%E5%85%AC%E5%91%8A')
# asyncio.get_event_loop().run_until_complete(check_up_char('prts'))
+104
View File
@@ -0,0 +1,104 @@
import nonebot
from pathlib import Path
try:
import ujson as json
except ModuleNotFoundError:
import json
# 方舟概率
PRTS_SIX_P = 0.02
PRTS_FIVE_P = 0.08
PRTS_FOUR_P = 0.48
PRTS_THREE_P = 0.42
# 原神概率
GENSHIN_FIVE_P = 0.006
GENSHIN_FOUR_P = 0.051
GENSHIN_THREE_P = 0.43
GENSHIN_G_FOUR_P = 0.13
GENSHIN_G_FIVE_P = 0.016
I72_ADD = 0.0585
# 赛马娘概率
PRETTY_THREE = 0.03
PRETTY_TWO = 0.18
PRETTY_ONE = 0.79
path_dict = {
'genshin': '原神',
'prts': '明日方舟',
'pretty': '赛马娘',
}
_draw_config = Path() / "data" / "draw_card" / "draw_card_config" / "draw_card_config.json"
driver: nonebot.Driver = nonebot.get_driver()
@driver.on_startup
def check_config():
global PRTS_SIX_P, PRTS_FOUR_P, PRTS_FIVE_P, PRTS_THREE_P, GENSHIN_G_FIVE_P, \
GENSHIN_G_FOUR_P, GENSHIN_FOUR_P, GENSHIN_FIVE_P, I72_ADD, path_dict, PRETTY_THREE, \
PRETTY_ONE, PRETTY_TWO, GENSHIN_THREE_P
if _draw_config.exists():
data = json.load(open(_draw_config, 'r', encoding='utf8'))
PRTS_SIX_P = float(data['prts']['six'])
PRTS_FIVE_P = float(data['prts']['five'])
PRTS_FOUR_P = float(data['prts']['four'])
PRTS_THREE_P = float(data['prts']['three'])
GENSHIN_FIVE_P = float(data['genshin']['five_char'])
GENSHIN_FOUR_P = float(data['genshin']['four_char'])
GENSHIN_THREE_P = float(data['genshin']['three_char'])
GENSHIN_G_FIVE_P = float(data['genshin']['five_weapon'])
GENSHIN_G_FOUR_P = float(data['genshin']['four_weapon'])
I72_ADD = float(data['genshin']['72_add'])
PRETTY_THREE = float(data['pretty']['three'])
PRETTY_TWO = float(data['pretty']['two'])
PRETTY_ONE = float(data['pretty']['one'])
path_dict = data['path_dict']
else:
_draw_config.parent.mkdir(parents=True, exist_ok=True)
config_dict = {
'path_dict': {
'genshin': '原神',
'prts': '明日方舟',
'pretty': '赛马娘',
},
'prts': {
'six': 0.02,
'five': 0.08,
'four': 0.48,
'three': 0.42,
},
'genshin': {
'five_char': 0.006,
'four_char': 0.051,
'three_char': 0.43,
'five_weapon': 0.13,
'four_weapon': 0.016,
'72_add': 0.0585,
},
'pretty': {
'three': 0.03,
'two': 0.18,
'one': 0.79,
}
}
json.dump(config_dict, open(_draw_config, 'w', encoding='utf8'), indent=4, ensure_ascii=False)
+149
View File
@@ -0,0 +1,149 @@
import os
import nonebot
import random
from .update_game_info import update_info
from .util import generate_img, init_star_rst, BaseData, set_list
from .config import GENSHIN_FIVE_P, GENSHIN_FOUR_P, GENSHIN_G_FIVE_P, GENSHIN_THREE_P, I72_ADD
from dataclasses import dataclass
from .init_card_pool import init_game_pool
from configs.path_config import DRAW_PATH
from util.init_result import image
try:
import ujson as json
except ModuleNotFoundError:
import json
driver: nonebot.Driver = nonebot.get_driver()
genshin_five = {}
genshin_count = {}
genshin_pl_count = {}
ALL_CHAR = []
ALL_ARM = []
@dataclass
class GenshinChar(BaseData):
pass
async def genshin_draw(user_id: int, count: int):
# 0 1 2
cnlist = ['★★★★★', '★★★★', '★★★']
genshin_list, five_list, five_olist, five_dict, star_list = _format_card_information(count, user_id)
rst = init_star_rst(star_list, cnlist, five_list, five_olist)
print(five_list)
temp = ''
if count > 90:
genshin_list = set_list(genshin_list)
return image(b64=await generate_img(genshin_list, 'genshin', star_list)) + '\n' + rst[:-1] + \
temp[:-1] + f'\n距离保底发还剩 {90 - genshin_count[user_id] if genshin_count.get(user_id) else "^"} 抽' \
+ "\n【五星:0.6%,四星:5.1%\n第72抽开始五星概率每抽加0.585%】"
async def update_genshin_info():
global ALL_CHAR, ALL_ARM
url = 'https://wiki.biligame.com/ys/角色筛选'
data, code = await update_info(url, 'genshin')
if code == 200:
ALL_CHAR = init_game_pool('genshin', data, GenshinChar)
url = 'https://wiki.biligame.com/ys/武器图鉴'
data, code = await update_info(url, 'genshin_arm', ['头像', '名称', '类型', '稀有度.alt', '初始基础属性1',
'初始基础属性2', '攻击力(MAX)', '副属性(MAX)', '技能'])
if code == 200:
ALL_ARM = init_game_pool('genshin', data, GenshinChar)
# asyncio.get_event_loop().run_until_complete(update_genshin_info())
@driver.on_startup
async def init_data():
global ALL_CHAR, ALL_ARM
if not os.path.exists(DRAW_PATH + '/draw_card_config/genshin.json') or \
not os.path.exists(DRAW_PATH + '/draw_card_config/genshin_arm.json'):
await update_genshin_info()
else:
with open(DRAW_PATH + '/draw_card_config/genshin.json', 'r', encoding='utf8') as f:
genshin_dict = json.load(f)
with open(DRAW_PATH + '/draw_card_config/genshin_arm.json', 'r', encoding='utf8') as f:
genshin_arm_dict = json.load(f)
ALL_CHAR = init_game_pool('genshin', genshin_dict, GenshinChar)
ALL_ARM = init_game_pool('genshin', genshin_arm_dict, GenshinChar)
# 抽取卡池
def _get_genshin_card(mode: int = 1, add: float = 0.0):
global ALL_ARM, ALL_CHAR
if mode == 1:
star = random.sample([5, 4, 3],
counts=[int(GENSHIN_FIVE_P * 1000) + int(add * 1000), int(GENSHIN_FOUR_P * 1000),
int(GENSHIN_THREE_P * 1000)],
k=1)[0]
elif mode == 2:
star = random.sample([5, 4],
counts=[int(GENSHIN_G_FIVE_P * 1000) + int(add * 1000), int(GENSHIN_FOUR_P * 1000)],
k=1)[0]
else:
star = 5
chars = [x for x in (ALL_ARM if random.random() < 0.5 or star == 3 else ALL_CHAR) if x.star == star]
return random.choice(chars), abs(star - 5)
def _format_card_information(_count: int, user_id):
genshin_list = []
star_list = [0, 0, 0]
five_index_list = []
five_list = []
five_dict = {}
add = 0.0
if genshin_count.get(user_id) and _count <= 90:
f_count = genshin_count[user_id]
else:
f_count = 0
if genshin_pl_count.get(user_id) and _count <= 90:
count = genshin_pl_count[user_id]
else:
count = 0
for i in range(_count):
count += 1
f_count += 1
# 十连保底
if count == 10 and f_count != 90:
if f_count >= 72:
add += I72_ADD
char, code = _get_genshin_card(2, add)
count = 0
# 大保底
elif f_count == 90:
char, code = _get_genshin_card(3)
else:
if f_count >= 72:
add += I72_ADD
char, code = _get_genshin_card(add=add)
if code == 1:
count = 0
star_list[code] += 1
if code == 0:
if _count <= 90:
genshin_five[user_id] = f_count
add = 0.0
f_count = 0
five_list.append(char.name)
five_index_list.append(i)
try:
five_dict[char.name] += 1
except KeyError:
five_dict[char.name] = 1
genshin_list.append(char)
if _count <= 90:
genshin_count[user_id] = f_count
genshin_pl_count[user_id] = count
return genshin_list, five_list, five_index_list, five_dict, star_list
def reset_count(user_id: int):
genshin_count[user_id] = 0
genshin_pl_count[user_id] = 0
+33
View File
@@ -0,0 +1,33 @@
from typing import Any
def init_game_pool(game: str, data: dict, Operator: Any):
tmp_lst = []
if game == 'prts':
for key in data.keys():
limited = False
recruit_only = False
event_only = False
if '限定寻访' in data[key]['获取途径']:
limited = True
if len(data[key]['获取途径']) == 1 and data[key]['获取途径'][0] == '公开招募':
recruit_only = True
if '活动获取' in data[key]['获取途径']:
event_only = True
if key.find('阿米娅') != -1:
continue
tmp_lst.append(Operator(name=key, star=int(data[key]['星级']),
limited=limited, recruit_only=recruit_only, event_only=event_only))
if game == 'genshin':
for key in data.keys():
if key.find('旅行者') != -1:
continue
tmp_lst.append(Operator(name=key, star=int(data[key]['稀有度'][:1]), limited=False))
if game == 'pretty':
for key in data.keys():
tmp_lst.append(Operator(name=key, star=data[key]['初始星级'], limited=False))
if game == 'pretty_card':
for key in data.keys():
tmp_lst.append(Operator(name=data[key]['中文名'], star=len(data[key]['稀有度']), limited=False))
return tmp_lst
+105
View File
@@ -0,0 +1,105 @@
import os
import nonebot
from util.init_result import image
from configs.path_config import DRAW_PATH
from .update_game_info import update_info
from .util import download_img, init_star_rst, generate_img, max_card, BaseData, set_list
import random
from .config import PRETTY_THREE, PRETTY_TWO, PRETTY_ONE
from dataclasses import dataclass
from .init_card_pool import init_game_pool
try:
import ujson as json
except ModuleNotFoundError:
import json
driver: nonebot.Driver = nonebot.get_driver()
ALL_CHAR = []
ALL_CARD = []
@dataclass
class PrettyChar(BaseData):
pass
async def pretty_draw(count: int, pool_name):
if pool_name == 'card':
cnlist = ['SSR', 'SR', 'R']
else:
cnlist = ['★★★', '★★', '★']
obj_list, obj_dict, three_list, star_list, three_olist = _format_card_information(count, pool_name)
rst = init_star_rst(star_list, cnlist, three_list, three_olist)
if count > 90:
obj_list = set_list(obj_list)
return image(b64=await generate_img(obj_list, 'pretty', star_list)) \
+ '\n' + rst[:-1] + '\n' + max_card(obj_dict)
async def update_pretty_info():
global ALL_CHAR, ALL_CARD
url = 'https://wiki.biligame.com/umamusume/赛马娘图鉴'
data, code = await update_info(url, 'pretty')
if code == 200:
ALL_CHAR = init_game_pool('pretty', data, PrettyChar)
url = 'https://wiki.biligame.com/umamusume/支援卡图鉴'
data, code = await update_info(url, 'pretty_card')
if code == 200:
ALL_CARD = init_game_pool('pretty_card', data, PrettyChar)
@driver.on_startup
async def init_data():
global ALL_CHAR, ALL_CARD
if not os.path.exists(DRAW_PATH + '/draw_card_config/pretty.json') or\
not os.path.exists(DRAW_PATH + '/draw_card_config/pretty_card.json'):
await update_pretty_info()
for icon_url in [
'https://patchwiki.biligame.com/images/umamusume/thumb/0/06/q23szwkbtd7pfkqrk3wcjlxxt9z595o.png'
'/40px-SSR.png',
'https://patchwiki.biligame.com/images/umamusume/thumb/3/3b/d1jmpwrsk4irkes1gdvoos4ic6rmuht.png'
'/40px-SR.png',
'https://patchwiki.biligame.com/images/umamusume/thumb/f/f7/afqs7h4snmvovsrlifq5ib8vlpu2wvk.png'
'/40px-R.png']:
await download_img(icon_url, 'pretty', icon_url.split('-')[-1][:-4])
else:
with open(DRAW_PATH + '/draw_card_config/pretty.json', 'r', encoding='utf8') as f:
pretty_char_dict = json.load(f)
with open(DRAW_PATH + '/draw_card_config/pretty_card.json', 'r', encoding='utf8') as f:
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)
# 抽取卡池
def _get_pretty_card(itype):
global ALL_CHAR, ALL_CARD
star = random.sample([3, 2, 1],
counts=[int(PRETTY_THREE * 100), int(PRETTY_TWO * 100),
int(PRETTY_ONE * 100)],
k=1)[0]
chars = [x for x in (ALL_CARD if itype == 'card' else ALL_CHAR) if x.star == star]
return random.choice(chars), abs(star - 3)
# 整理数据
def _format_card_information(count: int, pool_name: str):
three_list = []
three_olist = []
obj_list = []
obj_dict = {}
star_list = [0, 0, 0]
for i in range(count):
obj, code = _get_pretty_card(pool_name)
star_list[code] += 1
if code == 0:
three_list.append(obj.name)
three_olist.append(i)
try:
obj_dict[obj.name] += 1
except KeyError:
obj_dict[obj.name] = 1
obj_list.append(obj)
return obj_list, obj_dict, three_list, star_list, three_olist
+162
View File
@@ -0,0 +1,162 @@
import os
import nonebot
import random
from .config import PRTS_FIVE_P, PRTS_FOUR_P, PRTS_SIX_P, PRTS_THREE_P
from .update_game_info import update_info
from .util import generate_img, init_star_rst, max_card, BaseData, UpEvent, set_list
from .init_card_pool import init_game_pool
from pathlib import Path
from .announcement import PrtsAnnouncement
from dataclasses import dataclass
from util.init_result import image
from configs.path_config import DRAW_PATH
from services.log import logger
try:
import ujson as json
except ModuleNotFoundError:
import json
driver: nonebot.Driver = nonebot.get_driver()
up_char_file = Path() / "data" / "draw_card" / "draw_card_up" / "prts_up_char.json"
prts_dict = {}
UP_OPERATOR = []
ALL_OPERATOR = []
_CURRENT_POOL_TITLE = ''
@dataclass
class Operator(BaseData):
recruit_only: bool # 公招限定
event_only: bool # 活动获得干员
# special_only: bool # 升变/异格干员
async def prts_draw(count: int = 300):
cnlist = ['★★★★★★', '★★★★★', '★★★★', '★★★']
operator_list, operator_dict, six_list, star_list, six_olist = _format_card_information(count)
up_list = []
if _CURRENT_POOL_TITLE:
for x in UP_OPERATOR:
for operator in x.operators:
up_list.append(operator)
rst = init_star_rst(star_list, cnlist, six_list, six_olist, 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 + image(b64=await generate_img(operator_list, 'prts', star_list)) \
+ '\n' + rst[:-1] + '\n' + max_card(operator_dict)
async def update_prts_info():
global prts_dict, ALL_OPERATOR
url = 'https://wiki.biligame.com/arknights/干员数据表'
data, code = await update_info(url, 'prts', ['头像', '名称', '阵营', '星级', '性别', '是否感染', '初始生命', '初始防御',
'初始法抗', '再部署', '部署费用', '阻挡数', '攻击速度', '标签'])
if code == 200:
prts_dict = data
ALL_OPERATOR = init_game_pool('prts', prts_dict, Operator)
@driver.on_startup
async def init_data():
global prts_dict, ALL_OPERATOR
if not os.path.exists(DRAW_PATH + '/draw_card_config/prts.json'):
await update_prts_info()
else:
with open(DRAW_PATH + '/draw_card_config/prts.json', 'r', encoding='utf8') as f:
prts_dict = json.load(f)
ALL_OPERATOR = init_game_pool('prts', prts_dict, Operator)
await _init_up_char()
# print([x.operators for x in UP_OPERATOR if x.star == 5 and x.zoom > 1])
# 抽取干员
def _get_operator_card():
star = random.sample([6, 5, 4, 3],
counts=[int(PRTS_SIX_P * 100), int(PRTS_FIVE_P * 100),
int(PRTS_FOUR_P * 100), int(PRTS_THREE_P * 100)],
k=1)[0]
if _CURRENT_POOL_TITLE:
zooms = [x.zoom for x in UP_OPERATOR if x.star == star]
zoom = 0
weight = 0
# 分配概率和权重
for z in zooms:
if z < 1:
zoom = z
else:
weight = z
# UP
if random.random() < zoom:
up_operators = [x.operators for x in UP_OPERATOR if x.star == star and x.zoom < 1][0]
up_operator_name = random.choice(up_operators)
# print(up_operator_name)
acquire_operator = [x for x in ALL_OPERATOR if x.name == up_operator_name][0]
else:
all_star_operators = [x for x in ALL_OPERATOR if x.star == star
and not any([x.limited, x.event_only, x.recruit_only])]
weight_up_operators = [x.operators for x in UP_OPERATOR if x.star == star and x.zoom > 1]
# 权重
if weight_up_operators and random.random() < 1.0 / float(len(all_star_operators)) * weight:
up_operator_name = random.choice(weight_up_operators[0])
acquire_operator = [x for x in ALL_OPERATOR if x.name == up_operator_name][0]
else:
acquire_operator = random.choice(all_star_operators)
else:
acquire_operator = random.choice([x for x in ALL_OPERATOR if x.star == star
and not any([x.limited, x.event_only, x.recruit_only])])
# print(f'{acquire_operator}: {star}')
return acquire_operator, abs(star - 6)
# 整理抽卡数据
def _format_card_information(count: int):
operator_list = [] # 抽取的干员列表
operator_dict = {} # 抽取各干员次数
star_list = [0, 0, 0, 0] # 各个星级次数
six_list = [] # 六星干员列表
six_index_list = [] # 六星干员获取位置
for i in range(count):
operator, code = _get_operator_card()
star_list[code] += 1
if code == 0:
six_list.append(operator.name)
six_index_list.append(i)
try:
operator_dict[operator.name] += 1
except KeyError:
operator_dict[operator.name] = 1
operator_list.append(operator)
return operator_list, operator_dict, six_list, star_list, six_index_list
# 获取up干员和概率
async def _init_up_char():
global up_char_dict, _CURRENT_POOL_TITLE
up_char_dict = await PrtsAnnouncement.update_up_char()
# print(up_char_dict)
_CURRENT_POOL_TITLE = up_char_dict['title']
up_char_dict = up_char_dict['up_char']
logger.info(f'成功获取明日方舟当前up信息...当前up池: {_CURRENT_POOL_TITLE}')
average_dict = {'6': {}, '5': {}, '4': {}}
for star in up_char_dict.keys():
for key in up_char_dict[star].keys():
if average_dict[star].get(up_char_dict[star][key]):
average_dict[star][up_char_dict[star][key]].append(key)
else:
average_dict[star][up_char_dict[star][key]] = [key]
up_char_dict = {'6': {}, '5': {}, '4': {}}
for star in average_dict.keys():
for str_zoom in average_dict[star].keys():
if str_zoom[0] == '权':
zoom = float(str_zoom[1:])
else:
zoom = float(str_zoom) / 100
UP_OPERATOR.append(UpEvent(star=int(star), operators=average_dict[star][str_zoom], zoom=zoom))
async def reload_pool():
await _init_up_char()
+162
View File
@@ -0,0 +1,162 @@
#coding:utf-8
import aiohttp
from configs.path_config import DRAW_PATH
from asyncio.exceptions import TimeoutError
from services.log import logger
from bs4 import BeautifulSoup
from .util import download_img
from urllib.parse import unquote
import bs4
from util.user_agent import get_user_agent
import re
try:
import ujson as json
except ModuleNotFoundError:
import json
async def update_info(url: str, game_name: str, info_list: list = None) -> 'dict, int':
try:
with open(DRAW_PATH + f'/draw_card_config/{game_name}.json', 'r', encoding='utf8') as f:
data = json.load(f)
except (ValueError, FileNotFoundError):
data = {}
try:
async with aiohttp.ClientSession(headers=get_user_agent()) as session:
async with session.get(url, timeout=7) as response:
soup = BeautifulSoup(await response.text(), 'lxml')
max_count = 0
_tbody = None
for tbody in soup.find_all('tbody'):
if len(tbody.find_all('tr')) > max_count:
_tbody = tbody
max_count = len(tbody.find_all('tr'))
trs = _tbody.find_all('tr')
att_dict = {'头像': 0, '名称': 1}
index = 2
for th in trs[0].find_all('th')[2:]:
text = th.text
if text[-1] == '\n':
text = text[:-1]
att_dict[text] = index
index += 1
for tr in trs[1:]:
member_dict = {}
k_name = ''
tds = tr.find_all('td')
if not info_list:
info_list = att_dict.keys()
for key in info_list:
attr = ''
if key.find('.') != -1:
key = key.split('.')
attr = key[-1]
key = key[0]
td = tds[att_dict[key]]
last_tag = unquote(_find_last_tag(td, attr), 'utf-8')
if game_name.find('pretty') == -1 and last_tag.find('http') == -1:
last_tag = last_tag.split('.')[0]
if key == '名称':
k_name = last_tag
member_dict[key] = last_tag
if game_name == 'pretty' and key == '初始星级':
member_dict['初始星级'] = len(td.find_all('img'))
avatar_img = await _modify_avatar_url(session, game_name, member_dict["名称"])
if avatar_img:
member_dict['头像'] = avatar_img
name = member_dict['名称']
if game_name == 'pretty_card':
name = member_dict['中文名']
await download_img(member_dict['头像'], game_name, name)
if k_name:
data[k_name] = member_dict
logger.info(f'{k_name} is update...')
data = await _last_check(data, game_name, session)
except TimeoutError:
return {}, 999
with open(DRAW_PATH + f'/draw_card_config/{game_name}.json', 'w', encoding='utf8') as wf:
wf.write(json.dumps(data, ensure_ascii=False, indent=4))
return data, 200
def _find_last_tag(element: bs4.element.Tag, attr: str) -> str:
last_tag = []
for des in element.descendants:
last_tag.append(des)
if len(last_tag) == 1 and last_tag[0] == '\n':
last_tag = ''
elif last_tag[-1] == '\n':
last_tag = last_tag[-2]
else:
last_tag = last_tag[-1]
if attr and str(last_tag):
last_tag = last_tag[attr]
elif str(last_tag).find('<img') != -1:
if last_tag.get('srcset'):
last_tag = str(last_tag.get('srcset')).strip().split(' ')[-2].strip()
else:
last_tag = last_tag['src']
else:
last_tag = str(last_tag)
if str(last_tag) and str(last_tag)[-1] == '\n':
last_tag = str(last_tag)[:-1]
return last_tag
# 获取大图(小图快爬)
async def _modify_avatar_url(session: aiohttp.ClientSession, game_name: str, char_name: str):
if game_name == 'prts':
async with session.get(f'https://wiki.biligame.com/arknights/{char_name}', timeout=7) as res:
soup = BeautifulSoup(await res.text(), 'lxml')
try:
img_url = str(soup.find('img', {'class': 'img-bg'})['srcset']).split(' ')[-2]
except KeyError:
img_url = str(soup.find('img', {'class': 'img-bg'})['src'])
return img_url
if game_name == 'genshin':
return None
if game_name == 'pretty_card':
async with session.get(f'https://wiki.biligame.com/umamusume/{char_name}', timeout=7) as res:
soup = BeautifulSoup(await res.text(), 'lxml')
img_url = soup.find('div', {'class': 'support_card-left'}).find('div').find('img').get('src')
return img_url
# 数据最后处理(是否需要额外数据)
async def _last_check(data: dict, game_name: str, session: aiohttp.ClientSession):
if game_name == 'prts':
for key in data.keys():
async with session.get(f'https://wiki.biligame.com/arknights/{key}', timeout=7) as res:
soup = BeautifulSoup(await res.text(), 'lxml')
obtain = str(soup.find('table', {'class': 'wikitable'}).find('tbody').find_all('td')[-1])
obtain = re.search(r'<td.*?>([\s\S]*)</.*?', obtain).group(1)
obtain = obtain[:-1] if obtain[-1] == '\n' else obtain
if obtain.find('<br/>'):
obtain = obtain.split('<br/>')
elif obtain.find('<br>'):
obtain = obtain.split('<br>')
data[key]['获取途径'] = obtain
# if game_name == 'genshin':
# for key in data.keys():
# async with session.get(f'https://wiki.biligame.com/ys/{key}', timeout=7) as res:
# soup = BeautifulSoup(await res.text(), 'lxml')
# trs = soup.find('div', {'class': 'poke-bg'}).find('table').find('tbody').find_all('tr')
# for tr in trs:
# if tr.find('th').text.find('常驻/限定') != -1:
# data[key]['常驻/限定'] = tr.find('td').text
# break
if game_name == 'pretty':
for keys in data.keys():
for key in data[keys].keys():
# print(f'key --> {data[keys][key]}')
r = re.search(r'.*?40px-(.*)图标.png', str(data[keys][key]))
if r:
data[keys][key] = r.group(1)
return data
# ul = soup.find('div', {'class': 'savelist_bot'}).find('ul')
+161
View File
@@ -0,0 +1,161 @@
import os
import aiohttp
import aiofiles
from asyncio.exceptions import TimeoutError
from aiohttp.client_exceptions import InvalidURL
from typing import List, Union, Set
import asyncio
from pathlib import Path
from .config import path_dict
import nonebot
from util.utils import cn2py
from util.img_utils import CreateImg
from util.user_agent import get_user_agent
from configs.path_config import IMAGE_PATH
from dataclasses import dataclass
from services.log import logger
try:
import ujson as json
except ModuleNotFoundError:
import json
driver: nonebot.Driver = nonebot.get_driver()
@dataclass
class BaseData:
name: str
star: int
limited: bool # 限定
@dataclass
class UpEvent:
star: int # 对应up星级
operators: List[BaseData] # 干员列表
zoom: float # up提升倍率
async def download_img(url: str, path: str, name: str) -> bool:
path = path.split('_')[0]
codename = cn2py(name)
# if not _p.exists():
# _p.mkdir(parents=True, exist_ok=True)
if not os.path.exists(IMAGE_PATH + f'/draw_card/{path}/{codename}.png'):
try:
async with aiohttp.ClientSession(headers=get_user_agent()) as session:
async with session.get(url, timeout=7) as response:
async with aiofiles.open(IMAGE_PATH + f'/draw_card/{path}/{codename}.png', 'wb') as f:
await f.write(await response.read())
logger.info(f'下载 {path_dict[path]} 图片成功,名称:{name},url:{url}')
return True
except TimeoutError:
logger.info(f'下载 {path_dict[path]} 图片超时,名称:{name},url:{url}')
return False
except InvalidURL:
logger.info(f'下载 {path_dict[path]} 链接错误,名称:{name},url:{url}')
return False
else:
# logger.info(f'{path_dict[path]} 图片 {name} 已存在')
return False
@driver.on_startup
def _check_dir():
for dir_name in path_dict.keys():
_p = Path(IMAGE_PATH + f'/draw_card/' + dir_name)
if not _p.exists():
_p.mkdir(parents=True, exist_ok=True)
async def generate_img(card_set: Union[Set[BaseData], List[BaseData]], game_name: str, star_list: list) -> str:
# try:
img_list = []
color_list = []
for x in card_set:
if game_name == 'prts':
if x.star == 6:
color_list.append('#FFD700')
elif x.star == 5:
color_list.append('#DAA520')
elif x.star == 4:
color_list.append('#9370D8')
else:
color_list.append('white')
pyname = cn2py(x.name)
img_list.append(IMAGE_PATH + f'/draw_card/{game_name}/{pyname}.png')
img_len = len(img_list)
w = 100 * 10
if img_len <= 10:
w = 100 * img_len
h = 100
elif img_len % 10 == 0:
h = 100 * int(img_len / 10)
else:
h = 100 * int(img_len / 10) + 100
card_img = await asyncio.get_event_loop().run_in_executor(None, _pst, h, img_list, game_name, color_list)
num = 0
for n in star_list:
num += n
A = CreateImg(w, h)
A.paste(card_img)
return A.pic2bs4()
def _pst(h: int, img_list: list, game_name: str, color_list: list):
card_img = CreateImg(100 * 10, h, 100, 100)
idx = 0
for img in img_list:
try:
if game_name == 'prts':
bk = CreateImg(100, 100, color=color_list[idx])
b = CreateImg(94, 94, background=img)
bk.paste(b, (3, 3))
b = bk
idx += 1
else:
b = CreateImg(100, 100, background=img)
except FileNotFoundError:
print(f'{img} not exists')
b = CreateImg(100, 100, color='black')
card_img.paste(b)
return card_img
def init_star_rst(star_list: list, cnlist: list, max_star_list: list, max_star_olist: list, up_list: list = None) -> str:
if not up_list:
up_list = []
rst = ''
for i in range(len(star_list)):
if star_list[i]:
rst += f'[{cnlist[i]}×{star_list[i]}] '
rst += '\n'
for i in range(len(max_star_list)):
if max_star_list[i] in up_list:
rst += f'第 {max_star_olist[i]+1} 抽获取UP {max_star_list[i]}\n'
else:
rst += f'第 {max_star_olist[i]+1} 抽获取 {max_star_list[i]}\n'
return rst
def max_card(_dict: dict):
_max_value = max(_dict.values())
_max_user = list(_dict.keys())[list(_dict.values()).index(_max_value)]
return f'抽取到最多的是{_max_user},共抽取了{_max_value}次'
# ThreeHighest = nlargest(3, operator_dict, key=operator_dict.get)
# rst = '最喜欢你的前三位是干员是:\n'
# for name in ThreeHighest:
# rst += f'{name} 共投了 {operator_dict[name]} 份简历\n'
# return rst[:-1]
def set_list(lst: List[BaseData]) -> list:
tmp = []
name_lst = []
for x in lst:
if x.name not in name_lst:
tmp.append(x)
name_lst.append(x.name)
return tmp