新增终末地签到+更新gitignore

.gitignore

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

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

src/SKland_Auto_Sign/new_auto_sign.py

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

src/ark_helper_backend/backend.py

@@ -10,7 +10,7 @@ from pathlib import Path
 
 from src.util.utils import hg_get_new_user_token, load_user_config, validate_token
 from src.My_Logger.project_logger import LoggerClass
-from src.SKland_Auto_Sign.new_auto_sign import get_cred_by_token, single_sign
+from src.SKland_Auto_Sign.new_auto_sign import get_cred_by_token, single_endfield_sign, single_sign
 
 PROJECT_PATH = Path(__file__).parent.parent.parent.absolute()
 DATA_PATH = os.path.join(PROJECT_PATH, 'data', 'ArkHelperData')
@@ -137,17 +137,28 @@ class ArkHelperBackend:
             self.single_user_login(phone)
         self.save_data(payment_status=False, gacha_status=False)
 
-    def skland_sign_new(self):
+    def _run_sign_jobs(self, sign_jobs):
         for user, content in self.user_data_dict.items():
             token = content.get('token')
             if token:
                 try:
-                    msg, code = single_sign(get_cred_by_token(token))
+                    cred_resp = get_cred_by_token(token)
-                    self.logger.info(f'{msg}: {code}')
+                    for sign_name, sign_handler in sign_jobs:
+                        msg, code = sign_handler(cred_resp)
+                        self.logger.info(f'{user} {sign_name}签到结果: {msg} [{code}]')
                 except Exception as ex:
                     self.logger.error(f'签到失败，原因：{str(ex)}')
             time.sleep(random.randint(100, 200))
 
+    def skland_sign_new(self):
+        self._run_sign_jobs((('明日方舟', single_sign),))
+
+    def endfield_sign_new(self):
+        self._run_sign_jobs((('终末地', single_endfield_sign),))
+
+    def daily_sign_new(self):
+        self._run_sign_jobs((('明日方舟', single_sign), ('终末地', single_endfield_sign)))
+
     def update_gacha_data(self):
         for phone in self.user_data_dict:
             gacha_history_data = self.user_data_dict[phone]['gacha']
@@ -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()