更新代码

.gitignore

@@ -1 +1,3 @@
 /config/user_info_local.json
+/.venv/
+/data/ArknightsStoryJson-main/

arknights_gacha_analysis.py

@@ -0,0 +1,230 @@
+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)

arknights_gacha_analysis_new.py

@@ -0,0 +1,137 @@
+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
+
+
+class ArknightsInfo:
+    def __init__(self, phone):
+        self.phone = phone
+        self.gacha_data = {}
+        self.payment_data = {}
+
+        self.server_url = 'http://192.168.195.194:59001'
+        self.admin_key = "1145141919810"
+        self.get_data()
+        self.data_dict_by_banner, self.all_result_for_show, self.raw_pull_data = self.convert_data()
+
+    def get_data(self):
+        url = f'{self.server_url}/GetData'
+        data_dict = requests.post(url, data=json.dumps({"admin_key": self.admin_key})).json()
+        self.gacha_data = data_dict.get('gacha_data', {}).get(self.phone, {})
+        self. payment_data = data_dict.get('payment_data', {}).get(self.phone, {})
+
+    def convert_data(self):
+        content = self.gacha_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
+
+    @staticmethod
+    def analyze_gacha_data(data_list):
+        result = {
+            'by_star': {'3': 0, '4': 0, '5': 0, '6': 0},
+            'new_operator': []
+        }
+        if len(data_list) == 0:
+            return result
+        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
+
+    def show_all_gacha_data(self):
+        self.gacha_data_visualization(self.all_result_for_show)
+
+    @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_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
+
+
+if __name__ == '__main__':
+    ark = ArknightsInfo('18021026530')
+    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)

gui_test.py

@@ -0,0 +1,294 @@
+import datetime
+import tkinter
+import copy
+from tkinter import *
+import numpy as np
+import matplotlib
+import matplotlib.pyplot as plt
+from matplotlib.pylab import mpl
+from matplotlib.backends.backend_tkagg import FigureCanvasTkAgg, NavigationToolbar2Tk
+import pandas as pd
+
+from arknights_gacha_analysis_new import ArknightsInfo
+
+
+class ArknightsInfoDisplay(Frame):
+    """一个经典的GUI写法"""
+    def __init__(self, ark_info, master=None):
+        """初始化方法"""
+        super().__init__(master)  # 调用父类的初始化方法
+        self.figure = None
+        self.ark_info = ark_info
+        self.master = master
+        self.pack(side=TOP, fill=BOTH, expand=1)  # 此处填充父窗体
+        self.create_matplotlib()
+        self.createWidget(self.figure)
+
+        # 添加图1区域按钮
+        banner_title = [item for item in self.ark_info.data_dict_by_banner]
+        y_axis, x_axis = 120, 10
+        for item in banner_title:
+            tkinter.Button(self.master, text=item, command=lambda item=item: self.change_fig1_content(item),
+                           width=10).place(x=x_axis, y=10 + y_axis)
+            y_axis += 30
+            if y_axis >= 500:
+                y_axis = 120
+                x_axis += 80
+
+        # 添加图2区域按钮
+        y_axis, x_axis = 120, 1750
+        tkinter.Button(self.master, text='Separate by Star', command=lambda: self.change_fig2_content('separated'),
+                       width=20).place(x=x_axis, y=10 + y_axis)
+        tkinter.Button(self.master, text='Total Pull Count', command=lambda: self.change_fig2_content('sum'),
+                       width=20).place(x=x_axis, y=40 + y_axis)
+        tkinter.Button(self.master, text='Stacked Pull Count', command=lambda: self.change_fig2_content('stacked'),
+                       width=20).place(x=x_axis, y=70 + y_axis)
+        tkinter.Button(self.master, text='Probability Separated',
+                       command=lambda: self.change_fig2_content('separated', True),
+                       width=20).place(x=x_axis, y=100 + y_axis)
+        tkinter.Button(self.master, text='Probability Six Star',
+                       command=lambda: self.change_fig2_content('six_star_only', True),
+                       width=20).place(x=x_axis, y=130 + y_axis)
+
+    def change_fig1_content(self, banner_name):
+        # print(banner_name)
+        self.fig1.clear()
+        self.fig3.clear()
+        pull_count = self.ark_info.data_dict_by_banner[banner_name]['analysis']['total_pull']
+        self.gacha_data_visualization(self.fig1, self.ark_info.data_dict_by_banner[banner_name]['analysis'],
+                                      title=f'{banner_name}({pull_count} pulls)')
+
+        self.pull_time_visualization(self.fig3, banner=banner_name, mode='T')
+
+        self.canvas.draw()
+
+    def change_fig2_content(self, mode, probability_flag=False):
+        self.fig2.clear()
+        if not probability_flag:
+            self.show_all_banner_basic_info(self.fig2, mode)
+        else:
+            self.show_all_banner_probability(self.fig2, mode)
+        self.canvas.draw()
+
+    def createWidget(self, figure):
+        """创建组件"""
+        # self.label = Label(self, text='这是一个Tkinter和Matplotlib相结合的小例子')
+        # self.label.pack()
+        # 创建画布
+        self.canvas = FigureCanvasTkAgg(figure, self)
+        self.canvas.draw()
+        self.canvas.get_tk_widget().pack(side=TOP, fill=BOTH, expand=1)
+        # 把matplotlib绘制图形的导航工具栏显示到tkinter窗口上
+        # toolbar = NavigationToolbar2Tk(self.canvas, self)
+        # toolbar.update()
+        # self.canvas._tkcanvas.pack(side=TOP, fill=BOTH, expand=1)
+        # self.button = Button(master=self, text="退出", command=quit)
+        # # 按钮放在下边
+        # self.button.pack(side=BOTTOM)
+
+    def create_matplotlib(self):
+        """创建绘图对象"""
+        # 设置中文显示字体
+        mpl.rcParams['font.sans-serif'] = ['SimHei']  # 中文显示
+        mpl.rcParams['axes.unicode_minus'] = False  # 负号显示
+        # 创建绘图对象f figsize的单位是英寸 像素 = 英寸*分辨率na
+        self.figure = plt.figure(num=2, figsize=(7, 4), dpi=120, edgecolor='green', frameon=True)
+        # 一张图上显示4张小图
+        self.fig1 = plt.subplot(2, 2, 1)  # 先进行分块,最后一个参数是序号
+        self.gacha_data_visualization(self.fig1, self.ark_info.all_result_for_show,
+                                      f'All Banner Result({self.ark_info.all_result_for_show["total_pull"]} pulls)')
+        self.fig2 = plt.subplot(2, 2, 2)
+        self.show_all_banner_basic_info(self.fig2, 'separated')
+        self.fig3 = plt.subplot(2, 2, 3)
+
+        self.ax3 = plt.gca()
+        self.pull_time_visualization(self.fig3)
+        # self.show_all_banner_probability(self.fig3)
+
+        self.fig4 = plt.subplot(2, 2, 4)
+        # self.setplot(fig4, x, y4, 'y=square(x)', 'gold')
+        x = np.arange(12)
+        y = np.random.uniform(0.5, 1.0, 12) * (1 - x / float(12))
+        loc = zip(x, y)  # 将x, y 两两配对
+        plt.ylim(0, 1.2)  # 设置y轴的范围
+        plt.bar(x, y, facecolor='green', edgecolor='black')  # 绘制柱状图(填充颜色绿色，边框黑色)
+        for x, y in loc:
+            plt.text(x + 0.1, y + 0.01, '%.2f' % y, ha='center', va='bottom')
+
+    def setplot(self, fig, x, y, text, color='r'):
+        """绘制子图"""
+        line = fig.plot(x, y, color=color, label=text)
+        fig.set_xlabel('(x)横坐标')  # 确定坐标轴标题
+        fig.set_ylabel("(y)纵坐标")
+        fig.grid(which='major', axis='x', color='gray', linestyle='-', linewidth=0.5, alpha=0.2)  # 设置网格
+        fig.legend(loc='lower right', facecolor='orange', frameon=True, shadow=True, framealpha=0.7)
+
+    def gacha_data_visualization(self, fig, data_dict, title='All Banner Result'):
+        fig.set_xlabel(title)
+        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()]
+        for i in range(len(data)):
+            label[i] += f": {data[i]}"
+        fig.pie(data, explode=explode, labels=label, autopct='%1.2f%%')
+
+    def pull_time_visualization(self, fig, banner='全部结果', mode='M'):
+        mode_list = ['Y', 'M', 'D', 'H', 'T', 'S']
+        if banner == '全部结果':
+            mode = 'M'
+        raw_time_dict = self.ark_info.data_dict_by_banner[banner]['gacha_result_by_time']
+
+        start_time = (list(raw_time_dict.values())[0][1] - datetime.timedelta(seconds=60)).strftime('%Y-%m-%d %H:%M:%S')
+        end_time = (list(raw_time_dict.values())[-1][1] + datetime.timedelta(seconds=60)).strftime('%Y-%m-%d %H:%M:%S')
+        new_time_dict = {key: len(item[0]) for key, item in raw_time_dict.items()}
+        new_time_dict[start_time] = 0
+        new_time_dict[end_time] = 0
+        s = copy.deepcopy(pd.Series(new_time_dict))
+        s.index = pd.to_datetime(s.index)
+
+        # 重采样并求和
+        resampled = s.resample(mode).sum()
+        while len(resampled) > 10000:
+            try:
+                resampled = s.resample(mode_list[mode_list.index(mode)-1]).sum()
+            except:
+                break
+        resampled.plot(style='-', label=banner, ax=self.ax3)
+        fig.set_xlabel(banner)
+
+        # time_series = [item[1] for item in raw_time_dict.values()]
+        # pull_count_list = [len(item[0]) for item in raw_time_dict.values()]
+        #
+        # fig.plot(time_series, pull_count_list, label=banner)
+
+        # start_time = self.time_filter(list(raw_time_dict.values())[0][1], mode).strftime('%Y-%m-%d %H:%M:%S')
+        # end_time = self.time_filter(list(raw_time_dict.values())[-1][1], mode).strftime('%Y-%m-%d %H:%M:%S')
+        # time_stamp = pd.date_range(start=start_time, end=end_time, freq=mode_dict[mode], normalize=False)
+        # time_dict = {str(item): 0 for item in time_stamp}
+        # for item in raw_time_dict.values():
+        #     new_time = self.time_filter(item[1], mode)
+        #     if new_time.strftime('%Y-%m-%d %H:%M:%S') not in time_dict:
+        #         time_dict[new_time.strftime('%Y-%m-%d %H:%M:%S')] = len(item[0])
+        #     else:
+        #         time_dict[new_time.strftime('%Y-%m-%d %H:%M:%S')] += len(item[0])
+        #
+        # fig.plot(time_dict.keys(), time_dict.values(), label=banner)
+        #
+        # tick_spacing = int(len(time_dict) / 10) + 1
+        # ax3.xaxis.set_major_locator(matplotlib.ticker.MultipleLocator(tick_spacing))
+        # ax3.tick_params(axis='x', labelrotation=45, labelsize=8)
+        fig.legend()
+
+    @staticmethod
+    def time_filter(target_time, mode=4):
+        filter_list = [target_time.year, target_time.month, target_time.day, target_time.hour,
+                       target_time.minute, target_time.second]
+        for i in range(len(filter_list)):
+            if i >= mode:
+                filter_list[i] = 0
+        return datetime.datetime(filter_list[0], filter_list[1],
+                                 filter_list[2], filter_list[3], filter_list[4], filter_list[5])
+
+    def show_all_banner_basic_info(self, fig, mode='separated'):
+        tmp_dict = self.ark_info.data_dict_by_banner
+        banner_title = [item for item in tmp_dict][0:-1:1]
+        banner_data_3 = [value['analysis']['by_star']['3'] for _key, value in tmp_dict.items()][0:-1:1]
+        banner_data_4 = [value['analysis']['by_star']['4'] for _key, value in tmp_dict.items()][0:-1:1]
+        banner_data_5 = [value['analysis']['by_star']['5'] for _key, value in tmp_dict.items()][0:-1:1]
+        banner_data_6 = [value['analysis']['by_star']['6'] for _key, value in tmp_dict.items()][0:-1:1]
+        banner_data_sum = [value['analysis']['total_pull'] for _key, value in tmp_dict.items()][0:-1:1]
+
+        if mode == 'separated':
+            self.show_separated_data(fig, banner_title, banner_data_3, banner_data_4, banner_data_5, banner_data_6)
+        elif mode == 'sum':
+            self.show_sum_data(fig, banner_title, banner_data_sum, 'Total Pulls')
+        elif mode == 'stacked':
+            self.show_stacked_data(fig, banner_title, banner_data_3, banner_data_4, banner_data_5, banner_data_6)
+
+    def show_all_banner_probability(self, fig, mode='separated'):
+        tmp_dict = self.ark_info.data_dict_by_banner
+        banner_title = [item for item in tmp_dict]
+        banner_data_3 = [value['analysis']['three_star_percent'] for _key, value in tmp_dict.items()]
+        banner_data_4 = [value['analysis']['four_star_percent'] for _key, value in tmp_dict.items()]
+        banner_data_5 = [value['analysis']['five_star_percent'] for _key, value in tmp_dict.items()]
+        banner_data_6 = [value['analysis']['six_star_percent'] for _key, value in tmp_dict.items()]
+        if mode == 'separated':
+            self.show_separated_data(fig, banner_title, banner_data_3, banner_data_4, banner_data_5, banner_data_6)
+        elif mode == 'six_star_only':
+            self.show_sum_data(fig, banner_title, banner_data_6, 'Six Star Probability')
+
+    def show_separated_data(self, fig, banner_title, banner_data_3, banner_data_4, banner_data_5, banner_data_6):
+        y_axis = np.arange(len(banner_title))
+        width = 0.25
+        fig.barh(y_axis + width / 2, banner_data_3, width / 2, label='3 star')
+        fig.barh(y_axis, banner_data_4, width / 2, label='4 star')
+        fig.barh(y_axis - width / 2, banner_data_5, width / 2, label='5 star')
+        fig.barh(y_axis - width, banner_data_6, width / 2, label='6 star')
+        fig.set_yticks(y_axis, banner_title)
+
+        # 添加数值标签
+        for i in y_axis:
+            fig.text(x=banner_data_6[i] + 1, y=i - width, s=round(banner_data_6[i], 2),
+                     ha='center', fontsize=8, family='Calibri', va='center')
+            fig.text(x=banner_data_5[i] + 1, y=i - width / 2, s=round(banner_data_5[i], 2),
+                     ha='center', fontsize=8, family='Calibri', va='center')
+            fig.text(x=banner_data_4[i] + 1, y=i, s=round(banner_data_4[i], 2),
+                     ha='center', fontsize=8, family='Calibri', va='center')
+            fig.text(x=banner_data_3[i] + 1, y=i + width / 2, s=round(banner_data_3[i], 2),
+                     ha='center', fontsize=8, family='Calibri', va='center')
+        fig.legend()
+
+    def show_stacked_data(self, fig, banner_title, banner_data_3, banner_data_4, banner_data_5, banner_data_6):
+        y_axis = np.arange(len(banner_title))
+        width = 0.8
+        fig.barh(y_axis, banner_data_3, width, label='3 star')
+        fig.barh(y_axis, banner_data_4, width, left=banner_data_3, label='4 star')
+        fig.barh(y_axis, banner_data_5, width,
+                 left=[banner_data_3[i] + banner_data_4[i] for i in range(len(banner_data_4))], label='5 star')
+        fig.barh(y_axis, banner_data_6, width, label='6 star',
+                 left=[banner_data_3[i] + banner_data_4[i] + banner_data_5[i] for i in range(len(banner_data_4))])
+        fig.set_yticks(y_axis, banner_title)
+
+        # 添加数值标签
+        # off_set = -0.1
+        # for i in y_axis:
+        #     fig.text(x=banner_data_6[i] + 1, y=i - width + off_set, s=round(banner_data_6[i], 2),
+        #              ha='center', fontsize=8, family='Calibri')
+        #     fig.text(x=banner_data_5[i] + 1, y=i - width / 2 + off_set, s=round(banner_data_5[i], 2),
+        #              ha='center', fontsize=8, family='Calibri')
+        #     fig.text(x=banner_data_4[i] + 1, y=i + off_set, s=round(banner_data_4[i], 2),
+        #              ha='center', fontsize=8, family='Calibri')
+        #     fig.text(x=banner_data_3[i] + 1, y=i + width / 2 + off_set, s=round(banner_data_3[i], 2),
+        #              ha='center', fontsize=8, family='Calibri')
+        fig.legend()
+
+    def show_sum_data(self, fig, banner_title, sum_data, label='Total Pulls'):
+        y_axis = np.arange(len(banner_title))
+        width = 0.8
+        fig.barh(y_axis, sum_data, width, label=label)
+        fig.set_yticks(y_axis, banner_title)
+        # 添加数值标签
+        for i in y_axis:
+            fig.text(x=sum_data[i] / 2, y=i, s=round(sum_data[i], 2),
+                     ha='center', va='center', fontsize=8, family='Calibri', color='w')
+        fig.legend()
+
+    def destroy(self):
+        """重写destroy方法"""
+        super().destroy()
+        quit()
+
+    def quit(self):
+        """点击退出按钮时调用这个函数"""
+        root.quit()  # 结束主循环
+        root.destroy()  # 销毁窗口
+
+
+if __name__ == '__main__':
+    root = Tk()
+    root.title('Arknights Info Panel')
+    root.geometry('1920x1200')
+    # app = ArknightsInfoDisplay(ArknightsInfo('18021026530'), master=root)
+    app = ArknightsInfoDisplay(ArknightsInfo('13693680360'), master=root)
+
+    root.mainloop()

run_service_backend.py

@@ -34,12 +34,12 @@ def start_data_retrieve():
         return 'No data supplied'
     if input_param_dict.get('admin_key') != admin_key:
         return '403 Forbidden'
-    if ark_back.is_retrieving_data():
+    if ark_back.is_retrieving_data:
         return json.dumps({'status': 1, 'data': 'already started'})
     else:
         ark_back.is_retrieving_data = True
         s1 = threading.Thread(target=data_retrieve_thread)
-        s2 = threading.Thread(target=skland_sign_in_thread())
+        s2 = threading.Thread(target=skland_sign_in_thread)
         s1.start()
         s2.start()
     return json.dumps({'status': 0, 'data': ark_back.is_retrieving_data})
@@ -55,7 +55,9 @@ def change_status():
     if input_param_dict.get('status') == 'on going' and not ark_back.is_retrieving_data:
         ark_back.is_retrieving_data = True
         s1 = threading.Thread(target=data_retrieve_thread)
+        s2 = threading.Thread(target=skland_sign_in_thread)
         s1.start()
+        s2.start()
         return json.dumps({'status': 0, 'data': ark_back.is_retrieving_data})
     elif input_param_dict.get('status') == 'breaking' and ark_back.is_retrieving_data:
         ark_back.is_retrieving_data = False
@@ -112,7 +114,7 @@ def skland_sign_in_thread():
     try:
         last_run = None
         next_run_time = random_time(480, 720)
-        ark_back.logger.info(f'Next skland sign-in time: {next_run_time}')
+        ark_back.logger.info(f'Next daily sign-in time: {next_run_time}')
         while 1:
             if not ark_back.is_retrieving_data:
                 break
@@ -120,22 +122,31 @@ def skland_sign_in_thread():
             now = datetime.datetime.now()
             if now.strftime('%H:%M') == next_run_time and last_run != now.date():
                 next_run_time = random_time(480, 720)
-                ark_back.skland_sign_new()
+                ark_back.daily_sign_new()
                 last_run = now.date()
     except Exception as e:
         print(e, traceback.format_exc())
         ark_back.logger.error(traceback.format_exc())
-        new_send_email(target_email_list, 'Error in ArkHelper Skland Sign-in', str(traceback.format_exc()))
+        new_send_email(target_email_list, 'Error in ArkHelper Daily Sign-in', str(traceback.format_exc()))
     finally:
         ark_back.logger.info('Data Retrieve Stopped')
 
+logger = LoggerClass().get_log(logging.INFO)
+ark_back = ArkHelperBackend(logger=logger)
+# ark_back.update_gacha_data()
+# ark_back.update_payment_data()
+threading.Thread(target=skland_sign_in_thread).start()
+threading.Thread(target=data_retrieve_thread).start()
+
 
 if __name__ == '__main__':
     print('start')
-    logger = LoggerClass().get_log(logging.INFO)
+    ark_back.daily_sign_new()
-    ark_back = ArkHelperBackend(logger=logger)
+    # logger = LoggerClass().get_log(logging.INFO)
-    ark_back.update_payment_data()
+    # ark_back = ArkHelperBackend(logger=logger)
-    threading.Thread(target=skland_sign_in_thread).start()
+    # ark_back.update_gacha_data()
-    threading.Thread(target=data_retrieve_thread).start()
+    # ark_back.update_payment_data()
+    # threading.Thread(target=skland_sign_in_thread).start()
+    # threading.Thread(target=data_retrieve_thread).start()
     app.run(host="192.168.195.194", port=59001)
 

src/SKland_Auto_Sign/new_auto_sign.py

@@ -19,6 +19,12 @@ token_env = os.environ.get('TOKEN')
 # 现在想做什么？
 current_type = os.environ.get('SKYLAND_TYPE')
 
+ARKNIGHTS_APP_CODE = 'arknights'
+ENDFIELD_APP_CODE = 'endfield'
+ARKNIGHTS_GAME_ID = 1
+ENDFIELD_SIGN_PLATFORM = '3'
+ENDFIELD_SIGN_VNAME = '1.0.0'
+
 http_local = threading.local()
 header = {
     'cred': '',
@@ -44,6 +50,7 @@ header_for_sign = {
 
 # 签到url
 sign_url = "https://zonai.skland.com/api/v1/game/attendance"
+endfield_sign_url = "https://zonai.skland.com/api/v1/game/endfield/attendance"
 # 绑定的角色url
 binding_url = "https://zonai.skland.com/api/v1/game/player/binding"
 # 验证码url
@@ -57,6 +64,13 @@ grant_code_url = "https://as.hypergryph.com/user/oauth2/v2/grant"
 # 使用认证代码获得cred
 cred_code_url = "https://zonai.skland.com/web/v1/user/auth/generate_cred_by_code"
 
+header_for_endfield_sign = {
+    'platform': ENDFIELD_SIGN_PLATFORM,
+    'timestamp': '',
+    'dId': '',
+    'vName': ENDFIELD_SIGN_VNAME
+}
+
 
 def config_logger():
     current_date = date.today().strftime('%Y-%m-%d')
@@ -103,7 +117,7 @@ def config_logger():
     requests.post = post
 
 
-def generate_signature(token: str, path, body_or_query):
+def generate_signature(token: str, path, body_or_query, sign_header_template=None):
     """
     获得签名头
     接口地址+方法为Get请求？用query否则用body+时间戳+ 请求头的四个重要参数（dId，platform，timestamp，vName）.toJSON()
@@ -117,7 +131,7 @@ def generate_signature(token: str, path, body_or_query):
     # 总是说请勿修改设备时间，怕不是yj你的服务器有问题吧，所以这里特地-2
     t = str(int(time.time()) - 2)
     token = token.encode('utf-8')
-    header_ca = json.loads(json.dumps(header_for_sign))
+    header_ca = json.loads(json.dumps(sign_header_template or header_for_sign))
     header_ca['timestamp'] = t
     header_ca_str = json.dumps(header_ca, separators=(',', ':'))
     s = path + body_or_query + t + header_ca_str
@@ -204,7 +218,13 @@ def get_cred(grant):
     return resp['data']
 
 
-def get_binding_list():
+def setup_cred_session(cred_resp):
+    http_local.token = cred_resp['token']
+    http_local.header = header.copy()
+    http_local.header['cred'] = cred_resp['cred']
+
+
+def get_binding_list(target_app_code=ARKNIGHTS_APP_CODE):
     v = []
     resp = requests.get(binding_url, headers=get_sign_header(binding_url, 'get', None, http_local.header)).json()
 
@@ -215,67 +235,170 @@ def get_binding_list():
             os.remove(token_save_name)
             return []
     for i in resp['data']['list']:
-        if i.get('appCode') != 'arknights':
+        if i.get('appCode') != target_app_code:
             continue
-        v.extend(i.get('bindingList'))
+        v.extend(i.get('bindingList') or [])
     return v
 
 
+def is_duplicate_sign_message(message):
+    return '请勿重复签到' in (message or '') or 'Please do not sign in again!' in (message or '')
+
+
+def format_arknights_sign_message(character, resp):
+    role_name = character.get('nickName') or '未知角色'
+    channel_name = character.get('channelName') or '未知服务器'
+    if resp['code'] != 0:
+        error_message = resp.get('message') or '未知错误'
+        if is_duplicate_sign_message(error_message):
+            return True, f'角色{role_name}({channel_name})今日已签到，请勿重复签到'
+        return False, f'角色{role_name}({channel_name})签到失败了！原因：{error_message}'
+
+    awards = resp.get('data', {}).get('awards') or []
+    award_text = []
+    for award in awards:
+        resource = award.get('resource') or {}
+        resource_name = resource.get('name')
+        if not resource_name:
+            continue
+        award_text.append(f'{resource_name}×{award.get("count") or 1}')
+
+    if award_text:
+        return True, f'角色{role_name}({channel_name})签到成功，获得了{"、".join(award_text)}'
+    return True, f'角色{role_name}({channel_name})签到成功'
+
+
+def get_endfield_role(binding):
+    role = binding.get('defaultRole')
+    if role:
+        return role
+    roles = binding.get('roles') or []
+    return roles[0] if roles else None
+
+
+def get_endfield_sign_header(role_str):
+    parsed_url = parse.urlparse(endfield_sign_url)
+    sign, header_ca = generate_signature(http_local.token, parsed_url.path, '', header_for_endfield_sign)
+    sign_header = http_local.header.copy()
+    sign_header['sign'] = sign
+    sign_header['platform'] = header_ca['platform']
+    sign_header['timestamp'] = header_ca['timestamp']
+    sign_header['dId'] = header_ca['dId']
+    sign_header['vName'] = header_ca['vName']
+    sign_header['sk-game-role'] = role_str
+    sign_header['Content-Type'] = 'application/json'
+    return sign_header
+
+
+def format_endfield_awards(resp_data):
+    award_ids = resp_data.get('awardIds') or []
+    resource_map = resp_data.get('resourceInfoMap') or {}
+    award_text = []
+
+    for award in award_ids:
+        award_id = award.get('id')
+        if award_id is None:
+            continue
+        resource = resource_map.get(award_id) or resource_map.get(str(award_id))
+        if not resource:
+            continue
+        count = award.get('count') or resource.get('count') or 1
+        award_text.append(f'{resource.get("name", award_id)}×{count}')
+    return award_text
+
+
+def format_endfield_sign_message(binding, resp):
+    role = get_endfield_role(binding) or {}
+    role_name = role.get('nickname') or binding.get('nickName') or '未知角色'
+    channel_name = role.get('serverName') or binding.get('channelName') or '未知服务器'
+
+    if resp['code'] != 0:
+        error_message = resp.get('message') or '未知错误'
+        if is_duplicate_sign_message(error_message):
+            return True, f'终末地角色{role_name}({channel_name})今日已签到，请勿重复签到'
+        return False, f'终末地角色{role_name}({channel_name})签到失败了！原因：{error_message}'
+
+    award_text = format_endfield_awards(resp.get('data', {}))
+    if award_text:
+        return True, f'终末地角色{role_name}({channel_name})签到成功，获得了{"、".join(award_text)}'
+    return True, f'终末地角色{role_name}({channel_name})签到成功'
+
+
 def list_awards(game_id, uid):
     resp = requests.get(sign_url, headers=http_local.header, params={'gameId': game_id, 'uid': uid}).json()
     print(resp)
 
 
 def do_sign(cred_resp):
-    http_local.token = cred_resp['token']
+    setup_cred_session(cred_resp)
-    http_local.header = header.copy()
-    http_local.header['cred'] = cred_resp['cred']
     characters = get_binding_list()
 
     for i in characters:
         body = {
-            'gameId': 1,
+            'gameId': ARKNIGHTS_GAME_ID,
             'uid': i.get('uid')
         }
         # list_awards(1, i.get('uid'))
         resp = requests.post(sign_url, headers=get_sign_header(sign_url, 'post', body, http_local.header),
                              json=body).json()
-        if resp['code'] != 0:
+        _, msg = format_arknights_sign_message(i, resp)
-            print(f'角色{i.get("nickName")}({i.get("channelName")})签到失败了！原因：{resp.get("message")}')
+        print(msg)
-            continue
-        awards = resp['data']['awards']
-        for j in awards:
-            res = j['resource']
-            print(
-                f'角色{i.get("nickName")}({i.get("channelName")})签到成功，获得了{res["name"]}×{j.get("count") or 1}'
-            )
 
 
 def single_sign(cred_resp):
-    http_local.token = cred_resp['token']
+    setup_cred_session(cred_resp)
-    http_local.header = header.copy()
-    http_local.header['cred'] = cred_resp['cred']
     characters = get_binding_list()
 
-    code, msg = -1, 'Default Error'
+    if not characters:
+        return '未绑定明日方舟角色，已跳过', 0
+
+    code = 0
+    messages = []
     for i in characters:
         body = {
-            'gameId': 1,
+            'gameId': ARKNIGHTS_GAME_ID,
             'uid': i.get('uid')
         }
         # list_awards(1, i.get('uid'))
         resp = requests.post(sign_url, headers=get_sign_header(sign_url, 'post', body, http_local.header),
                              json=body).json()
-        if resp['code'] != 0:
+        success, msg = format_arknights_sign_message(i, resp)
-            msg = f'角色{i.get("nickName")}({i.get("channelName")})签到失败了！原因：{resp.get("message")}'
+        messages.append(msg)
+        if not success:
+            code = -1
+    return '\n'.join(messages), code
+
+
+def single_endfield_sign(cred_resp):
+    setup_cred_session(cred_resp)
+    bindings = get_binding_list(ENDFIELD_APP_CODE)
+
+    if not bindings:
+        return '未绑定终末地角色，已跳过', 0
+
+    code = 0
+    messages = []
+    for binding in bindings:
+        role = get_endfield_role(binding)
+        if not role:
+            messages.append('终末地角色信息缺失，已跳过')
             code = -1
             continue
-        awards = resp['data']['awards']
+
-        for j in awards:
+        role_id = role.get('roleId')
-            res = j['resource']
+        server_id = role.get('serverId')
-            msg = f'角色{i.get("nickName")}({i.get("channelName")})签到成功，获得了{res["name"]}×{j.get("count") or 1}'
+        if role_id is None or server_id is None:
-            code = 0
+            messages.append('终末地角色信息不完整，已跳过')
-    return msg, code
+            code = -1
+            continue
+
+        role_str = f'{ENDFIELD_SIGN_PLATFORM}_{role_id}_{server_id}'
+        resp = requests.post(endfield_sign_url, headers=get_endfield_sign_header(role_str), json=None).json()
+        success, msg = format_endfield_sign_message(binding, resp)
+        messages.append(msg)
+        if not success:
+            code = -1
+    return '\n'.join(messages), code
 
 
 def save(token):

src/ark_helper_backend/backend.py

@@ -10,7 +10,7 @@ from pathlib import Path
 
 from src.util.utils import hg_get_new_user_token, load_user_config, validate_token
 from src.My_Logger.project_logger import LoggerClass
-from src.SKland_Auto_Sign.new_auto_sign import get_cred_by_token, single_sign
+from src.SKland_Auto_Sign.new_auto_sign import get_cred_by_token, single_endfield_sign, single_sign
 
 PROJECT_PATH = Path(__file__).parent.parent.parent.absolute()
 DATA_PATH = os.path.join(PROJECT_PATH, 'data', 'ArkHelperData')
@@ -137,17 +137,28 @@ class ArkHelperBackend:
             self.single_user_login(phone)
         self.save_data(payment_status=False, gacha_status=False)
 
-    def skland_sign_new(self):
+    def _run_sign_jobs(self, sign_jobs):
         for user, content in self.user_data_dict.items():
             token = content.get('token')
             if token:
                 try:
-                    msg, code = single_sign(get_cred_by_token(token))
+                    cred_resp = get_cred_by_token(token)
-                    self.logger.info(f'{msg}: {code}')
+                    for sign_name, sign_handler in sign_jobs:
+                        msg, code = sign_handler(cred_resp)
+                        self.logger.info(f'{user} {sign_name}签到结果: {msg} [{code}]')
                 except Exception as ex:
                     self.logger.error(f'签到失败，原因：{str(ex)}')
             time.sleep(random.randint(100, 200))
 
+    def skland_sign_new(self):
+        self._run_sign_jobs((('明日方舟', single_sign),))
+
+    def endfield_sign_new(self):
+        self._run_sign_jobs((('终末地', single_endfield_sign),))
+
+    def daily_sign_new(self):
+        self._run_sign_jobs((('明日方舟', single_sign), ('终末地', single_endfield_sign)))
+
     def update_gacha_data(self):
         for phone in self.user_data_dict:
             gacha_history_data = self.user_data_dict[phone]['gacha']
@@ -197,7 +208,7 @@ class ArkHelperBackend:
     def data_retrieve(self):
         self.refresh_user_token()
         # self.update_payment_data()
-        self.update_gacha_data()
+        # self.update_gacha_data()
 
     def show_data(self):
         gacha_data = {user: content['gacha'] for user, content in self.user_data_dict.items()}
@@ -206,7 +217,18 @@ class ArkHelperBackend:
 
 
 if __name__ == '__main__':
-    ak_back = ArkHelperBackend()
+    # ak_back = ArkHelperBackend()
-    ak_back.update_gacha_data()
+    # ak_back.update_gacha_data()
     # ak_back.update_payment_data()
-    ak_back.skland_sign_new()
+    # ak_back.daily_sign_new()
+
+    login_url = 'https://ak.hypergryph.com/user/login'
+    session = requests.session()
+    session.get(login_url)
+    login_data = json.dumps({"phone": '18021026530', "password": 'Mrfz1790990971'})
+    response = session.post('https://as.hypergryph.com/user/auth/v1/token_by_phone_password', data=login_data,
+                            headers={"Content-Type": "application/json"})
+    token_data = json.loads(response.text)['data']['token']
+    save_cookie = response.cookies
+
+    print('done')

start_server.py

@@ -0,0 +1,66 @@
+import subprocess
+import time
+import socket
+import os
+import logging
+
+from datetime import datetime
+from pathlib import Path
+
+from src.My_Logger.project_logger import LoggerClass
+
+LOG_PATH = Path(__file__).parent / 'logs'
+
+
+def is_port_in_use(port: int) -> bool:
+    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+        return s.connect_ex(('192.168.195.194', port)) == 0
+
+
+def start_server():
+    """
+    launch a micro service using gunicorn
+    :return:
+    """
+    launch_file_name = 'run_service_backend'    # file name of the entrance point
+    pid_path = LOG_PATH / 'pid.txt'
+
+    IP = '192.168.195.194'
+    PORT = 59001
+    WORKER = 1
+    TIME_OUT = 60
+    THREADS = 5
+    logger = LoggerClass().get_log(logging.INFO)
+
+    if is_port_in_use(PORT):
+        logger.error(f'Port {PORT} already in use')
+        return
+
+    now = datetime.now()
+
+    cmd = [
+        "nohup",
+        "/home/armor/anaconda3/envs/py311/bin/gunicorn",
+        "--threads", str(THREADS),
+        "--workers", str(WORKER),
+        "--bind", f"{IP}:{PORT}", f"{launch_file_name}:app",
+        "--timeout", str(TIME_OUT),
+        "--pid", pid_path
+    ]
+    stdout_file = LOG_PATH / f"console_{now.year}-{now.month}-{now.day}.log"
+    pid = subprocess.Popen(cmd, stdout=stdout_file.open('a'), stderr=subprocess.STDOUT, close_fds=True, start_new_session=True).pid
+    time.sleep(5)
+
+    if not is_port_in_use(PORT) or not os.path.exists(pid_path) or str(pid) != pid_path.read_text().strip():
+        logger.error(f'Failed to start server, check {stdout_file} and {pid_path}')
+        print(f">> tail {stdout_file}")
+        subprocess.run(["tail", str(stdout_file)])
+        return
+
+    info_str = f"Server started"
+    logger.info(info_str)
+    logger.info(f"server running at {IP}:{PORT} with pid {pid}")
+
+
+if __name__ == "__main__":
+    start_server()

stop_server.py

@@ -0,0 +1,27 @@
+import os
+import psutil
+import signal
+
+
+def stop_server(port):
+    # 遍历当前所有运行的进程
+    for proc in psutil.process_iter(attrs=['pid', 'name']):
+        try:
+            # 查找进程监听的所有连接
+            for conn in proc.connections(kind='inet'):
+                if conn.laddr.port == port:
+                    print(f"Killing process {proc.info['name']} (PID: {proc.info['pid']}) on port {port}")
+                    os.kill(proc.info['pid'], signal.SIGTERM)
+                    return True
+        except (psutil.NoSuchProcess, psutil.AccessDenied):
+            continue
+    return False
+
+
+
+if __name__ == '__main__':
+    PORT = 59001
+    if stop_server(PORT):
+        print(f"Process on port {PORT} has been terminated.")
+    else:
+        print(f"No process found on port {PORT}.")

storyline/story_json_process.py

@@ -0,0 +1,78 @@
+import json
+import os
+import re
+
+from pathlib import Path
+
+project_path = Path(__file__).parent.parent.absolute()
+data_path = os.path.join(project_path, 'data')
+
+
+def process_single_json(json_path):
+    with open(json_path, 'r', encoding='utf-8') as file:
+        json_content = json.load(file)
+
+    if 'eventid' not in json_content:
+        return {}
+
+    result_dict = {
+        "event_id": json_content['eventid'],
+        "language": json_content['lang'],
+        "event_name": json_content['eventName'],
+        "entry_type": json_content['entryType'],
+        "story_code": json_content['storyCode'],
+        "avg_tag": json_content['avgTag'],
+        "story_name": json_content['storyName'],
+        "story_introduction": json_content['storyInfo'],
+    }
+    story_content = []
+    for item in json_content['storyList']:
+
+        if item.get('prop') == 'Sticker' and item.get('attributes', {}).get('text'):
+            text = item.get('attributes', {}).get('text', '').replace('<i>', '').replace('</i>', '')
+            if text:
+                story_content.append({'role': 'voiceover', 'content': text})
+        elif item.get('attributes', {}).get('content'):
+            if item.get('attributes', {}).get('name'):
+                story_content.append({'role': item['attributes']['name'], 'content': item['attributes']['content']})
+            else:
+                story_content.append({'role': 'voiceover', 'content': item['attributes']['content']})
+    result_dict['story_content'] = story_content
+    return result_dict
+
+
+def dump_data_to_json():
+    data_pool = {}
+    base_path = os.path.join(data_path, 'ArknightsStoryJson-main', 'zh_CN', 'gamedata')
+    for root, dirs, files in os.walk(base_path):
+        for file in files:
+            temp_dict = process_single_json(os.path.join(root, file))
+            if not temp_dict:
+                continue
+            if temp_dict['event_name'] not in data_pool:
+                data_pool[temp_dict['event_name']] = [temp_dict]
+            else:
+                data_pool[temp_dict['event_name']].append(temp_dict)
+
+    with open('data_pool.json', 'w', encoding='utf-8') as file:
+        file.write(json.dumps(data_pool, ensure_ascii=False))
+
+
+if __name__ == '__main__':
+    with open('data_pool.json', 'r', encoding='utf-8') as file:
+        content = json.load(file)
+
+    target_dict = {'活动': [], '主线': [], "其它": [], '干员密录': []}
+    for key, item in content.items():
+        if item[0]['entry_type'] in ('ACTIVITY', 'MINI_ACTIVITY'):
+            target_dict['活动'].append(item)
+        elif item[0]['entry_type'] == 'MAINLINE':
+            target_dict['主线'].append(item)
+        elif item[0]['entry_type'] == 'EXTRA':
+            target_dict['其它'].append(item)
+        elif item[0]['entry_type'] == 'NONE':
+            target_dict['干员密录'].append(item)
+        else:
+            raise Exception('分类异常')
+
+    print('done')