为实时便签与每日签到接入验证码打码接口

.env.prod

@@ -5,3 +5,6 @@ SUPERUSERS=["123456"] # 超级用户
 NICKNAME=["派蒙", "bot"] # 机器人的昵称
 COMMAND_START=[""] # 命令前缀,根据需要自行修改
 COMMAND_SEP=[""] # 命令分隔符
+CAPTCHA_ENABLED=true#启用打码平台
+CAPTCHA_API_KEY="YOUR-API-KEY"#验证码打码平台api key
+CAPTCHA_API_ENDPOINT="YOUR-API-PLATFORM"#验证码打码平台api调用网址

LittlePaimon/plugins/Paimon_Autobbs/draw.py

@@ -11,12 +11,13 @@ class SignResult(IntEnum):
     SUCCESS = auto()
     DONE = auto()
     FAIL = auto()
+    INVALID = auto()
 
 
 async def draw_result(group_id: int, data: list):
     img = PMImage(await load_image(RESOURCE_BASE_PATH / 'player_card' / 'white_bg.png'))
     success_list = [d for d in data if d['result'] in [SignResult.SUCCESS, SignResult.DONE]]
-    fail_list = [d['uid'] for d in data if d['result'] == SignResult.FAIL]
+    fail_list = [d['uid'] for d in data if d['result'] == SignResult.FAIL or d['uid'] for d in data if d['result'] == SignResult.INVALID]
     same_reward = defaultdict(list)
     for s in success_list:
         same_reward[s['reward']].append(s['uid'])

LittlePaimon/plugins/Paimon_Autobbs/sign_handle.py

@@ -5,6 +5,7 @@
 from collections import defaultdict
 from typing import Tuple, Union
 
+import nonebot
 from nonebot import get_bot
 
 from LittlePaimon.config import config
@@ -15,12 +16,12 @@
 from .draw import SignResult, draw_result
 
 SIGN_ACTION_API = 'https://api-takumi.mihoyo.com/event/bbs_sign_reward/sign'
-GEETEST_HEADER = {"Accept":           "*/*",
+GEETEST_HEADER = {"Accept": "*/*",
                   "X-Requested-With": "com.mihoyo.hyperion",
-                  "User-Agent":       'Mozilla/5.0 (Linux; Android 12; Unspecified Device) AppleWebKit/537.36 (KHTML, like Gecko) '
-                                      'Version/4.0 Chrome/103.0.5060.129 Mobile Safari/537.36 miHoYoBBS/2.35.2',
-                  "Referer":          "https://webstatic.mihoyo.com/",
-                  "Accept-Language":  "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7"
+                  "User-Agent": 'Mozilla/5.0 (Linux; Android 12; Unspecified Device) AppleWebKit/537.36 (KHTML, like Gecko) '
+                                'Version/4.0 Chrome/103.0.5060.129 Mobile Safari/537.36 miHoYoBBS/2.35.2',
+                  "Referer": "https://webstatic.mihoyo.com/",
+                  "Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7"
                   }
 sign_reward_list: dict = {}
 
@@ -31,7 +32,7 @@ async def sign_action(user_id: str, uid: str) -> Union[dict, str]:
     resp = await aiorequests.post(SIGN_ACTION_API, headers=mihoyo_sign_headers(cookie_info.cookie),
                                   json={
                                       'act_id': 'e202009291139501',
-                                      'uid':    uid,
+                                      'uid': uid,
                                       'region': server_id
                                   })
     data = resp.json()
@@ -54,14 +55,15 @@ async def mhy_bbs_sign(user_id: str, uid: str) -> Tuple[SignResult, str]:
     if isinstance(sign_info, str):
         logger.info('米游社原神签到', '➤', {'用户': user_id, 'UID': uid}, '未绑定私人cookie或已失效', False)
         await MihoyoBBSSub.filter(user_id=user_id, uid=uid).delete()
-        return SignResult.FAIL, sign_info
+        return SignResult.INVALID, sign_info
     elif sign_info['data']['is_sign']:
         signed_days = sign_info['data']['total_sign_day'] - 1
         logger.info('米游社原神签到', '➤', {'用户': user_id, 'UID': uid}, '今天已经签过了', True)
         if sign_reward_list:
             return SignResult.DONE, f'UID{uid}今天已经签过了，获得的奖励为\n{sign_reward_list[signed_days]["name"]}*{sign_reward_list[signed_days]["cnt"]}'
         else:
             return SignResult.DONE, f'UID{uid}今天已经签过了'
+    message = None
     for i in range(3):
         sign_data = await sign_action(user_id, uid)
         if isinstance(sign_data, str):
@@ -87,13 +89,84 @@ async def mhy_bbs_sign(user_id: str, uid: str) -> Tuple[SignResult, str]:
                 else:
                     return SignResult.SUCCESS, '签到成功'
             else:
-                wait_time = random.randint(90, 120)
-                logger.info('米游社原神签到', '➤', {'用户': user_id, 'UID': uid}, f'出现验证码，等待{wait_time}秒后进行第{i + 1}次尝试绕过', False)
-                await asyncio.sleep(wait_time)
-    logger.info('米游社原神签到', '➤', {'用户': user_id, 'UID': uid}, '尝试3次签到失败，无法绕过验证码', False)
-    return SignResult.FAIL, f'{uid}签到失败，无法绕过验证码'
+                result, message = await geetest_verify(user_id=user_id, uid=uid, sign_data=sign_data)
+                if result == SignResult.INVALID:
+                    return SignResult.INVALID, f'{uid}签到失败，你的UID{uid}的cookie疑似失效了'
+                elif result == SignResult.FAIL:
+                    logger.warning('米游社原神签到', '➤', f'{uid}第{i}次签到失败，正在重试，报错信息：{message}')
+                elif result == SignResult.SUCCESS:
+                    return SignResult.SUCCESS, '签到成功'
+                else:
+                    logger.warning('米游社原神签到', '➤', f'{uid}第{i}次签到失败，正在重试，报错信息：{message}')
+    return SignResult.FAIL, f'{uid}签到失败，报错信息{message}'
 
 
+async def sign_action_after_geetest(user_id: str,
+                                    uid: str,
+                                    geetest_challenge: str,
+                                    geetest_validate: str,
+                                    geetest_seccode: str):
+    server_id = 'cn_qd01' if uid[0] == '5' else 'cn_gf01'
+    cookie_info = await PrivateCookie.get_or_none(user_id=user_id, uid=uid)
+    headers = mihoyo_sign_headers(cookie_info.cookie)
+    headers["x-rpc-validate"] = geetest_validate
+    headers["x-rpc-challenge"] = geetest_challenge
+    headers["x-rpc-seccode"] = geetest_seccode
+    resp = await aiorequests.post(url=SIGN_ACTION_API, headers=headers,
+                                  json={
+                                      'act_id': 'e202009291139501',
+                                      'uid': uid,
+                                      'region': server_id
+                                  })
+    data = resp.json()
+    if await check_retcode(data, cookie_info, user_id, uid):
+        return data
+    else:
+        return f'你的UID{uid}的cookie疑似失效了'
+
+
+async def geetest_verify(user_id: str, uid: str, sign_data: dict):
+    global_config = nonebot.get_driver().config
+    captcha_enable = global_config.captcha_enabled
+    superuser = global_config.superusers
+    if not captcha_enable:
+        logger.info('米游社原神签到➤',
+                    '未开启验证码绕过，请添加captcha_api_key并根据需求修改geetest_verify(), sign_action_after_geetest(), '
+                    'geetest_handle_init(), 以及geetest_handle_finish()函数')
+        if user_id not in superuser:
+            return SignResult.FAIL, f'{uid}签到失败，请联系管理员检查后台日志并开启验证码绕过'
+        else:
+            return SignResult.FAIL, f'{uid}签到失败，请检查后台日志开启验证码绕过'
+    api_key = global_config.captcha_api_key
+    api_endpoint = global_config.captcha_api_endpoint
+    gt = sign_data['data']['gt']
+    challenge = sign_data['data']['challenge']
+    page_url = f'https://webstatic.mihoyo.com/bbs/event/signin-ys/index.html?bbs_auth_required=true&act_id' \
+               f'=e202009291139501&ltuid={uid}'
+    token, err_message = await geetest_handle_init(api_key=api_key, gt=gt, challenge=challenge,
+                                                  page_url=page_url, api_endpoint=api_endpoint)
+    if err_message == "":
+        sign_result = await sign_action_after_geetest(user_id=user_id,
+                                                      uid=uid,
+                                                      geetest_challenge=token['challenge'],
+                                                      geetest_seccode=token['seccode'],
+                                                      geetest_validate=token['validate'])
+        if sign_result == f'你的UID{uid}的cookie疑似失效了':
+            return SignResult.INVALID, f'你的UID{uid}的cookie疑似失效了'
+        else:
+            return SignResult.SUCCESS, '签到成功'
+    else:
+        return SignResult.FAIL, f'{uid}平台返回错误{err_message}'
+
+
+async def geetest_handle_init(api_key: str,
+                              api_endpoint: str,
+                              gt: str,
+                              challenge: str,
+                              page_url: str):
+    # TODO: Please modify this function according to the captcha resolver you are using
+    raise NotImplementedError
+
 @scheduler.scheduled_job('cron', hour=config.auto_sign_hour, minute=config.auto_sign_minute,
                          misfire_grace_time=10)
 async def _():
@@ -111,7 +184,8 @@ async def bbs_auto_sign():
     if not subs:
         # 如果没有米游社原神签到订阅，则不执行签到任务
         return
-    logger.info('米游社原神签到', f'开始执行米游社自动签到，共<m>{len(subs)}</m>个任务，预计花费<m>{len(subs) * 2}</m>分钟')
+    logger.info('米游社原神签到',
+                f'开始执行米游社自动签到，共<m>{len(subs)}</m>个任务，预计花费<m>{len(subs) * 2}</m>分钟')
     sign_result_group = defaultdict(list)
     sign_result_private = defaultdict(list)
     for sub in subs:
@@ -120,13 +194,13 @@ async def bbs_auto_sign():
         if sub.user_id != sub.group_id:
             sign_result_group[sub.group_id].append({
                 'user_id': sub.user_id,
-                'uid':     sub.uid,
-                'result':  result,
-                'reward':  msg.split('\n')[-1] if result in [SignResult.SUCCESS, SignResult.DONE] else ''
+                'uid': sub.uid,
+                'result': result,
+                'reward': msg.split('\n')[-1] if result in [SignResult.SUCCESS, SignResult.DONE] else ''
             })
         else:
             sign_result_private[sub.user_id].append({
-                'uid':    sub.uid,
+                'uid': sub.uid,
                 'result': result,
                 'reward': msg.split('\n')[-1] if result in [SignResult.SUCCESS, SignResult.DONE] else ''
             })