231 lines
10 KiB
Python
231 lines
10 KiB
Python
import copy
|
|
import json
|
|
import os.path
|
|
import requests
|
|
import time
|
|
import numpy as np
|
|
import datetime
|
|
|
|
import matplotlib.pyplot as plt
|
|
|
|
from logger_module import logger
|
|
from src.util.utils import hg_get_new_user_token
|
|
|
|
# proxies = {
|
|
# "http": "http://127.0.0.1:10809",
|
|
# "https": "http://127.0.0.1:10809",
|
|
# }
|
|
proxies = None
|
|
|
|
|
|
class ArknightsInfo:
|
|
def __init__(self, database_path, raw_data=None):
|
|
|
|
self.gacha_data = None
|
|
self.payment_path = os.path.join(database_path, 'arknights_payment_data')
|
|
self.payment_data = self.load_info_from_file(self.payment_path)
|
|
|
|
self.gacha_data_path = os.path.join(database_path, 'arknights_gacha_data')
|
|
self.token = self.load_info_from_file(os.path.join(database_path, 'token'))
|
|
self.userinfo = self.load_info_from_file(os.path.join(database_path, 'userinfo.txt'))
|
|
self.data_dict_by_banner, self.all_result_for_show, self.raw_pull_data = self.convert_data(raw_data)
|
|
|
|
self.logger = logger.Logger(database_path).get_log()
|
|
|
|
self.update_info()
|
|
|
|
def update_info(self):
|
|
self.login()
|
|
self.gacha_data = self.update_gacha_database()
|
|
self.get_payment_data()
|
|
self.logger.info('Info updated')
|
|
|
|
@staticmethod
|
|
def load_info_from_file(local_path):
|
|
file = open(local_path, 'r', encoding='utf-8')
|
|
result = json.loads(file.readlines()[0])
|
|
file.close()
|
|
if isinstance(result, dict): # 按照key进行排序
|
|
sort_data_list = sorted(result.items(), key=lambda x: x[0])
|
|
result = {item[0]: item[1] for item in sort_data_list}
|
|
return result
|
|
|
|
def get_new_user_token(self):
|
|
username = self.userinfo.get('username')
|
|
password = self.userinfo.get('password')
|
|
token_data = hg_get_new_user_token(username, password)
|
|
self.token = token_data
|
|
|
|
def login(self):
|
|
get_gacha_url = f'https://ak.hypergryph.com/user/api/inquiry/gacha?page=1&token={self.token}&channelId=1'
|
|
gacha_response = requests.get(get_gacha_url, proxies=proxies).text
|
|
if gacha_response == '{"code":3000,"msg":"登录失效"}':
|
|
self.get_new_user_token()
|
|
get_gacha_url = f'https://ak.hypergryph.com/user/api/inquiry/gacha?page=1&token={self.token}&channelId=1'
|
|
gacha_response = requests.get(get_gacha_url, proxies=proxies).text
|
|
if gacha_response == '{"code":3000,"msg":"登录失效"}':
|
|
print('用户名/密码错误')
|
|
raise ValueError
|
|
else:
|
|
file = open('data/token', 'w', encoding='utf-8')
|
|
file.write(json.dumps(self.token))
|
|
file.close()
|
|
|
|
def update_gacha_database(self):
|
|
history_data = self.load_info_from_file(self.gacha_data_path)
|
|
session = requests.session()
|
|
newly_extracted_data = {}
|
|
for i in range(1, 11):
|
|
get_gacha_url = f'https://ak.hypergryph.com/user/api/inquiry/gacha?page={i}&token={self.token}&channelId=1'
|
|
gacha_response = session.get(get_gacha_url, proxies=proxies).text
|
|
tmp_data = json.loads(gacha_response).get('data')
|
|
if not tmp_data.get('list'):
|
|
break
|
|
for item in tmp_data['list']:
|
|
pull_id = str(item['ts'])
|
|
banner = item['pool']
|
|
new_data = [[i['name'], i['rarity'], 1 if i['isNew'] else 0] for i in item['chars']]
|
|
newly_extracted_data[pull_id] = {'c': new_data, 'p': banner}
|
|
sort_data_list = sorted(newly_extracted_data.items(), key=lambda x: x[0])
|
|
sorted_data = {item[0]: item[1] for item in sort_data_list}
|
|
history_data.update(sorted_data)
|
|
file = open(self.gacha_data_path, 'w', encoding='utf-8')
|
|
file.write(json.dumps(history_data))
|
|
file.close()
|
|
self.data_dict_by_banner, self.all_result_for_show, self.raw_pull_data = self.convert_data()
|
|
return history_data
|
|
|
|
def convert_data(self, raw_data=None):
|
|
if not raw_data:
|
|
content = self.load_info_from_file(self.gacha_data_path)
|
|
else:
|
|
content = raw_data
|
|
banner_result = {}
|
|
banner_result_by_time = {}
|
|
total_gacha_result = []
|
|
for key, value in content.items():
|
|
pull_time = datetime.datetime.fromtimestamp(int(key)).strftime('%Y-%m-%d %H:%M:%S')
|
|
banner_result_by_time[pull_time] = (copy.deepcopy(value['c']), datetime.datetime.fromtimestamp(int(key)))
|
|
if value['p'] not in banner_result:
|
|
banner_result[value['p']] = {'gacha_result': copy.deepcopy(value['c']),
|
|
'gacha_result_by_time': {pull_time: (copy.deepcopy(value['c']), datetime.datetime.fromtimestamp(int(key)))}}
|
|
else:
|
|
banner_result[value['p']]['gacha_result'] += copy.deepcopy(value['c'])
|
|
banner_result[value['p']]['gacha_result_by_time'][pull_time] = (copy.deepcopy(value['c']), datetime.datetime.fromtimestamp(int(key)))
|
|
total_gacha_result += copy.deepcopy(value['c'])
|
|
for key, value in banner_result.items():
|
|
banner_result[key]['analysis'] = self.analyze_gacha_data(value['gacha_result'])
|
|
all_result = self.analyze_gacha_data(total_gacha_result)
|
|
banner_result['全部结果'] = {'analysis': all_result, 'gacha_result_by_time': banner_result_by_time}
|
|
|
|
return banner_result, all_result, content
|
|
|
|
def show_all_gacha_data(self):
|
|
self.gacha_data_visualization(self.all_result_for_show)
|
|
|
|
@staticmethod
|
|
def analyze_gacha_data(data_list):
|
|
result = {
|
|
'by_star': {'3': 0, '4': 0, '5': 0, '6': 0},
|
|
'new_operator': []
|
|
}
|
|
for data in data_list:
|
|
result['by_star'][str(data[1] + 1)] += 1
|
|
if data[2] == 1:
|
|
result['new_operator'].append(data)
|
|
result['six_star_percent'] = result['by_star']['6'] / len(data_list) * 100
|
|
result['five_star_percent'] = result['by_star']['5'] / len(data_list) * 100
|
|
result['four_star_percent'] = result['by_star']['4'] / len(data_list) * 100
|
|
result['three_star_percent'] = result['by_star']['3'] / len(data_list) * 100
|
|
result['total_pull'] = len(data_list)
|
|
return result
|
|
|
|
@staticmethod
|
|
def gacha_data_visualization(data_dict, title='Banner'):
|
|
plt.rcParams['font.sans-serif'] = 'SimHei' # 设置中文显示
|
|
plt.figure(figsize=(6, 6))
|
|
label = ['3 Star', '4 Star', '5 Star', '6 Star']
|
|
explode = [0.01, 0.01, 0.01, 0.01]
|
|
data = [values for _key, values in data_dict['by_star'].items()]
|
|
plt.pie(data, explode=explode, labels=label, autopct='%1.2f%%')
|
|
plt.title(title)
|
|
# plt.savefig('test')
|
|
plt.show()
|
|
|
|
def get_payment_data(self):
|
|
url = 'https://as.hypergryph.com/u8/pay/v1/recent'
|
|
data = {"appId": 1, "channelMasterId": 1, "channelToken": {"token": self.token}}
|
|
pay_data = json.loads(requests.post(url, json.dumps(data), proxies=proxies).text).get('data')
|
|
for item in pay_data:
|
|
i = pay_data.index(item)
|
|
pay_time = item['payTime']
|
|
pay_data[i]['payTime'] = time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(pay_time))
|
|
|
|
pay_dict = {item['orderId']: item for item in pay_data}
|
|
self.payment_data.update(pay_dict)
|
|
file = open(self.payment_path, 'w', encoding='utf-8')
|
|
file.write(json.dumps(self.payment_data))
|
|
file.close()
|
|
|
|
def get_all_banner_pull_data(self):
|
|
plt.rcParams['font.sans-serif'] = 'SimHei' # 设置中文显示
|
|
plt.figure(figsize=(6, 6))
|
|
banner_title = [item for item in self.data_dict_by_banner]
|
|
banner_data_6 = [value['analysis']['by_star']['6'] for _key, value in self.data_dict_by_banner.items()]
|
|
banner_data_5 = [value['analysis']['by_star']['5'] for _key, value in self.data_dict_by_banner.items()]
|
|
banner_data_4 = [value['analysis']['by_star']['4'] for _key, value in self.data_dict_by_banner.items()]
|
|
banner_data_3 = [value['analysis']['by_star']['3'] for _key, value in self.data_dict_by_banner.items()]
|
|
|
|
y_axis = np.arange(len(banner_title))
|
|
width = 0.25
|
|
plt.barh(y_axis - width, banner_data_6, width/2, label='6 star')
|
|
plt.barh(y_axis - width/2, banner_data_5, width/2, label='5 star')
|
|
plt.barh(y_axis, banner_data_4, width/2, label='4 star')
|
|
plt.barh(y_axis + width/2, banner_data_3, width/2, label='3 star')
|
|
|
|
plt.yticks(y_axis, banner_title)
|
|
|
|
# 添加数值标签
|
|
off_set = -0.1
|
|
for i in y_axis:
|
|
plt.text(x=banner_data_6[i] + 1, y=i - width + off_set, s=banner_data_6[i], ha='center', fontsize=8, family='Calibri')
|
|
plt.text(x=banner_data_5[i] + 1, y=i - width/2 + off_set, s=banner_data_5[i], ha='center', fontsize=8, family='Calibri')
|
|
plt.text(x=banner_data_4[i] + 1, y=i + off_set, s=banner_data_4[i], ha='center', fontsize=8, family='Calibri')
|
|
plt.text(x=banner_data_3[i] + 1, y=i + width/2 + off_set, s=banner_data_3[i], ha='center', fontsize=8, family='Calibri')
|
|
|
|
plt.legend()
|
|
plt.show()
|
|
return banner_title, banner_data_6, banner_data_5, banner_data_4, banner_data_3
|
|
|
|
|
|
def cal_gacha_probability(pull):
|
|
if pull <= 50:
|
|
return 0.98 ** pull
|
|
probability = cal_gacha_probability(50)
|
|
j = 0.04
|
|
for i in range(50, pull+1):
|
|
probability *= 1 - j
|
|
j += 0.02
|
|
return probability
|
|
|
|
|
|
def get_time_from_unix(data, if_string=False):
|
|
try:
|
|
if if_string:
|
|
return datetime.datetime.fromtimestamp(int(data)).strftime('%Y-%m-%d %H:%M:%S')
|
|
else:
|
|
return datetime.datetime.fromtimestamp(int(data))
|
|
except Exception as e:
|
|
print(e)
|
|
raise TypeError
|
|
|
|
|
|
if __name__ == '__main__':
|
|
ark = ArknightsInfo('data')
|
|
print('111')
|
|
# ark.get_all_banner_pull_data()
|
|
# ark.show_all_gacha_data()
|
|
# ark.gacha_data_visualization(ark.data_dict_by_banner['真理孑然']['analysis'], '真理孑然')
|
|
# print('done')
|
|
# print(cal_gacha_probability(79) * 10000)
|