auto sign

.gitignore

@@ -0,0 +1 @@
+/config/user_info_local.json

config/user_info.json

@@ -0,0 +1,3 @@
+{
+  "user_dict": {}
+}

src/My_Logger/project_logger.py

@@ -0,0 +1,66 @@
+import logging
+import datetime
+import os
+import copy
+
+from pathlib import Path
+
+project_path = Path(__file__).parent.parent.parent.absolute()
+data_path = os.path.join(project_path, 'data')
+log_path = os.path.join(project_path, 'logs')
+
+
+class LoggerClass:
+    def __init__(self):
+        # 创建日志器对象
+        self.format = logging.Formatter(f'[%(filename)s: %(funcName)s:%(lineno)4d][%(asctime)s][%(levelname)-.5s]: '
+                                        f'%(message)s',
+                                        datefmt='%Y-%m-%d %H:%M:%S')
+
+    def console_handler(self, level):
+        # 创建控制台的日志处理器
+        console_handler = logging.StreamHandler()
+        console_handler.setLevel(level)
+
+        # 处理器添加输出格式
+        console_handler.setFormatter(self.format)
+
+        # 返回控制器
+        return console_handler
+
+    def file_handler(self, level, log_file_path):
+        # 创建文件的日志处理器
+        if os.path.isdir(log_file_path):
+            path = os.path.join(log_file_path, 'default_log.txt')
+        else:
+            path = log_file_path if log_file_path.endswith('.txt') else os.path.join(log_file_path, 'default_log.txt')
+        file_handler = logging.FileHandler(path, mode="a", encoding="utf-8")
+        file_handler.setLevel(level)
+
+        # 处理器添加输出格式
+        file_handler.setFormatter(self.format)
+
+        # 返回控制器
+        return file_handler
+
+    def get_log(self, level, log_file_path=log_path):
+        logger = logging.getLogger()
+        logger.setLevel(level)
+        # 日志器中添加控制台处理器
+        logger.addHandler(self.console_handler(level))
+        # 日志器中添加文件处理器
+        logger.addHandler(self.file_handler(level, log_file_path))
+
+        # 返回日志实例对象
+        return logger
+
+
+if __name__ == '__main__':
+    # log = LoggerClass()
+    # test_logger = log.get_log(logging.DEBUG)
+    # test_logger.debug('debug')
+    # test_logger.info('info')
+    # test_logger.warning('warning')
+    # test_logger.error('error')
+    # test_logger.critical('critical')
+    pass

src/SKland_Auto_Sign/auto_sign.py

@@ -0,0 +1,161 @@
+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/ark_helper_backend/backend.py

@@ -0,0 +1,150 @@
+import json
+import os
+import logging
+import copy
+import requests
+
+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
+
+PROJECT_PATH = Path(__file__).parent.parent.parent.absolute()
+DATA_PATH = os.path.join(PROJECT_PATH, 'data', 'ArkHelperData')
+CONFIG_PATH = os.path.join(PROJECT_PATH, 'config')
+
+
+class ArkHelperBackend:
+    def __init__(self, logger=None):
+        if not isinstance(logger, logging.RootLogger):
+            self.logger = LoggerClass().get_log(logging.INFO)
+        else:
+            self.logger = logger
+
+        self.user_config_dict = self.load_users()
+        self.user_data_dict = self.initiate_user_data()
+
+        # # 更新用户token
+        # self.refresh_user_token()
+
+        self.logger.info('ArkHelperBackend initiated')
+
+    def load_users(self):
+        """
+        load user config from disk. Only load when status = True
+        :return:
+        """
+        user_config_dict = load_user_config().get('user_dict')
+        result_dict = {}
+        for phone, content in user_config_dict.items():
+            if content.get('status'):
+                result_dict[phone] = content
+            else:
+                self.logger.info(f'{phone} is not loaded.')
+        return result_dict
+
+    def initiate_user_data(self):
+        """
+        Build user data dict. If local file exists, use local file.
+        :return:
+        """
+        user_data_dict = copy.deepcopy(self.user_config_dict)
+        token_path = os.path.join(DATA_PATH, 'token_data.json')
+        gacha_path = os.path.join(DATA_PATH, 'gacha_data.json')
+        payment_path = os.path.join(DATA_PATH, 'payment_data.json')
+
+        if not os.path.exists(token_path):
+            self.logger.info('initializing token data.')
+            token_data = {}
+            self.dump_json_data(token_path, token_data)
+        else:
+            token_data = self.load_json_data(token_path)
+
+        if not os.path.exists(gacha_path):
+            self.logger.info('initializing gacha data.')
+            gacha_data = {}
+            self.dump_json_data(gacha_path, gacha_data)
+        else:
+            gacha_data = self.load_json_data(gacha_path)
+
+        if not os.path.exists(payment_path):
+            self.logger.info('initializing payment data.')
+            payment_data = {}
+            self.dump_json_data(payment_path, payment_data)
+        else:
+            payment_data = self.load_json_data(payment_path)
+
+        for phone, content in user_data_dict.items():
+            user_data_dict[phone]['token'] = token_data.get(phone, None)
+            user_data_dict[phone]['gacha'] = gacha_data.get(phone, None)
+            user_data_dict[phone]['payment'] = payment_data.get(phone, None)
+        return user_data_dict
+
+    def dump_json_data(self, json_path, json_data):
+        try:
+            with open(json_path, 'w', encoding='utf-8') as f:
+                f.write(json.dumps(json_data, ensure_ascii=False))
+        except FileNotFoundError:
+            self.logger.error(f'{json_path} not found')
+
+    def load_json_data(self, json_path):
+        try:
+            with open(json_path, 'r', encoding='utf-8') as f:
+                json_data = json.loads(f.read())
+            return json_data
+        except FileNotFoundError:
+            self.logger.error(f'{json_path} not found')
+
+    def save_data(self, token_status=True, payment_status=True, gacha_status=True):
+        token_path = os.path.join(DATA_PATH, 'token_data.json')
+        gacha_path = os.path.join(DATA_PATH, 'gacha_data.json')
+        payment_path = os.path.join(DATA_PATH, 'payment_data.json')
+        result_list = []
+        if token_status:
+            token_data = {user: content['token'] for user, content in self.user_data_dict.items()}
+            self.dump_json_data(token_path, token_data)
+            result_list.append('Token data')
+        if gacha_status:
+            gacha_data = {user: content['gacha'] for user, content in self.user_data_dict.items()}
+            self.dump_json_data(gacha_path, gacha_data)
+            result_list.append('Gacha data')
+        if payment_status:
+            payment_data = {user: content['payment'] for user, content in self.user_data_dict.items()}
+            self.dump_json_data(payment_path, payment_data)
+            result_list.append('Payment data')
+        self.logger.info(f'{", ".join(result_list)} saved')
+
+    def single_user_login(self, username):
+        user_token = self.user_data_dict[username]['token']
+        password = self.user_data_dict[username]['password']
+        token_status = validate_token(user_token)
+        if not token_status:
+            user_token = hg_get_new_user_token(username, password)
+            token_status = validate_token(user_token)
+            if not token_status:
+                self.logger.error(f'user: {username} wrong password, disabled for now')
+                self.user_config_dict[username]['status'] = False
+                return
+        self.user_data_dict[username]['token'] = user_token
+
+    def refresh_user_token(self):
+        for phone in self.user_data_dict:
+            self.single_user_login(phone)
+        self.save_data(payment_status=False, gacha_status=False)
+
+    def skland_sign(self):
+        token_list = [content['token'] for _, content in self.user_data_dict.items()]
+        for token in token_list:
+            try:
+                credit_dict = get_cred_by_token(token)
+                msg, code = single_sign(credit_dict)
+                self.logger.info(f'{msg}: {code}')
+            except Exception as ex:
+                self.logger.error(f'签到失败，原因：{str(ex)}')
+
+
+if __name__ == '__main__':
+    ak_back = ArkHelperBackend()
+    ak_back.skland_sign()
+

src/util/utils.py

@@ -4,8 +4,21 @@ import os
 
 from pathlib import Path
 
-project_path = Path(__file__).parent.parent.absolute()
+project_path = Path(__file__).parent.parent.parent.absolute()
 data_path = os.path.join(project_path, 'data')
+config_path = os.path.join(project_path, 'config')
+
+
+def load_user_config():
+    if os.path.exists(os.path.join(config_path, 'user_info_local.json')):
+        config_file_path = os.path.join(config_path, 'user_info_local.json')
+    elif os.path.exists(os.path.join(config_path, 'user_info.json')):
+        config_file_path = os.path.join(config_path, 'user_info.json')
+    else:
+        raise FileNotFoundError('user_info_local.json or user_info.json not exist in config dir!')
+    with open(config_file_path, 'r', encoding='utf-8') as f:
+        user_config = json.load(f)
+    return user_config
 
 
 def hg_get_new_user_token(username, password):
@@ -38,3 +51,16 @@ def hg_login(username, password, token_path=None):
         if gacha_response == '{"code":3000,"msg":"登录失效"}':
             print('用户名/密码错误')
             raise ValueError
+
+
+def validate_token(token):
+    get_gacha_url = f'https://ak.hypergryph.com/user/api/inquiry/gacha?page=1&token={token}&channelId=1'
+    gacha_response = requests.get(get_gacha_url).text
+    if gacha_response == '{"code":3000,"msg":"登录失效"}':
+        return False
+    else:
+        return True
+
+
+if __name__ == '__main__':
+    print(load_user_config())