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)