✨ 给签到添加ajax请求或打码平台

core/config.py

@@ -12,7 +12,7 @@ from pydantic import (
     BaseSettings,
 )
 
-__all__ = ['BotConfig', 'config']
+__all__ = ["BotConfig", "config"]
 
 from utils.const import PROJECT_ROOT
 
@@ -38,17 +38,19 @@ class BotConfig(BaseSettings):
     api_id: Optional[int]
     api_hash: Optional[str]
 
-    channels: List['ConfigChannel'] = []
-    admins: List['ConfigUser'] = []
+    channels: List["ConfigChannel"] = []
+    admins: List["ConfigUser"] = []
     verify_groups: List[Union[int, str]] = []
 
     logger_width: int = 180
-    logger_log_path: str = './logs'
+    logger_log_path: str = "./logs"
     logger_time_format: str = "[%Y-%m-%d %X]"
     logger_traceback_max_frames: int = 20
-    logger_render_keywords: List[str] = ['BOT']
+    logger_render_keywords: List[str] = ["BOT"]
 
     enka_network_api_agent: str = ""
+    pass_challenge_api: str = ""
+    pass_challenge_app_key: str = ""
 
     class Config:
         case_sensitive = False
@@ -110,7 +112,7 @@ class MySqlConfig(BaseModel):
 
 
 class RedisConfig(BaseModel):
-    host: str = '127.0.0.1'
+    host: str = "127.0.0.1"
     port: int
     database: int = 0
 
@@ -119,8 +121,8 @@ class LoggerConfig(BaseModel):
     width: int = 180
     time_format: str = "[%Y-%m-%d %X]"
     traceback_max_frames: int = 20
-    path: Path = PROJECT_ROOT / 'logs'
-    render_keywords: List[str] = ['BOT']
+    path: Path = PROJECT_ROOT / "logs"
+    render_keywords: List[str] = ["BOT"]
 
 
 class MTProtoConfig(BaseModel):

plugins/genshin/sign.py

@@ -1,13 +1,18 @@
 import datetime
 import time
+from json import JSONDecodeError
+from typing import Optional, Dict
 
 from genshin import Game, GenshinException, AlreadyClaimed, Client
+from httpx import AsyncClient, Timeout
 from telegram import Update
+from telegram.constants import ChatAction
 from telegram.ext import CommandHandler, CallbackContext
 from telegram.ext import MessageHandler, filters
 
 from core.admin.services import BotAdminService
 from core.baseplugin import BasePlugin
+from core.config import config
 from core.cookies.error import CookiesNotFoundError
 from core.cookies.services import CookiesService
 from core.plugin import Plugin, handler
@@ -27,13 +32,105 @@ class Sign(Plugin, BasePlugin):
 
     CHECK_SERVER, COMMAND_RESULT = range(10400, 10402)
 
-    def __init__(self, user_service: UserService = None, cookies_service: CookiesService = None,
-                 sign_service: SignServices = None, bot_admin_service: BotAdminService = None):
+    def __init__(
+        self,
+        user_service: UserService = None,
+        cookies_service: CookiesService = None,
+        sign_service: SignServices = None,
+        bot_admin_service: BotAdminService = None,
+    ):
         self.bot_admin_service = bot_admin_service
         self.cookies_service = cookies_service
         self.user_service = user_service
         self.sign_service = sign_service
 
+    @staticmethod
+    async def pass_challenge(gt: str, challenge: str, referer: str = None) -> Optional[Dict]:
+        """尝试自动通过验证，感谢 @coolxitech 大佬提供的方案
+
+        https://github.com/coolxitech/mihoyo
+        """
+        if not gt or not challenge:
+            return None
+        if not referer:
+            referer = (
+                "https://webstatic.mihoyo.com/bbs/event/signin-ys/index.html?"
+                "bbs_auth_required=true&act_id=e202009291139501&utm_source=bbs&utm_medium=mys&utm_campaign=icon"
+            )
+        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.37.1",
+            "Referer": referer,
+            "Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7",
+        }
+        # ajax auto pass
+        async with AsyncClient() as client:
+            try:
+                # gt={gt}&challenge={challenge}&lang=zh-cn&pt=3
+                # client_type=web_mobile&callback=geetest_{int(time.time() * 1000)}
+                req = await client.get(
+                    "https://api.geetest.com/ajax.php",
+                    params={
+                        "gt": gt,
+                        "challenge": challenge,
+                        "lang": "zh-cn",
+                        "pt": 3,
+                        "client_type": "web_mobile",
+                        "callback": f"geetest_{int(time.time() * 1000)}",
+                    },
+                    headers=header,
+                    timeout=20,
+                )
+                logger.info(f"ajax 返回：{req.text}")
+                if req.status_code != 200:
+                    raise RuntimeError
+                data = req.json()
+                if "success" in data["status"] and "success" in data["data"]["result"]:
+                    return {
+                        "x-rpc-challenge": challenge,
+                        "x-rpc-validate": data["data"]["validate"],
+                        "x-rpc-seccode": f'{data["data"]["validate"]}|jordan',
+                    }
+            except (
+                JSONDecodeError,
+                KeyError,
+                Timeout,
+                RuntimeError,
+            ) as exc:
+                logger.warning(f"ajax 自动通过失败：{repr(exc)}")
+        logger.warning("ajax 自动通过失败")
+        if not config.pass_challenge_api:
+            return None
+        pass_challenge_params = {
+            "gt": gt,
+            "challenge": challenge,
+            "referer": referer,
+        }
+        if config.pass_challenge_app_key:
+            pass_challenge_params["appkey"] = config.pass_challenge_app_key
+        # custom api auto pass
+        async with AsyncClient() as client:
+            try:
+                resp = await client.post(
+                    config.pass_challenge_api,
+                    params=pass_challenge_params,
+                    timeout=45,
+                )
+                logger.info(f"签到自定义打码平台返回：{resp.text}")
+                data = resp.json()
+                if data["code"] != 0:
+                    raise RuntimeError
+                return {
+                    "x-rpc-challenge": data["data"]["challenge"],
+                    "x-rpc-validate": data["data"]["validate"],
+                    "x-rpc-seccode": f'{data["data"]["validate"]}|jordan',
+                }
+            except (JSONDecodeError, KeyError, Timeout, RuntimeError) as exc:
+                logger.warning(f"签到自定义打码平台自动通过失败：{repr(exc)}")
+        return None
+
     @staticmethod
     async def _start_sign(client: Client) -> str:
         try:
@@ -48,11 +145,29 @@ class Sign(Plugin, BasePlugin):
             return f"获取签到状态失败，API返回信息为 {str(error)}"
         if not daily_reward_info.signed_in:
             try:
-                request_daily_reward = await client.request_daily_reward("sign", method="POST",
-                                                                         game=Game.GENSHIN, lang="zh-cn")
+                request_daily_reward = await client.request_daily_reward(
+                    "sign", method="POST", game=Game.GENSHIN, lang="zh-cn"
+                )
                 if request_daily_reward and request_daily_reward.get("success", 0) == 1:
-                    logger.warning(f"UID {client.uid} 签到失败，触发验证码风控")
-                    return f"UID {client.uid} 签到失败，触发验证码风控，请尝试重新签到。"
+                    # 米游社国内签到自动打码
+                    headers = await Sign.pass_challenge(
+                        request_daily_reward.get("gt", ""),
+                        request_daily_reward.get("challenge", ""),
+                    )
+                    if not headers:
+                        logger.warning(f"UID {client.uid} 签到失败，触发验证码风控")
+                        return f"UID {client.uid} 签到失败，触发验证码风控，请尝试重新签到。"
+                    request_daily_reward = await client.request_daily_reward(
+                        "sign",
+                        method="POST",
+                        game=Game.GENSHIN,
+                        lang="zh-cn",
+                        headers=headers,
+                    )
+                    if request_daily_reward and request_daily_reward.get("success", 0) == 1:
+                        logger.warning(f"UID {client.uid} 签到失败，触发验证码风控")
+                        return f"UID {client.uid} 签到失败，触发验证码风控，请尝试重新签到。"
+                    logger.info(f"UID {client.uid} 通过自动打码签到成功")
             except AlreadyClaimed:
                 logger.info(f"UID {client.uid} 已经签到")
                 result = "今天旅行者已经签到过了~"
@@ -73,11 +188,13 @@ class Sign(Plugin, BasePlugin):
         missed_days = now.day - daily_reward_info.claimed_rewards
         if not daily_reward_info.signed_in:
             missed_days -= 1
-        message = f"#### {today} (UTC+8) ####\n" \
-                  f"UID: {client.uid}\n" \
-                  f"今日奖励: {reward.name} × {reward.amount}\n" \
-                  f"本月漏签次数：{missed_days}\n" \
-                  f"签到结果: {result}"
+        message = (
+            f"#### {today} (UTC+8) ####\n"
+            f"UID: {client.uid}\n"
+            f"今日奖励: {reward.name} × {reward.amount}\n"
+            f"本月漏签次数：{missed_days}\n"
+            f"签到结果: {result}"
+        )
         return message
 
     async def _process_auto_sign(self, user_id: int, chat_id: int, method: str) -> str:
@@ -99,8 +216,12 @@ class Sign(Plugin, BasePlugin):
         elif method == "关闭":
             return "您还没有开启自动签到"
         elif method == "开启":
-            user = SignUser(user_id=user_id, chat_id=chat_id, time_created=datetime.datetime.now(),
-                            status=SignStatusEnum.STATUS_SUCCESS)
+            user = SignUser(
+                user_id=user_id,
+                chat_id=chat_id,
+                time_created=datetime.datetime.now(),
+                status=SignStatusEnum.STATUS_SUCCESS,
+            )
             await self.sign_service.add(user)
             return "开启自动签到成功"
 
@@ -134,6 +255,7 @@ class Sign(Plugin, BasePlugin):
             self._add_delete_message_job(context, message.chat_id, message.message_id)
         try:
             client = await get_genshin_client(user.id)
+            await message.reply_chat_action(ChatAction.TYPING)
             sign_text = await self._start_sign(client)
             reply_message = await message.reply_text(sign_text, allow_sending_without_reply=True)
             if filters.ChatType.GROUPS.filter(reply_message):

plugins/jobs/sign.py

@@ -13,6 +13,7 @@ from core.plugin import Plugin, job
 from core.sign.models import SignStatusEnum
 from core.sign.services import SignServices
 from core.user import UserService
+from plugins.genshin.sign import Sign
 from utils.helpers import get_genshin_client
 from utils.log import logger
 
@@ -22,13 +23,64 @@ class NeedChallenge(Exception):
 
 
 class SignJob(Plugin):
-
-    def __init__(self, sign_service: SignServices = None, user_service: UserService = None,
-                 cookies_service: CookiesService = None):
+    def __init__(
+        self,
+        sign_service: SignServices = None,
+        user_service: UserService = None,
+        cookies_service: CookiesService = None,
+    ):
         self.sign_service = sign_service
         self.cookies_service = cookies_service
         self.user_service = user_service
 
+    @staticmethod
+    async def single_sign(user_id: int) -> str:
+        client = await get_genshin_client(user_id)
+        rewards = await client.get_monthly_rewards(game=Game.GENSHIN, lang="zh-cn")
+        daily_reward_info = await client.get_reward_info(game=Game.GENSHIN)
+        if not daily_reward_info.signed_in:
+            request_daily_reward = await client.request_daily_reward("sign", method="POST", game=Game.GENSHIN)
+            if request_daily_reward and request_daily_reward.get("success", 0) == 1:
+                # 米游社国内签到自动打码
+                headers = await Sign.pass_challenge(
+                    request_daily_reward.get("gt", ""),
+                    request_daily_reward.get("challenge", ""),
+                )
+                if not headers:
+                    logger.warning(f"UID {client.uid} 签到失败，触发验证码风控 | 打码平台打码失败，请检查")
+                    raise NeedChallenge
+                request_daily_reward = await client.request_daily_reward(
+                    "sign",
+                    method="POST",
+                    game=Game.GENSHIN,
+                    lang="zh-cn",
+                    headers=headers,
+                )
+                if request_daily_reward and request_daily_reward.get("success", 0) == 1:
+                    logger.warning(f"UID {client.uid} 签到失败，触发验证码风控 | 打码平台打码失败，请检查")
+                    raise NeedChallenge
+                logger.info(f"UID {client.uid} 签到请求 {request_daily_reward} | 通过自动打码签到成功")
+            else:
+                logger.info(f"UID {client.uid} 签到请求 {request_daily_reward}")
+            result = "OK"
+        else:
+            result = "今天旅行者已经签到过了~"
+        reward = rewards[daily_reward_info.claimed_rewards - (1 if daily_reward_info.signed_in else 0)]
+        today = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
+        cn_timezone = datetime.timezone(datetime.timedelta(hours=8))
+        now = datetime.datetime.now(cn_timezone)
+        missed_days = now.day - daily_reward_info.claimed_rewards
+        if not daily_reward_info.signed_in:
+            missed_days -= 1
+        return (
+            f"########### 定时签到 ###########\n"
+            f"#### {today} (UTC+8) ####\n"
+            f"UID: {client.uid}\n"
+            f"今日奖励: {reward.name} × {reward.amount}\n"
+            f"本月漏签次数：{missed_days}\n"
+            f"签到结果: {result}"
+        )
+
     @job.run_daily(time=datetime.time(hour=0, minute=1, second=0), name="SignJob")
     async def sign(self, context: CallbackContext):
         logger.info("正在执行自动签到")
@@ -38,32 +90,7 @@ class SignJob(Plugin):
                 continue
             user_id = sign_db.user_id
             try:
-                client = await get_genshin_client(user_id)
-                rewards = await client.get_monthly_rewards(game=Game.GENSHIN, lang="zh-cn")
-                daily_reward_info = await client.get_reward_info(game=Game.GENSHIN)
-                if not daily_reward_info.signed_in:
-                    request_daily_reward = await client.request_daily_reward("sign", method="POST", game=Game.GENSHIN)
-                    if request_daily_reward and request_daily_reward.get("success", 0) == 1:
-                        logger.warning(f"UID {client.uid} 签到失败，触发验证码风控")
-                        raise NeedChallenge
-                    else:
-                        logger.info(f"UID {client.uid} 签到请求 {request_daily_reward}")
-                        result = "OK"
-                else:
-                    result = "今天旅行者已经签到过了~"
-                reward = rewards[daily_reward_info.claimed_rewards - (1 if daily_reward_info.signed_in else 0)]
-                today = time.strftime("%Y-%m-%d %H:%M:%S", time.localtime())
-                cn_timezone = datetime.timezone(datetime.timedelta(hours=8))
-                now = datetime.datetime.now(cn_timezone)
-                missed_days = now.day - daily_reward_info.claimed_rewards
-                if not daily_reward_info.signed_in:
-                    missed_days -= 1
-                text = f"########### 定时签到 ###########\n" \
-                       f"#### {today} (UTC+8) ####\n" \
-                       f"UID: {client.uid}\n" \
-                       f"今日奖励: {reward.name} × {reward.amount}\n" \
-                       f"本月漏签次数：{missed_days}\n" \
-                       f"签到结果: {result}"
+                text = await self.single_sign(user_id)
             except InvalidCookies:
                 text = "自动签到执行失败，Cookie无效"
                 sign_db.status = SignStatusEnum.INVALID_COOKIES
@@ -84,7 +111,7 @@ class SignJob(Plugin):
                 logger.exception(exc)
                 text = "签到失败了呜呜呜 ~ 执行自动签到时发生错误"
             if sign_db.chat_id < 0:
-                text = f"<a href=\"tg://user?id={sign_db.user_id}\">NOTICE {sign_db.user_id}</a>\n\n{text}"
+                text = f'<a href="tg://user?id={sign_db.user_id}">NOTICE {sign_db.user_id}</a>\n\n{text}'
             try:
                 await context.bot.send_message(sign_db.chat_id, text, parse_mode=ParseMode.HTML)
                 await asyncio.sleep(5)  # 回复延迟5S避免触发洪水防御

plugins/system/sign_all.py

@@ -0,0 +1,85 @@
+import datetime
+import asyncio
+
+from aiohttp import ClientConnectorError
+from genshin import InvalidCookies, AlreadyClaimed, GenshinException
+from telegram import Update
+from telegram.constants import ParseMode
+from telegram.error import BadRequest, Forbidden
+from telegram.ext import CommandHandler, CallbackContext
+
+from core.cookies import CookiesService
+from core.plugin import Plugin, handler
+from core.sign import SignServices
+from core.sign.models import SignStatusEnum
+from core.user import UserService
+from plugins.jobs.sign import NeedChallenge, SignJob
+from utils.decorators.admins import bot_admins_rights_check
+from utils.log import logger
+
+
+class SignAll(Plugin):
+    def __init__(
+        self,
+        sign_service: SignServices = None,
+        user_service: UserService = None,
+        cookies_service: CookiesService = None,
+    ):
+        self.sign_service = sign_service
+        self.cookies_service = cookies_service
+        self.user_service = user_service
+
+    @handler(CommandHandler, command="sign_all", block=False)
+    @bot_admins_rights_check
+    async def sign_all(self, update: Update, context: CallbackContext):
+        user = update.effective_user
+        logger.info(f"用户 {user.full_name}[{user.id}] sign_all 命令请求")
+        message = update.effective_message
+        reply = await message.reply_text("正在全部重新签到，请稍后...")
+        sign_list = await self.sign_service.get_all()
+        for sign_db in sign_list:
+            user_id = sign_db.user_id
+            old_status = sign_db.status
+            try:
+                text = await SignJob.single_sign(user_id)
+            except InvalidCookies:
+                text = "自动签到执行失败，Cookie无效"
+                sign_db.status = SignStatusEnum.INVALID_COOKIES
+            except AlreadyClaimed:
+                text = "今天旅行者已经签到过了~"
+                sign_db.status = SignStatusEnum.ALREADY_CLAIMED
+            except GenshinException as exc:
+                text = f"自动签到执行失败，API返回信息为 {str(exc)}"
+                sign_db.status = SignStatusEnum.GENSHIN_EXCEPTION
+            except ClientConnectorError:
+                text = "签到失败了呜呜呜 ~ 服务器连接超时 服务器熟啦 ~ "
+                sign_db.status = SignStatusEnum.TIMEOUT_ERROR
+            except NeedChallenge:
+                text = "签到失败，触发验证码风控，自动签到自动关闭"
+                sign_db.status = SignStatusEnum.NEED_CHALLENGE
+            except BaseException as exc:
+                logger.error(f"执行自动签到时发生错误 用户UID[{user_id}]")
+                logger.exception(exc)
+                text = "签到失败了呜呜呜 ~ 执行自动签到时发生错误"
+            if sign_db.chat_id < 0:
+                text = f'<a href="tg://user?id={sign_db.user_id}">NOTICE {sign_db.user_id}</a>\n\n{text}'
+            try:
+                if "今天旅行者已经签到过了~" not in text:
+                    await context.bot.send_message(sign_db.chat_id, text, parse_mode=ParseMode.HTML)
+                    await asyncio.sleep(5)  # 回复延迟5S避免触发洪水防御
+            except BadRequest as exc:
+                logger.error(f"执行自动签到时发生错误 用户UID[{user_id}]")
+                logger.exception(exc)
+                sign_db.status = SignStatusEnum.BAD_REQUEST
+            except Forbidden as exc:
+                logger.error(f"执行自动签到时发生错误 用户UID[{user_id}]")
+                logger.exception(exc)
+                sign_db.status = SignStatusEnum.FORBIDDEN
+            except BaseException as exc:
+                logger.error(f"执行自动签到时发生错误 用户UID[{user_id}]")
+                logger.exception(exc)
+                continue
+            sign_db.time_updated = datetime.datetime.now()
+            if sign_db.status != old_status:
+                await self.sign_service.update(sign_db)
+        await reply.edit_text("全部账号重新签到完成")