diff --git a/.gitignore b/.gitignore index f26e5cc..e747eb5 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,3 @@ /config/user_info_local.json +/.venv/ +/data/ArknightsStoryJson-main/ diff --git a/run_service_backend.py b/run_service_backend.py index 0cfa80b..98c777e 100644 --- a/run_service_backend.py +++ b/run_service_backend.py @@ -34,12 +34,12 @@ def start_data_retrieve(): return 'No data supplied' if input_param_dict.get('admin_key') != admin_key: return '403 Forbidden' - if ark_back.is_retrieving_data(): + if ark_back.is_retrieving_data: return json.dumps({'status': 1, 'data': 'already started'}) else: ark_back.is_retrieving_data = True s1 = threading.Thread(target=data_retrieve_thread) - s2 = threading.Thread(target=skland_sign_in_thread()) + s2 = threading.Thread(target=skland_sign_in_thread) s1.start() s2.start() return json.dumps({'status': 0, 'data': ark_back.is_retrieving_data}) @@ -55,7 +55,9 @@ def change_status(): if input_param_dict.get('status') == 'on going' and not ark_back.is_retrieving_data: ark_back.is_retrieving_data = True s1 = threading.Thread(target=data_retrieve_thread) + s2 = threading.Thread(target=skland_sign_in_thread) s1.start() + s2.start() return json.dumps({'status': 0, 'data': ark_back.is_retrieving_data}) elif input_param_dict.get('status') == 'breaking' and ark_back.is_retrieving_data: ark_back.is_retrieving_data = False @@ -112,7 +114,7 @@ def skland_sign_in_thread(): try: last_run = None next_run_time = random_time(480, 720) - ark_back.logger.info(f'Next skland sign-in time: {next_run_time}') + ark_back.logger.info(f'Next daily sign-in time: {next_run_time}') while 1: if not ark_back.is_retrieving_data: break @@ -120,12 +122,12 @@ def skland_sign_in_thread(): now = datetime.datetime.now() if now.strftime('%H:%M') == next_run_time and last_run != now.date(): next_run_time = random_time(480, 720) - ark_back.skland_sign_new() + ark_back.daily_sign_new() last_run = now.date() except Exception as e: print(e, traceback.format_exc()) ark_back.logger.error(traceback.format_exc()) - new_send_email(target_email_list, 'Error in ArkHelper Skland Sign-in', str(traceback.format_exc())) + new_send_email(target_email_list, 'Error in ArkHelper Daily Sign-in', str(traceback.format_exc())) finally: ark_back.logger.info('Data Retrieve Stopped') @@ -139,7 +141,7 @@ threading.Thread(target=data_retrieve_thread).start() if __name__ == '__main__': print('start') - ark_back.skland_sign_new() + ark_back.daily_sign_new() # logger = LoggerClass().get_log(logging.INFO) # ark_back = ArkHelperBackend(logger=logger) # ark_back.update_gacha_data() diff --git a/src/SKland_Auto_Sign/new_auto_sign.py b/src/SKland_Auto_Sign/new_auto_sign.py index 609dcd9..a7da05a 100644 --- a/src/SKland_Auto_Sign/new_auto_sign.py +++ b/src/SKland_Auto_Sign/new_auto_sign.py @@ -19,6 +19,12 @@ token_env = os.environ.get('TOKEN') # 现在想做什么？ current_type = os.environ.get('SKYLAND_TYPE') +ARKNIGHTS_APP_CODE = 'arknights' +ENDFIELD_APP_CODE = 'endfield' +ARKNIGHTS_GAME_ID = 1 +ENDFIELD_SIGN_PLATFORM = '3' +ENDFIELD_SIGN_VNAME = '1.0.0' + http_local = threading.local() header = { 'cred': '', @@ -44,6 +50,7 @@ header_for_sign = { # 签到url sign_url = "https://zonai.skland.com/api/v1/game/attendance" +endfield_sign_url = "https://zonai.skland.com/api/v1/game/endfield/attendance" # 绑定的角色url binding_url = "https://zonai.skland.com/api/v1/game/player/binding" # 验证码url @@ -57,6 +64,13 @@ grant_code_url = "https://as.hypergryph.com/user/oauth2/v2/grant" # 使用认证代码获得cred cred_code_url = "https://zonai.skland.com/web/v1/user/auth/generate_cred_by_code" +header_for_endfield_sign = { + 'platform': ENDFIELD_SIGN_PLATFORM, + 'timestamp': '', + 'dId': '', + 'vName': ENDFIELD_SIGN_VNAME +} + def config_logger(): current_date = date.today().strftime('%Y-%m-%d') @@ -103,7 +117,7 @@ def config_logger(): requests.post = post -def generate_signature(token: str, path, body_or_query): +def generate_signature(token: str, path, body_or_query, sign_header_template=None): """ 获得签名头 接口地址+方法为Get请求？用query否则用body+时间戳+ 请求头的四个重要参数（dId，platform，timestamp，vName）.toJSON() @@ -117,7 +131,7 @@ def generate_signature(token: str, path, body_or_query): # 总是说请勿修改设备时间，怕不是yj你的服务器有问题吧，所以这里特地-2 t = str(int(time.time()) - 2) token = token.encode('utf-8') - header_ca = json.loads(json.dumps(header_for_sign)) + header_ca = json.loads(json.dumps(sign_header_template or header_for_sign)) header_ca['timestamp'] = t header_ca_str = json.dumps(header_ca, separators=(',', ':')) s = path + body_or_query + t + header_ca_str @@ -204,7 +218,13 @@ def get_cred(grant): return resp['data'] -def get_binding_list(): +def setup_cred_session(cred_resp): + http_local.token = cred_resp['token'] + http_local.header = header.copy() + http_local.header['cred'] = cred_resp['cred'] + + +def get_binding_list(target_app_code=ARKNIGHTS_APP_CODE): v = [] resp = requests.get(binding_url, headers=get_sign_header(binding_url, 'get', None, http_local.header)).json() @@ -215,67 +235,170 @@ def get_binding_list(): os.remove(token_save_name) return [] for i in resp['data']['list']: - if i.get('appCode') != 'arknights': + if i.get('appCode') != target_app_code: continue - v.extend(i.get('bindingList')) + v.extend(i.get('bindingList') or []) return v +def is_duplicate_sign_message(message): + return '请勿重复签到' in (message or '') or 'Please do not sign in again!' in (message or '') + + +def format_arknights_sign_message(character, resp): + role_name = character.get('nickName') or '未知角色' + channel_name = character.get('channelName') or '未知服务器' + if resp['code'] != 0: + error_message = resp.get('message') or '未知错误' + if is_duplicate_sign_message(error_message): + return True, f'角色{role_name}({channel_name})今日已签到，请勿重复签到' + return False, f'角色{role_name}({channel_name})签到失败了！原因：{error_message}' + + awards = resp.get('data', {}).get('awards') or [] + award_text = [] + for award in awards: + resource = award.get('resource') or {} + resource_name = resource.get('name') + if not resource_name: + continue + award_text.append(f'{resource_name}×{award.get("count") or 1}') + + if award_text: + return True, f'角色{role_name}({channel_name})签到成功，获得了{"、".join(award_text)}' + return True, f'角色{role_name}({channel_name})签到成功' + + +def get_endfield_role(binding): + role = binding.get('defaultRole') + if role: + return role + roles = binding.get('roles') or [] + return roles[0] if roles else None + + +def get_endfield_sign_header(role_str): + parsed_url = parse.urlparse(endfield_sign_url) + sign, header_ca = generate_signature(http_local.token, parsed_url.path, '', header_for_endfield_sign) + sign_header = http_local.header.copy() + sign_header['sign'] = sign + sign_header['platform'] = header_ca['platform'] + sign_header['timestamp'] = header_ca['timestamp'] + sign_header['dId'] = header_ca['dId'] + sign_header['vName'] = header_ca['vName'] + sign_header['sk-game-role'] = role_str + sign_header['Content-Type'] = 'application/json' + return sign_header + + +def format_endfield_awards(resp_data): + award_ids = resp_data.get('awardIds') or [] + resource_map = resp_data.get('resourceInfoMap') or {} + award_text = [] + + for award in award_ids: + award_id = award.get('id') + if award_id is None: + continue + resource = resource_map.get(award_id) or resource_map.get(str(award_id)) + if not resource: + continue + count = award.get('count') or resource.get('count') or 1 + award_text.append(f'{resource.get("name", award_id)}×{count}') + return award_text + + +def format_endfield_sign_message(binding, resp): + role = get_endfield_role(binding) or {} + role_name = role.get('nickname') or binding.get('nickName') or '未知角色' + channel_name = role.get('serverName') or binding.get('channelName') or '未知服务器' + + if resp['code'] != 0: + error_message = resp.get('message') or '未知错误' + if is_duplicate_sign_message(error_message): + return True, f'终末地角色{role_name}({channel_name})今日已签到，请勿重复签到' + return False, f'终末地角色{role_name}({channel_name})签到失败了！原因：{error_message}' + + award_text = format_endfield_awards(resp.get('data', {})) + if award_text: + return True, f'终末地角色{role_name}({channel_name})签到成功，获得了{"、".join(award_text)}' + return True, f'终末地角色{role_name}({channel_name})签到成功' + + def list_awards(game_id, uid): resp = requests.get(sign_url, headers=http_local.header, params={'gameId': game_id, 'uid': uid}).json() print(resp) def do_sign(cred_resp): - http_local.token = cred_resp['token'] - http_local.header = header.copy() - http_local.header['cred'] = cred_resp['cred'] + setup_cred_session(cred_resp) characters = get_binding_list() for i in characters: body = { - 'gameId': 1, + 'gameId': ARKNIGHTS_GAME_ID, 'uid': i.get('uid') } # list_awards(1, i.get('uid')) resp = requests.post(sign_url, headers=get_sign_header(sign_url, 'post', body, http_local.header), json=body).json() - if resp['code'] != 0: - print(f'角色{i.get("nickName")}({i.get("channelName")})签到失败了！原因：{resp.get("message")}') - continue - awards = resp['data']['awards'] - for j in awards: - res = j['resource'] - print( - f'角色{i.get("nickName")}({i.get("channelName")})签到成功，获得了{res["name"]}×{j.get("count") or 1}' - ) + _, msg = format_arknights_sign_message(i, resp) + print(msg) def single_sign(cred_resp): - http_local.token = cred_resp['token'] - http_local.header = header.copy() - http_local.header['cred'] = cred_resp['cred'] + setup_cred_session(cred_resp) characters = get_binding_list() - code, msg = -1, 'Default Error' + if not characters: + return '未绑定明日方舟角色，已跳过', 0 + + code = 0 + messages = [] for i in characters: body = { - 'gameId': 1, + 'gameId': ARKNIGHTS_GAME_ID, 'uid': i.get('uid') } # list_awards(1, i.get('uid')) resp = requests.post(sign_url, headers=get_sign_header(sign_url, 'post', body, http_local.header), json=body).json() - if resp['code'] != 0: - msg = f'角色{i.get("nickName")}({i.get("channelName")})签到失败了！原因：{resp.get("message")}' + 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 - awards = resp['data']['awards'] - for j in awards: - res = j['resource'] - msg = f'角色{i.get("nickName")}({i.get("channelName")})签到成功，获得了{res["name"]}×{j.get("count") or 1}' - code = 0 - return msg, code + + 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): diff --git a/src/ark_helper_backend/backend.py b/src/ark_helper_backend/backend.py index 7124846..837d082 100644 --- a/src/ark_helper_backend/backend.py +++ b/src/ark_helper_backend/backend.py @@ -10,7 +10,7 @@ from pathlib import Path from src.util.utils import hg_get_new_user_token, load_user_config, validate_token from src.My_Logger.project_logger import LoggerClass -from src.SKland_Auto_Sign.new_auto_sign import get_cred_by_token, single_sign +from src.SKland_Auto_Sign.new_auto_sign import get_cred_by_token, single_endfield_sign, single_sign PROJECT_PATH = Path(__file__).parent.parent.parent.absolute() DATA_PATH = os.path.join(PROJECT_PATH, 'data', 'ArkHelperData') @@ -137,17 +137,28 @@ class ArkHelperBackend: self.single_user_login(phone) self.save_data(payment_status=False, gacha_status=False) - def skland_sign_new(self): + def _run_sign_jobs(self, sign_jobs): for user, content in self.user_data_dict.items(): token = content.get('token') if token: try: - msg, code = single_sign(get_cred_by_token(token)) - self.logger.info(f'{msg}: {code}') + cred_resp = get_cred_by_token(token) + for sign_name, sign_handler in sign_jobs: + msg, code = sign_handler(cred_resp) + self.logger.info(f'{user} {sign_name}签到结果: {msg} [{code}]') except Exception as ex: self.logger.error(f'签到失败，原因：{str(ex)}') time.sleep(random.randint(100, 200)) + def skland_sign_new(self): + self._run_sign_jobs((('明日方舟', single_sign),)) + + def endfield_sign_new(self): + self._run_sign_jobs((('终末地', single_endfield_sign),)) + + def daily_sign_new(self): + self._run_sign_jobs((('明日方舟', single_sign), ('终末地', single_endfield_sign))) + def update_gacha_data(self): for phone in self.user_data_dict: gacha_history_data = self.user_data_dict[phone]['gacha'] @@ -209,7 +220,7 @@ if __name__ == '__main__': # ak_back = ArkHelperBackend() # ak_back.update_gacha_data() # ak_back.update_payment_data() - # ak_back.skland_sign_new() + # ak_back.daily_sign_new() login_url = 'https://ak.hypergryph.com/user/login' session = requests.session()