diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f26e5cc --- /dev/null +++ b/.gitignore @@ -0,0 +1 @@ +/config/user_info_local.json diff --git a/config/user_info.json b/config/user_info.json new file mode 100644 index 0000000..ea085c3 --- /dev/null +++ b/config/user_info.json @@ -0,0 +1,3 @@ +{ + "user_dict": {} +} \ No newline at end of file diff --git a/data/ArkHelperData/placeholder b/data/ArkHelperData/placeholder new file mode 100644 index 0000000..e69de29 diff --git a/logs/placeholder b/logs/placeholder new file mode 100644 index 0000000..e69de29 diff --git a/src/My_Logger/__init__.py b/src/My_Logger/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/My_Logger/project_logger.py b/src/My_Logger/project_logger.py new file mode 100644 index 0000000..f403a3e --- /dev/null +++ b/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 diff --git a/src/SKland_Auto_Sign/__init__.py b/src/SKland_Auto_Sign/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/SKland_Auto_Sign/auto_sign.py b/src/SKland_Auto_Sign/auto_sign.py new file mode 100644 index 0000000..9f467b4 --- /dev/null +++ b/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() diff --git a/src/ark_helper_backend/__init__.py b/src/ark_helper_backend/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/src/ark_helper_backend/backend.py b/src/ark_helper_backend/backend.py new file mode 100644 index 0000000..b336fbc --- /dev/null +++ b/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() + diff --git a/src/util/utils.py b/src/util/utils.py index 6facc55..cf0f7c9 100644 --- a/src/util/utils.py +++ b/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())