更新代码

.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

@@ -0,0 +1,152 @@
+import traceback
+import threading
+import logging
+import json
+import time
+import datetime
+
+from flask import Flask, request
+from flask_cors import CORS
+
+from src.My_Logger.project_logger import LoggerClass
+from src.ark_helper_backend.backend import ArkHelperBackend
+from src.util.mail import send_batch_email, new_send_email
+from src.util.utils import random_time
+
+app = Flask(__name__)
+CORS(app, supports_credentials=True)
+admin_key = '1145141919810'
+target_email_list = ['ArmorServer@outlook.com']
+
+
+@app.route("/GetCurrentStatus", methods=["POST"])
+def get_current_status():
+    input_param_dict = json.loads(request.get_data()) if request.get_data() else {}
+    if not input_param_dict:
+        return 'No data supplied'
+    return json.dumps({'status': 0, 'data': ark_back.is_retrieving_data})
+
+
+@app.route("/StartDataRetrieve", methods=["POST"])
+def start_data_retrieve():
+    input_param_dict = json.loads(request.get_data()) if request.get_data() else {}
+    if not input_param_dict:
+        return 'No data supplied'
+    if input_param_dict.get('admin_key') != admin_key:
+        return '403 Forbidden'
+    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)
+        s1.start()
+        s2.start()
+    return json.dumps({'status': 0, 'data': ark_back.is_retrieving_data})
+
+
+@app.route("/ChangeStatus", methods=["POST"])
+def change_status():
+    input_param_dict = json.loads(request.get_data()) if request.get_data() else {}
+    if not input_param_dict:
+        return 'No data supplied'
+    if input_param_dict.get('admin_key') != admin_key:
+        return '403 Forbidden'
+    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
+        return json.dumps({'status': 0, 'data': ark_back.is_retrieving_data})
+    else:
+        return json.dumps({'status': -1})
+
+
+@app.route("/GetData", methods=["POST"])
+def get_data():
+    input_param_dict = json.loads(request.get_data()) if request.get_data() else {}
+    if not input_param_dict:
+        return 'No data supplied'
+    if input_param_dict.get('admin_key') != admin_key:
+        return '403 Forbidden'
+    gacha_data, payment_data = ark_back.show_data()
+    data = {'status': 0, 'gacha_data': gacha_data, 'payment_data': payment_data}
+    return json.dumps(data)
+
+
+@app.route("/RunGetDataNow", methods=["POST"])
+def run_get_data_now():
+    input_param_dict = json.loads(request.get_data()) if request.get_data() else {}
+    if not input_param_dict:
+        return 'No data supplied'
+    if input_param_dict.get('admin_key') != admin_key:
+        return '403 Forbidden'
+
+    ark_back.data_retrieve()
+    gacha_data, payment_data = ark_back.show_data()
+    data = {'status': 0, 'gacha_data': gacha_data, 'payment_data': payment_data}
+    return json.dumps(data)
+
+
+def data_retrieve_thread():
+    try:
+        while 1:
+            # time.sleep(3600)
+            if ark_back.is_retrieving_data:
+                ark_back.data_retrieve()
+            else:
+                ark_back.logger.info('Breaking')
+                break
+            time.sleep(3600)
+    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 data retrieval', str(traceback.format_exc()))
+    finally:
+        ark_back.logger.info('Data Retrieve Stopped')
+
+
+def skland_sign_in_thread():
+    try:
+        last_run = None
+        next_run_time = random_time(480, 720)
+        ark_back.logger.info(f'Next daily sign-in time: {next_run_time}')
+        while 1:
+            if not ark_back.is_retrieving_data:
+                break
+            time.sleep(20)
+            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.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 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')
+    ark_back.daily_sign_new()
+    # 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()
+    app.run(host="192.168.195.194", port=59001)
+

src/SKland_Auto_Sign/SecuritySm.py

@@ -0,0 +1,317 @@
+import base64
+import gzip
+import hashlib
+# 数美加密方法类
+import json
+import time
+import uuid
+
+import requests
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric import padding
+from cryptography.hazmat.primitives.ciphers.algorithms import AES
+from cryptography.hazmat.decrepit.ciphers.algorithms import TripleDES
+from cryptography.hazmat.primitives.ciphers.base import Cipher
+from cryptography.hazmat.primitives.ciphers.modes import CBC, ECB
+
+# 查询dId请求头
+devices_info_url = "https://fp-it.portal101.cn/deviceprofile/v4"
+
+# 数美配置
+SM_CONFIG = {
+    "organization": "UWXspnCCJN4sfYlNfqps",
+    "appId": "default",
+    "publicKey": "MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQCmxMNr7n8ZeT0tE1R9j/mPixoinPkeM+k4VGIn/s0k7N5rJAfnZ0eMER+QhwFvshzo0LNmeUkpR8uIlU/GEVr8mN28sKmwd2gpygqj0ePnBmOW4v0ZVwbSYK+izkhVFk2V/doLoMbWy6b+UnA8mkjvg0iYWRByfRsK2gdl7llqCwIDAQAB",
+    "protocol": "https",
+    "apiHost": "fp-it.portal101.cn"
+}
+
+PK = serialization.load_der_public_key(base64.b64decode(SM_CONFIG['publicKey']))
+
+DES_RULE = {
+    "appId": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "uy7mzc4h",
+        "obfuscated_name": "xx"
+    },
+    "box": {
+        "is_encrypt": 0,
+        "obfuscated_name": "jf"
+    },
+    "canvas": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "snrn887t",
+        "obfuscated_name": "yk"
+    },
+    "clientSize": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "cpmjjgsu",
+        "obfuscated_name": "zx"
+    },
+    "organization": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "78moqjfc",
+        "obfuscated_name": "dp"
+    },
+    "os": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "je6vk6t4",
+        "obfuscated_name": "pj"
+    },
+    "platform": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "pakxhcd2",
+        "obfuscated_name": "gm"
+    },
+    "plugins": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "v51m3pzl",
+        "obfuscated_name": "kq"
+    },
+    "pmf": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "2mdeslu3",
+        "obfuscated_name": "vw"
+    },
+    "protocol": {
+        "is_encrypt": 0,
+        "obfuscated_name": "protocol"
+    },
+    "referer": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "y7bmrjlc",
+        "obfuscated_name": "ab"
+    },
+    "res": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "whxqm2a7",
+        "obfuscated_name": "hf"
+    },
+    "rtype": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "x8o2h2bl",
+        "obfuscated_name": "lo"
+    },
+    "sdkver": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "9q3dcxp2",
+        "obfuscated_name": "sc"
+    },
+    "status": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "2jbrxxw4",
+        "obfuscated_name": "an"
+    },
+    "subVersion": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "eo3i2puh",
+        "obfuscated_name": "ns"
+    },
+    "svm": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "fzj3kaeh",
+        "obfuscated_name": "qr"
+    },
+    "time": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "q2t3odsk",
+        "obfuscated_name": "nb"
+    },
+    "timezone": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "1uv05lj5",
+        "obfuscated_name": "as"
+    },
+    "tn": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "x9nzj1bp",
+        "obfuscated_name": "py"
+    },
+    "trees": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "acfs0xo4",
+        "obfuscated_name": "pi"
+    },
+    "ua": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "k92crp1t",
+        "obfuscated_name": "bj"
+    },
+    "url": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "y95hjkoo",
+        "obfuscated_name": "cf"
+    },
+    "version": {
+        "is_encrypt": 0,
+        "obfuscated_name": "version"
+    },
+    "vpw": {
+        "cipher": "DES",
+        "is_encrypt": 1,
+        "key": "r9924ab5",
+        "obfuscated_name": "ca"
+    }
+}
+
+BROWSER_ENV = {
+    'plugins': 'MicrosoftEdgePDFPluginPortableDocumentFormatinternal-pdf-viewer1,MicrosoftEdgePDFViewermhjfbmdgcfjbbpaeojofohoefgiehjai1',
+    'ua': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36 Edg/129.0.0.0',
+    'canvas': '259ffe69',  # 基于浏览器的canvas获得的值，不知道复用行不行
+    'timezone': -480,  # 时区，应该是固定值吧
+    'platform': 'Win32',
+    'url': 'https://www.skland.com/',  # 固定值
+    'referer': '',
+    'res': '1920_1080_24_1.25',  # 屏幕宽度_高度_色深_window.devicePixelRatio
+    'clientSize': '0_0_1080_1920_1920_1080_1920_1080',
+    'status': '0011',  # 不知道在干啥
+}
+
+
+# // 将浏览器环境对象的key全部排序，然后对其所有的值及其子对象的值加入数字并字符串相加。若值为数字，则乘以10000(0x2710)再将其转成字符串存入数组,最后再做md5,存入tn变量（tn变量要做加密）
+# //把这个对象用加密规则进行加密，然后对结果做GZIP压缩（结果是对象，应该有序列化），最后做AES加密（加密细节目前不清除），密钥为变量priId
+# //加密规则：新对象的key使用相对应加解密规则的obfuscated_name值，value为字符串化后进行进行DES加密，再进行btoa加密
+
+# 通过测试
+def _DES(o: dict):
+    result = {}
+    for i in o.keys():
+        if i in DES_RULE.keys():
+            rule = DES_RULE[i]
+            res = o[i]
+            if rule['is_encrypt'] == 1:
+                c = Cipher(TripleDES(rule['key'].encode('utf-8')), ECB())
+                data = str(res).encode('utf-8')
+                # 补足字节
+                data += b'\x00' * 8
+                res = base64.b64encode(c.encryptor().update(data)).decode('utf-8')
+            result[rule['obfuscated_name']] = res
+        else:
+            result[i] = o[i]
+    return result
+
+
+# 通过测试
+def _AES(v: bytes, k: bytes):
+    iv = '0102030405060708'
+    key = AES(k)
+    c = Cipher(key, CBC(iv.encode('utf-8')))
+    c.encryptor()
+    # 填充明文
+    v += b'\x00'
+    while len(v) % 16 != 0:
+        v += b'\x00'
+    return c.encryptor().update(v).hex()
+
+
+def GZIP(o: dict):
+    # 这个压缩结果似乎和前台不太一样,不清楚是否会影响
+    json_str = json.dumps(o, ensure_ascii=False)
+    stream = gzip.compress(json_str.encode('utf-8'), 2, mtime=0)
+    return base64.b64encode(stream)
+
+
+# 获得tn的值,后续做DES加密用
+# 通过测试
+def get_tn(o: dict):
+    sorted_keys = sorted(o.keys())
+
+    result_list = []
+
+    for i in sorted_keys:
+        v = o[i]
+        if isinstance(v, (int, float)):
+            v = str(v * 10000)
+        elif isinstance(v, dict):
+            v = get_tn(v)
+        result_list.append(v)
+    return ''.join(result_list)
+
+
+def get_smid():
+    t = time.localtime()
+    _time = '{}{:0>2d}{:0>2d}{:0>2d}{:0>2d}{:0>2d}'.format(t.tm_year, t.tm_mon, t.tm_mday, t.tm_hour, t.tm_min,
+                                                           t.tm_sec)
+    uid = str(uuid.uuid4())
+    v = _time + hashlib.md5(uid.encode('utf-8')).hexdigest() + '00'
+    smsk_web = hashlib.md5(('smsk_web_' + v).encode('utf-8')).hexdigest()[0:14]
+    return v + smsk_web + '0'
+
+
+def get_d_id():
+    # storageName = '.thumbcache_' + md5(SM_CONFIG['organization']) // 用于从本地存储获得值
+    # uid = uuid()
+    # priId=md5(uid)[0:16]
+    # ep=rsa(uid,publicKey)
+    # SMID = localStorage.get(storageName);// 获得本地存储存的值
+    # _0x30b2eb为递归md5
+
+    uid = str(uuid.uuid4()).encode('utf-8')
+    priId = hashlib.md5(uid).hexdigest()[0:16]
+    # ep不一定对，先走走看
+    ep = PK.encrypt(uid, padding.PKCS1v15())
+    ep = base64.b64encode(ep).decode('utf-8')
+
+    browser = BROWSER_ENV.copy()
+    current_time = int(time.time() * 1000)
+    browser.update({
+        'vpw': str(uuid.uuid4()),
+        'svm': current_time,
+        'trees': str(uuid.uuid4()),
+        'pmf': current_time
+    })
+
+    des_target = {
+        **browser,
+        'protocol': 102,
+        'organization': SM_CONFIG['organization'],
+        'appId': SM_CONFIG['appId'],
+        'os': 'web',
+        'version': '3.0.0',
+        'sdkver': '3.0.0',
+        'box': '',  # 似乎是个SMID，但是第一次的时候是空,不过不影响结果
+        'rtype': 'all',
+        'smid': get_smid(),
+        'subVersion': '1.0.0',
+        'time': 0
+    }
+    des_target['tn'] = hashlib.md5(get_tn(des_target).encode()).hexdigest()
+
+    des_result = _AES(GZIP(_DES(des_target)), priId.encode('utf-8'))
+
+    response = requests.post(devices_info_url, json={
+        'appId': 'default',
+        'compress': 2,
+        'data': des_result,
+        'encode': 5,
+        'ep': ep,
+        'organization': SM_CONFIG['organization'],
+        'os': 'web'  # 固定值
+    })
+
+    resp = response.json()
+    if resp['code'] != 1100:
+        raise Exception("did计算失败，请联系作者")
+    # 开头必须是B
+    return 'B' + resp['detail']['deviceId']

src/SKland_Auto_Sign/auto_sign.py

@@ -1,161 +0,0 @@
-import hashlib
-import hmac
-import json
-import os.path
-import time
-import logging
-import requests
-
-from urllib import parse
-
-app_code = '4ca99fa6b56cc2ba'
-token_env = os.environ.get('TOKEN')
-header = {
-    'cred': '',
-    'User-Agent': 'Skland/1.0.1 (com.hypergryph.skland; build:100001014; Android 31; ) Okhttp/4.11.0',
-    'Accept-Encoding': 'gzip',
-    'Connection': 'close'
-}
-header_login = {
-    'User-Agent': 'Skland/1.0.1 (com.hypergryph.skland; build:100001014; Android 31; ) Okhttp/4.11.0',
-    'Accept-Encoding': 'gzip',
-    'Connection': 'close'
-}
-
-# 签名请求头一定要这个顺序，否则失败
-# timestamp是必填的,其它三个随便填,不要为none即可
-header_for_sign = {
-    'platform': '',
-    'timestamp': '',
-    'dId': '',
-    'vName': ''
-}
-
-# 签到url
-sign_url = "https://zonai.skland.com/api/v1/game/attendance"
-# 绑定的角色url
-binding_url = "https://zonai.skland.com/api/v1/game/player/binding"
-# 使用token获得认证代码
-grant_code_url = "https://as.hypergryph.com/user/oauth2/v2/grant"
-# 使用认证代码获得cred
-cred_code_url = "https://zonai.skland.com/api/v1/user/auth/generate_cred_by_code"
-
-
-def generate_signature(token: str, path, body_or_query):
-    """
-    获得签名头
-    接口地址+方法为Get请求？用query否则用body+时间戳+ 请求头的四个重要参数（dId，platform，timestamp，vName）.toJSON()
-    将此字符串做HMAC加密，算法为SHA-256，密钥token为请求cred接口会返回的一个token值
-    再将加密后的字符串做MD5即得到sign
-    :param token: 拿cred时候的token
-    :param path: 请求路径（不包括网址）
-    :param body_or_query: 如果是GET，则是它的query。POST则为它的body
-    :return: 计算完毕的sign
-    """
-    # 总是说请勿修改设备时间，怕不是yj你的服务器有问题吧，所以这里特地-2
-    t = str(int(time.time()) - 2)
-    token = token.encode('utf-8')
-    header_ca = json.loads(json.dumps(header_for_sign))
-    header_ca['timestamp'] = t
-    header_ca_str = json.dumps(header_ca, separators=(',', ':'))
-    s = path + body_or_query + t + header_ca_str
-    hex_s = hmac.new(token, s.encode('utf-8'), hashlib.sha256).hexdigest()
-    md5 = hashlib.md5(hex_s.encode('utf-8')).hexdigest().encode('utf-8').decode('utf-8')
-    return md5, header_ca
-
-
-def get_sign_header(url: str, method, body, old_header, local_s_token):
-    h = json.loads(json.dumps(old_header))
-    p = parse.urlparse(url)
-    if method.lower() == 'get':
-        h['sign'], header_ca = generate_signature(local_s_token, p.path, p.query)
-    else:
-        h['sign'], header_ca = generate_signature(local_s_token, p.path, json.dumps(body))
-    for i in header_ca:
-        h[i] = header_ca[i]
-    return h
-
-
-def get_cred_by_token(token):
-    grant_code = get_grant_code(token)
-    return get_cred(grant_code)
-
-
-def get_grant_code(token):
-    response = requests.post(grant_code_url, json={
-        'appCode': app_code,
-        'token': token,
-        'type': 0
-    }, headers=header_login)
-    resp = response.json()
-    if response.status_code != 200:
-        raise Exception(f'获得认证代码失败：{resp}')
-    if resp.get('status') != 0:
-        raise Exception(f'获得认证代码失败：{resp["msg"]}')
-    return resp['data']['code']
-
-
-def get_cred(grant):
-    resp = requests.post(cred_code_url, json={
-        'code': grant,
-        'kind': 1
-    }, headers=header_login).json()
-    if resp['code'] != 0:
-        raise Exception(f'获得cred失败：{resp["message"]}')
-    return resp['data']
-
-
-def get_binding_list(s_token):
-    error_message, error_code = 'Success', 0
-    result_list = []
-    header_dict = get_sign_header(binding_url, 'get', None, header, s_token)
-    resp = requests.get(binding_url, headers=header_dict).json()
-
-    if resp['code'] != 0:
-        error_message = f"请求角色列表出现问题：{resp['message']}"
-        error_code = -1
-        return error_message, error_code, []
-    for i in resp['data']['list']:
-        if i.get('appCode') != 'arknights':
-            continue
-        result_list.extend(i.get('bindingList'))
-    return error_message, error_code, result_list
-
-
-def single_sign(cred_resp):
-    s_token = cred_resp['token']
-    header['cred'] = cred_resp['cred']
-    msg, code, characters = get_binding_list(s_token)
-    if code != 0 or not characters:
-        return msg, code
-
-    content = characters[0]
-    body = {'gameId': 1, 'uid': content.get('uid')}
-
-    resp = requests.post(sign_url, headers=get_sign_header(sign_url, 'post', body, header,s_token), json=body).json()
-    if resp['code'] != 0:
-        msg = f'角色{content.get("nickName")}({content.get("channelName")})签到失败了！原因：{resp.get("message")}'
-        code = -1
-        return msg, code
-    awards = resp['data']['awards']
-    for j in awards:
-        res = j['resource']
-        msg = f'角色{content.get("nickName")}({content.get("channelName")})签到成功，获得{res["name"]}×{j.get("count") or 1}'
-        code = 0
-    return msg, code
-
-
-def new_start():
-    result_list = []
-    token_list = ['fqyRkfJrLlkuax8SzFffxc0W', 'i//Ozb2bFsXjxF9ZKgAgVi6K']
-    for i in token_list:
-        try:
-            credit_dict = get_cred_by_token(i)
-            msg, code = single_sign(credit_dict)
-        except Exception as ex:
-            print(f'签到失败，原因：{str(ex)}')
-    print("签到完成！")
-
-
-if __name__ == '__main__':
-    new_start()

src/SKland_Auto_Sign/new_auto_sign.py

@@ -0,0 +1,498 @@
+import hashlib
+import hmac
+import json
+import logging
+import os.path
+import threading
+import time
+from datetime import date
+from getpass import getpass
+from urllib import parse
+
+import requests
+
+from src.SKland_Auto_Sign.SecuritySm import get_d_id
+
+token_save_name = 'TOKEN.txt'
+app_code = '4ca99fa6b56cc2ba'
+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': '',
+    'User-Agent': 'Skland/1.0.1 (com.hypergryph.skland; build:100001014; Android 31; ) Okhttp/4.11.0',
+    'Accept-Encoding': 'gzip',
+    'Connection': 'close'
+}
+header_login = {
+    'User-Agent': 'Skland/1.0.1 (com.hypergryph.skland; build:100001014; Android 31; ) Okhttp/4.11.0',
+    'Accept-Encoding': 'gzip',
+    'Connection': 'close',
+    'dId': get_d_id()
+}
+
+# 签名请求头一定要这个顺序，否则失败
+# timestamp是必填的,其它三个随便填,不要为none即可
+header_for_sign = {
+    'platform': '',
+    'timestamp': '',
+    'dId': '',
+    'vName': ''
+}
+
+# 签到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
+login_code_url = "https://as.hypergryph.com/general/v1/send_phone_code"
+# 验证码登录
+token_phone_code_url = "https://as.hypergryph.com/user/auth/v2/token_by_phone_code"
+# 密码登录
+token_password_url = "https://as.hypergryph.com/user/auth/v1/token_by_phone_password"
+# 使用token获得认证代码
+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')
+    if not os.path.exists('logs'):
+        os.mkdir('logs')
+    logger = logging.getLogger()
+
+    file_handler = logging.FileHandler(f'./logs/{current_date}.log', encoding='utf-8')
+    logger.addHandler(file_handler)
+    logging.getLogger().setLevel(logging.DEBUG)
+    file_handler.setLevel(logging.INFO)
+    formatter = logging.Formatter('%(asctime)s - %(name)s - %(levelname)s - %(message)s')
+    file_handler.setFormatter(formatter)
+
+    def filter_code(text):
+        filter_key = ['code', 'cred', 'token']
+        try:
+            j = json.loads(text)
+            if not j.get('data'):
+                return text
+            data = j['data']
+            for i in filter_key:
+                if i in data:
+                    data[i] = '*****'
+            return json.dumps(j, ensure_ascii=False)
+        except:
+            return text
+
+    _get = requests.get
+    _post = requests.post
+
+    def get(*args, **kwargs):
+        response = _get(*args, **kwargs)
+        logger.info(f'GET {args[0]} - {response.status_code} - {filter_code(response.text)}')
+        return response
+
+    def post(*args, **kwargs):
+        response = _post(*args, **kwargs)
+        logger.info(f'POST {args[0]} - {response.status_code} - {filter_code(response.text)}')
+        return response
+
+    # 替换 requests 中的方法
+    requests.get = get
+    requests.post = post
+
+
+def generate_signature(token: str, path, body_or_query, sign_header_template=None):
+    """
+    获得签名头
+    接口地址+方法为Get请求？用query否则用body+时间戳+ 请求头的四个重要参数（dId，platform，timestamp，vName）.toJSON()
+    将此字符串做HMAC加密，算法为SHA-256，密钥token为请求cred接口会返回的一个token值
+    再将加密后的字符串做MD5即得到sign
+    :param token: 拿cred时候的token
+    :param path: 请求路径（不包括网址）
+    :param body_or_query: 如果是GET，则是它的query。POST则为它的body
+    :return: 计算完毕的sign
+    """
+    # 总是说请勿修改设备时间，怕不是yj你的服务器有问题吧，所以这里特地-2
+    t = str(int(time.time()) - 2)
+    token = token.encode('utf-8')
+    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
+    hex_s = hmac.new(token, s.encode('utf-8'), hashlib.sha256).hexdigest()
+    md5 = hashlib.md5(hex_s.encode('utf-8')).hexdigest().encode('utf-8').decode('utf-8')
+    # logging.info(f'算出签名: {md5}')
+    return md5, header_ca
+
+
+def get_sign_header(url: str, method, body, h):
+    p = parse.urlparse(url)
+    if method.lower() == 'get':
+        h['sign'], header_ca = generate_signature(http_local.token, p.path, p.query)
+    else:
+        h['sign'], header_ca = generate_signature(http_local.token, p.path, json.dumps(body))
+    for i in header_ca:
+        h[i] = header_ca[i]
+    return h
+
+
+def login_by_code():
+    phone = input('请输入手机号码：')
+    resp = requests.post(login_code_url, json={'phone': phone, 'type': 2}, headers=header_login).json()
+    if resp.get("status") != 0:
+        raise Exception(f"发送手机验证码出现错误：{resp['msg']}")
+    code = input("请输入手机验证码：")
+    r = requests.post(token_phone_code_url, json={"phone": phone, "code": code}, headers=header_login).json()
+    return get_token(r)
+
+
+def login_by_token():
+    token_code = input("请输入（登录森空岛电脑官网后请访问这个网址：https://web-api.skland.com/account/info/hg）:")
+    return parse_user_token(token_code)
+
+
+def parse_user_token(t):
+    try:
+        t = json.loads(t)
+        return t['data']['content']
+    except:
+        pass
+    return t
+
+
+def login_by_password():
+    phone = input('请输入手机号码：')
+    password = getpass('请输入密码(不会显示在屏幕上面)：')
+    r = requests.post(token_password_url, json={"phone": phone, "password": password}, headers=header_login).json()
+    return get_token(r)
+
+
+def get_cred_by_token(token):
+    grant_code = get_grant_code(token)
+    return get_cred(grant_code)
+
+
+def get_token(resp):
+    if resp.get('status') != 0:
+        raise Exception(f'获得token失败：{resp["msg"]}')
+    return resp['data']['token']
+
+
+def get_grant_code(token):
+    response = requests.post(grant_code_url, json={
+        'appCode': app_code,
+        'token': token,
+        'type': 0
+    }, headers=header_login)
+    resp = response.json()
+    if response.status_code != 200:
+        raise Exception(f'获得认证代码失败：{resp}')
+    if resp.get('status') != 0:
+        raise Exception(f'获得认证代码失败：{resp["msg"]}')
+    return resp['data']['code']
+
+
+def get_cred(grant):
+    resp = requests.post(cred_code_url, json={
+        'code': grant,
+        'kind': 1
+    }, headers=header_login).json()
+    if resp['code'] != 0:
+        raise Exception(f'获得cred失败：{resp["message"]}')
+    return resp['data']
+
+
+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()
+
+    if resp['code'] != 0:
+        print(f"请求角色列表出现问题：{resp['message']}")
+        if resp.get('message') == '用户未登录':
+            print(f'用户登录可能失效了，请重新运行此程序！')
+            os.remove(token_save_name)
+            return []
+    for i in resp['data']['list']:
+        if i.get('appCode') != target_app_code:
+            continue
+        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):
+    setup_cred_session(cred_resp)
+    characters = get_binding_list()
+
+    for i in characters:
+        body = {
+            '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()
+        _, msg = format_arknights_sign_message(i, resp)
+        print(msg)
+
+
+def single_sign(cred_resp):
+    setup_cred_session(cred_resp)
+    characters = get_binding_list()
+
+    if not characters:
+        return '未绑定明日方舟角色，已跳过', 0
+
+    code = 0
+    messages = []
+    for i in characters:
+        body = {
+            '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()
+        success, msg = format_arknights_sign_message(i, resp)
+        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
+
+        role_id = role.get('roleId')
+        server_id = role.get('serverId')
+        if role_id is None or server_id is None:
+            messages.append('终末地角色信息不完整，已跳过')
+            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):
+    with open(token_save_name, 'w') as f:
+        f.write(token)
+    print(
+        f'您的鹰角网络通行证保存在{token_save_name}, 打开这个可以把它复制到云函数服务器上执行!\n双击添加账号即可再次添加账号')
+
+
+def read(path):
+    if not os.path.exists(token_save_name):
+        return []
+    v = []
+    with open(path, 'r', encoding='utf-8') as f:
+        for i in f.readlines():
+            i = i.strip()
+            i and i not in v and v.append(i)
+    return v
+
+
+def read_from_env():
+    v = []
+    token_list = token_env.split(',')
+    for i in token_list:
+        i = i.strip()
+        if i and i not in v:
+            v.append(parse_user_token(i))
+    print(f'从环境变量中读取到{len(v)}个token...')
+    return v
+
+
+def init_token():
+    if token_env:
+        print('使用环境变量里面的token')
+        # 对于github action,不需要存储token,因为token在环境变量里
+        return read_from_env()
+    tokens = []
+    tokens.extend(read(token_save_name))
+    add_account = current_type == 'add_account'
+    if add_account:
+        print('！！！您启用了添加账号模式，将不会签到！！！')
+    if len(tokens) == 0 or add_account:
+        tokens.append(input_for_token())
+    save('\n'.join(tokens))
+    return [] if add_account else tokens
+
+
+def input_for_token():
+    print("请输入你需要做什么：")
+    print("1.使用用户名密码登录（非常推荐）")
+    print("2.使用手机验证码登录（非常推荐，但可能因为人机验证失败）")
+    print("3.手动输入鹰角网络通行证账号登录(推荐)")
+    mode = input('请输入（1，2，3）：')
+    if mode == '' or mode == '1':
+        token = login_by_password()
+    elif mode == '2':
+        token = login_by_code()
+    elif mode == '3':
+        token = login_by_token()
+    else:
+        exit(-1)
+    return token
+
+
+def start():
+    # token = init_token()
+    token = []
+    data_dict = {
+      "user_dict": {
+        "18021026530": {"password": "Mrfz1790990971", "status": True},
+        "13693680360": {"password": "Tanhuobin2022", "status": True}
+      }
+    }
+    for user, content in data_dict['user_dict'].items():
+        r = requests.post(token_password_url, json={"phone": user, "password": content['password']}, headers=header_login).json()
+        token.append(get_token(r))
+    for i in token:
+        try:
+            do_sign(get_cred_by_token(i))
+        except Exception as ex:
+            print(f'签到失败，原因：{str(ex)}')
+            logging.error('', exc_info=ex)
+    print("签到完成！")
+
+
+if __name__ == '__main__':
+    print('本项目源代码仓库：https://github.com/xxyz30/skyland-auto-sign(已被github官方封禁)')
+    print('https://gitee.com/FancyCabbage/skyland-auto-sign')
+    config_logger()
+
+    # logging.info('=========starting==========')
+
+    start_time = time.time()
+    start()
+    end_time = time.time()
+    # logging.info(f'complete with {(end_time - start_time) * 1000} ms')
+    # logging.info('===========ending============')

src/ark_helper_backend/backend.py

@@ -4,12 +4,13 @@ import logging
 import copy
 import requests
 import time
+import random
 
 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.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')
@@ -26,9 +27,9 @@ class ArkHelperBackend:
         self.user_config_dict = self.load_users()
         self.user_data_dict = self.initiate_user_data()
 
-        # # 更新用户token
+        # 更新用户token
-        # self.refresh_user_token()
+        self.refresh_user_token()
-
+        self.is_retrieving_data = True
         self.logger.info('ArkHelperBackend initiated')
 
     def load_users(self):
@@ -84,6 +85,8 @@ class ArkHelperBackend:
 
     def dump_json_data(self, json_path, json_data):
         try:
+            og_data = self.load_json_data(json_path)
+            og_data.update(json_data)
             with open(json_path, 'w', encoding='utf-8') as f:
                 f.write(json.dumps(json_data, ensure_ascii=False))
         except FileNotFoundError:
@@ -134,15 +137,27 @@ class ArkHelperBackend:
             self.single_user_login(phone)
         self.save_data(payment_status=False, gacha_status=False)
 
-    def skland_sign(self):
+    def _run_sign_jobs(self, sign_jobs):
-        token_list = [content['token'] for _, content in self.user_data_dict.items()]
+        for user, content in self.user_data_dict.items():
-        for token in token_list:
+            token = content.get('token')
-            try:
+            if token:
-                credit_dict = get_cred_by_token(token)
+                try:
-                msg, code = single_sign(credit_dict)
+                    cred_resp = get_cred_by_token(token)
-                self.logger.info(f'{msg}: {code}')
+                    for sign_name, sign_handler in sign_jobs:
-            except Exception as ex:
+                        msg, code = sign_handler(cred_resp)
-                self.logger.error(f'签到失败，原因：{str(ex)}')
+                        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:
@@ -175,7 +190,10 @@ class ArkHelperBackend:
                 payment_data = {}
             token = self.user_data_dict[phone]['token']
             data = {"appId": 1, "channelMasterId": 1, "channelToken": {"token": token}}
-            pay_data = json.loads(requests.post(url, json.dumps(data)).text).get('data')
+
+            resp = requests.post(url, json=data)
+
+            pay_data = resp.json().get('data')
             for item in pay_data:
                 i = pay_data.index(item)
                 pay_time = item['payTime']
@@ -184,10 +202,33 @@ class ArkHelperBackend:
             pay_dict = {item['orderId']: item for item in pay_data}
             payment_data.update(pay_dict)
             self.user_data_dict[phone]['payment'] = payment_data
+            self.logger.info(f'{phone} payment data updated')
         self.save_data(token_status=False, gacha_status=False)
 
+    def data_retrieve(self):
+        self.refresh_user_token()
+        # self.update_payment_data()
+        # self.update_gacha_data()
+
+    def show_data(self):
+        gacha_data = {user: content['gacha'] for user, content in self.user_data_dict.items()}
+        payment_data = {user: content['payment'] for user, content in self.user_data_dict.items()}
+        return gacha_data, payment_data
+
 
 if __name__ == '__main__':
-    ak_back = ArkHelperBackend()
+    # ak_back = ArkHelperBackend()
-    ak_back.update_payment_data()
+    # ak_back.update_gacha_data()
+    # ak_back.update_payment_data()
+    # 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')

src/util/check_connection.py

@@ -0,0 +1,7 @@
+import requests
+
+url = 'http://192.168.195.103:61000/GetFieldData'
+r = requests.post(url, {
+	"field_name": "supported_llm"
+}, timeout=10)
+print(r.text)

src/util/mail.py

@@ -0,0 +1,137 @@
+import smtplib
+import traceback
+
+from email.mime.text import MIMEText
+from email.mime.multipart import MIMEMultipart
+from email.mime.base import MIMEBase
+from email import encoders
+
+
+def send_batch_email(receiver_email_list, subject, content, file_path=None):
+    """
+    Send Email to the receiver list. Can attach files if needed.
+    :param receiver_email_list:
+    :param subject: Title
+    :param content: Word Content
+    :param file_path: file list
+    :return:
+    """
+    # sender config
+    sender_email = "ArmorServer@outlook.com"
+    # password = "ServerOnly2024!"
+    password = 'fczojhcahxwonosh'
+
+    # mail content
+    message = MIMEMultipart()
+    message["Subject"] = subject
+    message["From"] = sender_email
+
+    # main body
+    part1 = MIMEText(content, "plain")
+    message.attach(part1)
+
+    if file_path and isinstance(file_path, list):
+        for filename in file_path:
+            # 指定附件的路径和文件名
+            attachment = open(filename, 'rb')
+
+            # 创建MIMEBase实例
+            part = MIMEBase('application', 'octet-stream')
+            part.set_payload(attachment.read())
+            encoders.encode_base64(part)
+
+            # 添加邮件头
+            part.add_header(
+                'Content-Disposition',
+                f'attachment; filename= {filename}',
+            )
+            message.attach(part)
+            attachment.close()
+
+    # 登录SMTP服务器并发送邮件
+    # 使用Hotmail的SMTP服务器
+    server = smtplib.SMTP('smtp.office365.com', 587)
+    try:
+        server.starttls()  # 启用安全传输模式
+        server.login(sender_email, password)
+
+        for receiver_email in receiver_email_list:
+            message["To"] = receiver_email
+            server.sendmail(sender_email, receiver_email, message.as_string())
+            print(f"Email to {receiver_email} sent successfully!")
+    except Exception as e:
+        print(f"Error: {e}")
+    finally:
+        server.quit()
+
+
+def new_send_email(receiver_email_list, subject, content, file_path=None):
+    """
+    Send Email to the receiver list. Can attach files if needed.
+    :param receiver_email_list:
+    :param subject: Title
+    :param content: Word Content
+    :param file_path: file list
+    :return:
+    """
+    # sender config
+    sender_email = "2654988228@qq.com"
+    password = 'mdibrqwjqmdyeabe'
+
+    # mail content
+    message = MIMEMultipart()
+    message["Subject"] = subject
+    message["From"] = sender_email
+
+    # main body
+    part1 = MIMEText(content, "plain")
+    message.attach(part1)
+
+    if file_path and isinstance(file_path, list):
+        for filename in file_path:
+            # 指定附件的路径和文件名
+            attachment = open(filename, 'rb')
+
+            # 创建MIMEBase实例
+            part = MIMEBase('application', 'octet-stream')
+            part.set_payload(attachment.read())
+            encoders.encode_base64(part)
+
+            # 添加邮件头
+            part.add_header(
+                'Content-Disposition',
+                f'attachment; filename= {filename}',
+            )
+            message.attach(part)
+            attachment.close()
+
+    # 登录SMTP服务器并发送邮件
+    # 使用Hotmail的SMTP服务器
+    server = smtplib.SMTP('smtp.qq.com', 587)
+    try:
+        server.starttls()  # 启用安全传输模式
+        server.login(sender_email, password)
+
+        for receiver_email in receiver_email_list:
+            message["To"] = receiver_email
+            server.sendmail(sender_email, receiver_email, message.as_string())
+            print(f"Email to {receiver_email} sent successfully!")
+    except Exception as e:
+        print(f"Error: {e}")
+        print(str(traceback.format_exc()))
+    finally:
+        server.quit()
+
+
+if __name__ == '__main__':
+    # target_email_list = ['2654988228@qq.com']
+    target_email_list = ['ArmorServer@outlook.com', '2654988228@qq.com']
+    email_subject = 'Test Message'
+    email_body = """
+    This is a test email 
+    """
+
+    email_file_list = []
+
+    new_send_email(target_email_list, email_subject, email_body, email_file_list)
+

src/util/utils.py

@@ -1,6 +1,7 @@
 import requests
 import json
 import os
+import random
 
 from pathlib import Path
 
@@ -62,5 +63,17 @@ def validate_token(token):
         return True
 
 
+def random_time(start_time=480, end_time=720):
+    # 生成从8*60(480)到12*60(720)之间的随机分钟数
+    random_minutes = random.randint(start_time, end_time)
+
+    # 将分钟数转换成小时和分钟
+    hour = random_minutes // 60
+    minute = random_minutes % 60
+
+    # 返回格式化的时间
+    return "{:02}:{:02}".format(hour, minute)
+
+
 if __name__ == '__main__':
     print(load_user_config())

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')