Merge pull request #290 from Ljzd-PRO/dev

更新至 v2.4.0

README.md

@@ -12,23 +12,6 @@
 
 # mysTool - 米游社辅助工具插件
 
-## 📣 更新内容
-
-> [!IMPORTANT]
-> v2.0.0 突破性更新
-> https://github.com/Ljzd-PRO/nonebot-plugin-mystool/releases/tag/v2.0.0
-
-### 2024.3.8 - v2.3.0
-
-#### 💡 新特性
-- 增加微博超话兑换码推送功能 - by @Joseandluue (#272)
-- 更改每日任务执行顺序。先执行游戏签到，再执行米游币任务，以降低执行米游币任务时出现人机验证的概率 - by @Sakamakiiizayoi @Joseandluue
-
-#### 🐛 Bug 修复
-- 修复每日任务自动进行游戏签到后，QQ聊天的主动私信推送失败的问题 (#270)
-- 更改 `UserAccount.mission_games`（用户米游币任务目标分区） 默认值为 `["BBSMission"]`，并在执行时检查该配置是否未空 (#261)
-- 修复可能出现的启动失败的问题（`AttributeError: 'NoneType' object has no attribute 'metadata'`） (#271)
-
 ## ⚡ 功能和特性
 
 - 支持QQ聊天和QQ频道

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "nonebot-plugin-mystool"
-version = "v2.3.0"
+version = "v2.4.0"
 description = "QQ聊天、频道机器人插件 | 米游社工具-每日米游币任务、游戏签到、商品兑换、免抓包登录、原神崩铁便笺提醒"
 license = "MIT"
 authors = [

src/nonebot_plugin_mystool/__init__.py

@@ -15,6 +15,7 @@
     "\n🎁 {HEAD}商品 ➢ 查看米游币商品信息(商品ID)"
     "\n📊 {HEAD}原神便笺 ➢ 查看原神实时便笺(原神树脂、洞天财瓮等)"
     "\n📊 {HEAD}铁道便笺 ➢ 查看星穹铁道实时便笺(开拓力、每日实训等)"
+    "\n👁️‍🗨️ {HEAD}wb兑换 ➢ 查看微博本期超话签到的兑换码(暂支持原神和星穹铁道)"
     "\n⚙️ {HEAD}设置 ➢ 设置是否开启通知、每日任务等相关选项"
     "\n🔑 {HEAD}账号设置 ➢ 设置设备平台、是否开启每日计划任务、频道任务"
     "\n🔔 {HEAD}通知设置 ➢ 设置是否开启每日米游币任务、游戏签到的结果通知"

src/nonebot_plugin_mystool/_version.py

@@ -1 +1 @@
-__version__ = "v2.3.0"
+__version__ = "v2.4.0"

src/nonebot_plugin_mystool/api/common.py

@@ -1,6 +1,7 @@
+import json
 import time
 from typing import List, Optional, Tuple, Dict, Any, Union, Type
-from urllib.parse import urlencode
+from urllib.parse import urlencode, urlparse, parse_qs
 
 import httpx
 import tenacity
@@ -11,7 +12,7 @@
     GetCookieStatus, \
     CreateMobileCaptchaStatus, GetGoodDetailStatus, ExchangeStatus, GeetestResultV4, GenshinNote, GenshinNoteStatus, \
     GetFpStatus, StarRailNoteStatus, StarRailNote, UserAccount, BBSCookies, ExchangePlan, ExchangeResult, plugin_env, \
-    plugin_config
+    plugin_config, QueryGameTokenQrCodeStatus
 from ..utils import generate_device_id, logger, generate_ds, \
     get_async_retry, generate_seed_id, generate_fp_locally
 
@@ -47,6 +48,10 @@
 URL_STARRAIL_NOTE_WIDGET = "https://api-takumi-record.mihoyo.com/game_record/app/hkrpg/aapi/widget"
 URL_CREATE_VERIFICATION = "https://bbs-api.miyoushe.com/misc/api/createVerification?is_high=true"
 URL_VERIFY_VERIFICATION = "https://bbs-api.miyoushe.com/misc/api/verifyVerification"
+URL_FETCH_GAME_TOKEN_QRCODE = "https://hk4e-sdk.mihoyo.com/hk4e_cn/combo/panda/qrcode/fetch"
+URL_QUERY_GAME_TOKEN_QRCODE = "https://hk4e-sdk.mihoyo.com/hk4e_cn/combo/panda/qrcode/query"
+URL_GET_TOKEN_BY_GAME_TOKEN = "https://api-takumi.mihoyo.com/account/ma-cn-session/app/getTokenByGameToken"
+URL_GET_COOKIE_TOKEN_BY_GAME_TOKEN = "https://api-takumi.mihoyo.com/auth/api/getCookieAccountInfoByGameToken"
 
 HEADERS_WEBAPI = {
     "Host": "webapi.account.mihoyo.com",
@@ -337,9 +342,9 @@ def __init__(self, content: Dict[str, Any]):
         self.data = self.content.get("data")
 
         for key in ["retcode", "status"]:
-            if not self.retcode:
+            if self.retcode is None:
                 self.retcode = self.content.get(key)
-                if not self.retcode:
+                if self.retcode is None:
                     self.retcode = self.data.get(key) if self.data else None
 
         self.message: Optional[str] = None
@@ -1679,3 +1684,190 @@ async def verify_verification(
         else:
             logger.exception("验证人机验证结果(verify_verification) - 请求失败")
             return BaseApiStatus(network_error=True)
+
+
+async def fetch_game_token_qrcode(
+        device_id: str,
+        app_id: str = "1",
+        retry: bool = True
+) -> Tuple[BaseApiStatus, Optional[Tuple[str, str]]]:
+    """
+    获取米游社扫码登录（GameToken）二维码
+
+    :param device_id: 设备ID
+    :param app_id: 登录的应用标识符
+    :param retry: 是否允许重试
+    :return 其中 ``Tuple[str, str]`` 为二维码URL和用于查询二维码扫描状态的 ``token``
+    """
+    try:
+        async for attempt in get_async_retry(retry):
+            with attempt:
+                content = {
+                    "app_id": app_id,
+                    "device": device_id,
+                }
+                async with httpx.AsyncClient() as client:
+                    res = await client.post(
+                        URL_FETCH_GAME_TOKEN_QRCODE,
+                        json=content,
+                        timeout=plugin_config.preference.timeout
+                    )
+                api_result = ApiResultHandler(res.json())
+                if api_result.retcode == 0:
+                    qrcode_url = api_result.data["url"]
+                    url = urlparse(qrcode_url)
+                    return BaseApiStatus(success=True), (qrcode_url, parse_qs(url.query)["ticket"][0])
+                else:
+                    logger.debug(f"网络请求返回: {res.text}")
+                    return BaseApiStatus(), None
+    except tenacity.RetryError as e:
+        if is_incorrect_return(e):
+            logger.exception("获取米游社扫码登录(fetch_game_token_qrcode) - 服务器没有正确返回")
+            logger.debug(f"网络请求返回: {res.text}")
+            return BaseApiStatus(incorrect_return=True), None
+        else:
+            logger.exception("获取米游社扫码登录(fetch_game_token_qrcode) - 请求失败")
+            return BaseApiStatus(network_error=True), None
+
+
+async def query_game_token_qrcode(
+        ticket: str,
+        device_id: str,
+        app_id: str = "1",
+        retry: bool = True
+) -> Tuple[QueryGameTokenQrCodeStatus, Optional[Tuple[str, str]]]:
+    """
+    查询米游社扫码登录（GameToken）二维码扫描状态
+
+    :param ticket: 生成二维码时返回的 URL 参数中 ``ticket`` 字段的值
+    :param device_id: 设备ID
+    :param app_id: 登录的应用标识符
+    :param retry: 是否允许重试
+    :return 其中 ``Tuple[str, str]`` 为米游社账号ID和 GameToken
+    """
+    try:
+        async for attempt in get_async_retry(retry):
+            with attempt:
+                content = {
+                    "app_id": app_id,
+                    "device": device_id,
+                    "ticket": ticket
+                }
+                async with httpx.AsyncClient() as client:
+                    res = await client.post(
+                        URL_QUERY_GAME_TOKEN_QRCODE,
+                        json=content,
+                        timeout=plugin_config.preference.timeout
+                    )
+                api_result = ApiResultHandler(res.json())
+                if api_result.retcode == 0:
+                    if api_result.data["stat"] == "Init":
+                        return QueryGameTokenQrCodeStatus(qrcode_init=True), None
+                    elif api_result.data["stat"] == "Scanned":
+                        return QueryGameTokenQrCodeStatus(qrcode_scanned=True), None
+                    else:
+                        payload_raw = api_result.data["payload"]["raw"]
+                        parsed_payload: Dict[str, str] = json.loads(payload_raw)
+                        return QueryGameTokenQrCodeStatus(success=True), (
+                            parsed_payload["uid"],
+                            parsed_payload["token"]
+                        )
+                elif api_result.retcode == -106:
+                    return QueryGameTokenQrCodeStatus(qrcode_expired=True), None
+                else:
+                    return QueryGameTokenQrCodeStatus(), None
+    except tenacity.RetryError as e:
+        if is_incorrect_return(e):
+            logger.exception("查询米游社扫码登录(query_game_token_qrcode) - 服务器没有正确返回")
+            logger.debug(f"网络请求返回: {res.text}")
+            return QueryGameTokenQrCodeStatus(incorrect_return=True), None
+        else:
+            logger.exception("查询米游社扫码登录(query_game_token_qrcode) - 请求失败")
+            return QueryGameTokenQrCodeStatus(network_error=True), None
+
+
+async def get_token_by_game_token(
+        bbs_uid: str,
+        game_token: str,
+        retry: bool = True
+) -> Tuple[BaseApiStatus, Optional[BBSCookies]]:
+    """
+    通过 GameToken 获取 STokenV2 和 mid
+
+    :param bbs_uid: 米游社账号 UID
+    :param game_token: 有效的 GameToken
+    :param retry: 是否允许重试
+    """
+    try:
+        async for attempt in get_async_retry(retry):
+            with attempt:
+                content = {
+                    "account_id": int(bbs_uid),
+                    "game_token": game_token
+                }
+                async with httpx.AsyncClient() as client:
+                    res = await client.post(
+                        URL_GET_TOKEN_BY_GAME_TOKEN,
+                        headers={"x-rpc-app_id": "bll8iq97cem8"},
+                        json=content,
+                        timeout=plugin_config.preference.timeout
+                    )
+                api_result = ApiResultHandler(res.json())
+                if api_result.retcode == 0:
+                    stoken_v2 = api_result.data["token"]["token"]
+                    mid = api_result.data["user_info"]["mid"]
+                    return BaseApiStatus(success=True), BBSCookies(stoken_v2=stoken_v2, mid=mid)
+                else:
+                    logger.debug(f"网络请求返回: {res.text}")
+                    return BaseApiStatus(), None
+    except tenacity.RetryError as e:
+        if is_incorrect_return(e):
+            logger.exception("通过 GameToken 获取 SToken(get_token_by_game_token) - 服务器没有正确返回")
+            logger.debug(f"网络请求返回: {res.text}")
+            return BaseApiStatus(incorrect_return=True), None
+        else:
+            logger.exception("通过 GameToken 获取 SToken(get_token_by_game_token) - 请求失败")
+            return BaseApiStatus(network_error=True), None
+
+
+async def get_cookie_token_by_game_token(
+        bbs_uid: str,
+        game_token: str,
+        retry: bool = True
+) -> Tuple[BaseApiStatus, Optional[BBSCookies]]:
+    """
+    通过 GameToken 获取 CookieToken
+
+    :param bbs_uid: 米游社账号 UID
+    :param game_token: 有效的 GameToken
+    :param retry: 是否允许重试
+    """
+    try:
+        async for attempt in get_async_retry(retry):
+            with attempt:
+                content = {
+                    "account_id": int(bbs_uid),
+                    "game_token": game_token
+                }
+                async with httpx.AsyncClient() as client:
+                    res = await client.post(
+                        URL_GET_COOKIE_TOKEN_BY_GAME_TOKEN,
+                        headers={"x-rpc-app_id": "bll8iq97cem8"},
+                        json=content,
+                        timeout=plugin_config.preference.timeout
+                    )
+                api_result = ApiResultHandler(res.json())
+                if api_result.retcode == 0:
+                    cookie_token = api_result.data["token"]["token"]
+                    return BaseApiStatus(success=True), BBSCookies(cookie_token=cookie_token)
+                else:
+                    logger.debug(f"网络请求返回: {res.text}")
+                    return BaseApiStatus(), None
+    except tenacity.RetryError as e:
+        if is_incorrect_return(e):
+            logger.exception("通过 GameToken 获取 CookieToken(get_cookie_token_by_game_token) - 服务器没有正确返回")
+            logger.debug(f"网络请求返回: {res.text}")
+            return BaseApiStatus(incorrect_return=True), None
+        else:
+            logger.exception("通过 GameToken 获取 CookieToken(get_cookie_token_by_game_token) - 请求失败")
+            return BaseApiStatus(network_error=True), None

src/nonebot_plugin_mystool/api/myb_missions_api.py

@@ -7,7 +7,7 @@
 from ..api.common import ApiResultHandler, is_incorrect_return, create_verification, \
     verify_verification
 from ..model import BaseApiStatus, MissionStatus, MissionData, \
-    MissionState, UserAccount, plugin_config, plugin_env
+    MissionState, UserAccount, plugin_config, plugin_env, UserData
 from ..utils import logger, generate_ds, \
     get_async_retry, get_validate
 
@@ -109,7 +109,7 @@ def __init__(self, account: UserAccount) -> None:
         self.headers = HEADERS_BASE.copy()
         self.headers["x-rpc-device_id"] = account.device_id_android
 
-    async def sign(self, retry: bool = True) -> Tuple[MissionStatus, Optional[int]]:
+    async def sign(self, user: UserData, retry: bool = True) -> Tuple[MissionStatus, Optional[int]]:
         """
         签到
 
@@ -146,10 +146,10 @@ async def sign(self, retry: bool = True) -> Tuple[MissionStatus, Optional[int]]:
                         logger.error(
                             f"米游币任务 - 讨论区签到: 用户 {self.account.display_name} 需要完成人机验证")
                         logger.debug(f"网络请求返回: {res.text}")
-                        if plugin_config.preference.geetest_url:
+                        if plugin_config.preference.geetest_url or user.geetest_url:
                             create_status, mmt_data = await create_verification(self.account)
                             if create_status:
-                                if geetest_result := await get_validate(mmt_data.gt, mmt_data.challenge):
+                                if geetest_result := await get_validate(user, mmt_data.gt, mmt_data.challenge):
                                     if await verify_verification(mmt_data, geetest_result, self.account):
                                         logger.success(
                                             f"米游币任务 - 讨论区签到: 用户 {self.account.display_name} 人机验证通过")