feat: 添加扫码登录，cookies复写，修正错误提示，添加成长值显示 (#246)

* feat: 添加扫码登录，cookies复写，修正错误提示 * On branch qr_login * chore: 遵守代码规范 * chore: not invalid-name * chore: 添加成长值显示 * imp: 更新版本号 --------- Co-authored-by: github-actions[bot] <github-actions[bot]@users.noreply.github.com> Co-authored-by: 0-8-4 <[email protected]>
0-8-4 · Dec 18, 2023 · 3991a86 · 3991a86
1 parent 5044bcc
commit 3991a86
Show file tree

Hide file tree

Showing 10 changed files with 277 additions and 53 deletions.
diff --git a/.gitignore b/.gitignore
@@ -142,5 +142,4 @@ dmypy.json
 logs
 data
 .pdm-python
-test.py
-test2.py
+test*.py
diff --git a/miuitask.py b/miuitask.py
@@ -1,13 +1,15 @@
 '''
 Date: 2023-11-13 20:29:19
 LastEditors: Night-stars-1 [email protected]
-LastEditTime: 2023-12-03 01:53:48
+LastEditTime: 2023-12-18 19:21:36
 '''
 # new Env("MIUI-Auto-Task") # pylint: disable=missing-module-docstring
 # cron 30 8 * * * miuitask.py
 
 import asyncio
 
+from tenacity import RetryError, Retrying, stop_after_attempt
+
 from utils.api.login import Login
 from utils.api.sign import BaseSign, CheckIn
 from utils.config import ConfigManager
@@ -23,24 +25,33 @@ async def main():
     """启动签到"""
     print_info()
     for account in _conf.accounts:
-        login_obj = Login(account)
-        if cookies := await login_obj.login():
-            sign_obj = BaseSign(cookies)
-            daily_tasks = await sign_obj.check_daily_tasks()
-            sign_task_obj = sign_obj.AVAILABLE_SIGNS  # 签到任务对象合集
-            for task in daily_tasks:
-                if not task.showType:
-                    log.info(f"开始执行{task.name}任务")
-                    if task_obj := sign_task_obj.get(task.name):  # 签到任务对象
-                        if getattr(account, task_obj.__name__):
+        try:
+            for attempt in Retrying(stop=stop_after_attempt(2)):
+                with attempt:
+                    login_obj = Login(account)
+                    if cookies := await login_obj.login():
+                        sign_obj = BaseSign(cookies, account.user_agent)
+                        daily_tasks = await sign_obj.check_daily_tasks()
+                        sign_task_obj = sign_obj.AVAILABLE_SIGNS  # 签到任务对象合集
+                        for task in daily_tasks:
+                            log.info(f"开始执行{task.name}任务")
+                            if task.showType:
+                                log.info(f"{task.name}任务已完成")
+                                continue
+                            if not (task_obj := sign_task_obj.get(task.name)): # 签到任务对象
+                                log.error(f"未找到{task.name}任务")
+                                continue
+                            if not getattr(account, task_obj.__name__):
+                                log.info(f"任务{task.name}被禁用")
+                                continue
                             token = await get_token(cookies["cUserId"]) if task_obj == CheckIn else None
-                            await task_obj(cookies, token).sign()
-                        else:
-                            log.info(f"任务{task.name}被禁用")
-                    else:
-                        log.error(f"未找到{task.name}任务")
-                else:
-                    log.info(f"{task.name}任务已完成")
+                            status, reason = await task_obj(cookies, token).sign()
+                            if not status and reason == "cookie":
+                                raise ValueError("Cookie失效")
+                        user_info = await sign_obj.user_info()
+                        log.info(f"{user_info.title} 成长值: {user_info.point}")
+        except RetryError:
+            ...
     notify_me(InterceptHandler.message)
 
 

diff --git a/pdm.lock b/pdm.lock
diff --git a/pyproject.toml b/pyproject.toml
@@ -13,6 +13,7 @@ dependencies = [
     "tenacity>=8.2.3",
     "tzdata>=2023.3",
     "onepush>=1.3.0",
+    "qrcode>=7.4.2",
 ]
 requires-python = ">=3.11"
 license = {text = "MIT"}

diff --git a/utils/api/login.py b/utils/api/login.py
@@ -3,8 +3,9 @@
 LastEditors: Night-stars-1 [email protected]
 LastEditTime: 2023-11-13 12:32:26
 """
+import time
 from os import getenv
-from typing import Dict, Union
+from typing import Dict, Optional, Tuple, Union
 
 import orjson
 
@@ -13,6 +14,8 @@
 from ..logger import log
 from ..request import get, post
 from .sign import BaseSign
+from ..utils import generate_qrcode
+
 
 class Login:
     """登录类"""
@@ -24,7 +27,7 @@ def __init__(self, account: Account) -> None:
         self.password = account.password
         self.cookies = account.cookies
 
-    async def login(self) -> Union[Dict[str, str], bool]:
+    async def login(self) -> Union[Dict[str, str], bool]: # pylint: disable=too-many-return-statements
         """登录小米账号"""
         headers = {
             'Content-Type': 'application/x-www-form-urlencoded; charset=UTF-8',
@@ -68,9 +71,17 @@ async def login(self) -> Union[Dict[str, str], bool]:
             repo_owner = getenv('GITHUB_REPOSITORY_OWNER')
             if repo_owner not in [None, "0-8-4"]:
                 return False
-            if self.cookies != {} and await BaseSign(self.cookies).check_daily_tasks(nolog=True) != []:
+            if self.cookies != {} and await BaseSign(self.cookies, self.user_agent).check_daily_tasks(nolog=True) != []:
                 log.info("Cookie有效，跳过登录")
                 return self.cookies
+            elif self.cookies.get("passToken") and \
+                (cookies := await self.get_cookies_by_passtk(user_id=self.uid,
+                                                             pass_token=self.cookies["passToken"])):
+                log.info("Cookie无效，重新复写")
+                self.cookies.update(cookies)
+                self.account.cookies = self.cookies
+                write_plugin_data()
+                return cookies
             response = await post('https://account.xiaomi.com/pass/serviceLoginAuth2', headers=headers, data=data)
             log.debug(response.text)
             result = response.text.lstrip('&').lstrip('START').lstrip('&')
@@ -83,14 +94,21 @@ async def login(self) -> Union[Dict[str, str], bool]:
                 self.account.cookies = cookies
                 write_plugin_data()
                 return cookies
-            elif not api_data.pwd_wrong:
-                log.error(f'小米账号登录失败：{api_data.message}')
+            elif api_data.pwd_wrong:
+                log.error('小米账号登录失败：用户名或密码不正确')
+                check_url = await self.qr_login()
+                userid, cookies = await self.check_login(check_url)
+                self.cookies.update(cookies)
+                self.account.cookies = self.cookies
+                self.account.uid = userid
+                write_plugin_data()
+                return cookies
             elif api_data.need_captcha:
                 log.error('当前账号需要短信验证码, 请尝试修改UA或设备ID')
             else:
-                log.error('小米账号登录失败：用户名或密码不正确')
+                log.error(f'小米账号登录失败：{api_data.message}')
             return False
-        except Exception: # pylint: disable=broad-exception-caught
+        except Exception:  # pylint: disable=broad-exception-caught
             log.exception("登录小米账号出错")
             return False
 
@@ -100,7 +118,107 @@ async def get_cookie(self, url: str) -> Union[Dict[str, str], bool]:
             response = await get(url, follow_redirects=False)
             log.debug(response.text)
             return dict(response.cookies)
-        except Exception: # pylint: disable=broad-exception-caught
+        except Exception:  # pylint: disable=broad-exception-caught
             log.exception("社区获取 Cookie 失败")
             return False
-
+
+    async def get_cookies_by_passtk(self, user_id: str, pass_token: str) -> Union[Dict[str, str], bool]:
+        """使用passToken获取签到cookies"""
+        try:
+            headers = {
+                'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7',
+                'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
+                'Cache-Control': 'no-cache',
+                'Connection': 'keep-alive',
+                'Pragma': 'no-cache',
+                'Referer': 'https://web.vip.miui.com/',
+                'Sec-Fetch-Dest': 'document',
+                'Sec-Fetch-Mode': 'navigate',
+                'Sec-Fetch-Site': 'same-site',
+                'Upgrade-Insecure-Requests': '1',
+                'User-Agent': self.user_agent,
+                'sec-ch-ua': '"Not_A Brand";v="8", "Chromium";v="120", "Microsoft Edge";v="120"',
+                'sec-ch-ua-mobile': '?0',
+                'sec-ch-ua-platform': '"Windows"',
+            }
+
+            params = {
+                'destUrl': 'https://web.vip.miui.com/page/info/mio/mio/checkIn?app_version=dev.230904',
+                'time': round(time.time() * 1000),
+            }
+            cookies = {
+                "userId": user_id,
+                "passToken": pass_token
+            }
+            response = await get('https://api.vip.miui.com/page/login', params=params, headers=headers)
+            url = response.headers.get("location")
+
+            response = await get(url, cookies=cookies, headers=headers)
+            url = response.headers.get("location")
+            response = await get(url, cookies=cookies, headers=headers)
+            return dict(response.cookies)
+        except Exception:  # pylint: disable=broad-exception-caught
+            log.exception("从passToken获取 Cookie 失败")
+            return {}
+
+    async def qr_login(self) -> Tuple[str, bytes]:
+        """二维码登录"""
+        headers = {
+            'Accept': 'application/json, text/plain, */*',
+            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
+            'Cache-Control': 'no-cache',
+            'Connection': 'keep-alive',
+            'Pragma': 'no-cache',
+            'Referer': 'https://account.xiaomi.com/',
+            'Sec-Fetch-Dest': 'empty',
+            'Sec-Fetch-Mode': 'cors',
+            'Sec-Fetch-Site': 'same-origin',
+            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0',
+            'X-Requested-With': 'XMLHttpRequest',
+            'sec-ch-ua': '"Not_A Brand";v="8", "Chromium";v="120", "Microsoft Edge";v="120"',
+            'sec-ch-ua-mobile': '?0',
+            'sec-ch-ua-platform': '"Windows"',
+        }
+
+        response = await get(
+            'https://account.xiaomi.com/longPolling/loginUrl?_group=DEFAULT&_qrsize=240&qs=%253Fcallback%253Dhttps%25253A%25252F%25252Faccount.xiaomi.com%25252Fsts%25253Fsign%25253DZvAtJIzsDsFe60LdaPa76nNNP58%2525253D%252526followup%25253Dhttps%2525253A%2525252F%2525252Faccount.xiaomi.com%2525252Fpass%2525252Fauth%2525252Fsecurity%2525252Fhome%252526sid%25253Dpassport%2526sid%253Dpassport%2526_group%253DDEFAULT&bizDeviceType=&callback=https:%2F%2Faccount.xiaomi.com%2Fsts%3Fsign%3DZvAtJIzsDsFe60LdaPa76nNNP58%253D%26followup%3Dhttps%253A%252F%252Faccount.xiaomi.com%252Fpass%252Fauth%252Fsecurity%252Fhome%26sid%3Dpassport&theme=&sid=passport&needTheme=false&showActiveX=false&serviceParam=%7B%22checkSafePhone%22:false,%22checkSafeAddress%22:false,%22lsrp_score%22:0.0%7D&_locale=zh_CN&_sign=2%26V1_passport%26BUcblfwZ4tX84axhVUaw8t6yi2E%3D&_dc=1702105962382', # pylint: disable=line-too-long
+            headers=headers,
+        )
+        result = response.text.replace("&&&START&&&", "")
+        data = orjson.loads(result) # pylint: disable=no-member
+        login_url = data["loginUrl"]
+        check_url = data["lp"]
+        generate_qrcode(login_url)
+        return check_url
+
+    async def check_login(self, url: str) -> Tuple[Optional[int], Optional[dict]]:
+        """检查扫码登录状态"""
+        try:
+            headers = {
+                'Accept': 'application/json, text/plain, */*',
+                'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6',
+                'Cache-Control': 'no-cache',
+                'Connection': 'keep-alive',
+                'Pragma': 'no-cache',
+                'Referer': 'https://account.xiaomi.com/',
+                'Sec-Fetch-Dest': 'empty',
+                'Sec-Fetch-Mode': 'cors',
+                'Sec-Fetch-Site': 'same-origin',
+                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0',
+                'X-Requested-With': 'XMLHttpRequest',
+                'sec-ch-ua': '"Not_A Brand";v="8", "Chromium";v="120", "Microsoft Edge";v="120"',
+                'sec-ch-ua-mobile': '?0',
+                'sec-ch-ua-platform': '"Windows"',
+            }
+            response = await get(url, headers=headers)
+            result = response.text.replace("&&&START&&&", "")
+            data = orjson.loads(result) # pylint: disable=no-member
+            pass_token = data["passToken"]
+            user_id = str(data["userId"])
+            cookies = await self.get_cookies_by_passtk(user_id=user_id, pass_token=pass_token)
+            cookies.update({
+                "passToken": pass_token
+            })
+            return user_id, cookies
+        except Exception: # pylint: disable=broad-exception-caught
+            return None, None