feat: 添加小程序签到，移除pydantic和异步 (#306)

* feat: 添加视频浏览任务

* fix: 修复无法找到视频任务配置的问题

* feat: 添加未设置`login_user_agent`提示

* fix: string statement has no effect

* feat: 添加微信小程序签到

* chore: 移除异步

* chore: 移除pydantic

* chore: 适配pylint规范

* skip: Use of weak MD5 hash for security.

.bandit

@@ -0,0 +1,8 @@
+"""
+Author: Night-stars-1 nujj1042633805@gmail.com
+Date: 2025-01-19 17:16:35
+LastEditors: Night-stars-1 nujj1042633805@gmail.com
+LastEditTime: 2025-01-19 17:16:43
+"""
+
+skips: ["B324"]

.pylintrc

@@ -2,4 +2,4 @@
 max-line-length=200
 
 [MESSAGES CONTROL]
-disable=no-else-return, too-few-public-methods
+disable=no-else-return, too-few-public-methods, invalid-name

miuitask.py

@@ -1,10 +1,9 @@
 """
 Author: Night-stars-1 nujj1042633805@gmail.com
 Date: 2024-02-21 22:50:45
-LastEditTime: 2024-08-20 22:36:18
+LastEditTime: 2025-01-19 17:09:37
 LastEditors: Night-stars-1 nujj1042633805@gmail.com
 """
-import asyncio
 
 from tenacity import Retrying, stop_after_attempt
 
@@ -19,18 +18,18 @@
 _conf = ConfigManager.data_obj
 
 
-async def main():
+def main():
     """启动签到"""
     print_info()
     for account in _conf.accounts:
         try:
             for attempt in Retrying(stop=stop_after_attempt(2)):
                 with attempt:
                     login_obj = Login(account)
-                    if cookies := await login_obj.login():
-                        await login_obj.checkin_info()
+                    if cookies := login_obj.login():
+                        login_obj.checkin_info()
                         sign_obj = BaseSign(account)
-                        daily_tasks = await sign_obj.check_daily_tasks()
+                        daily_tasks = sign_obj.check_daily_tasks()
                         sign_task_obj = sign_obj.AVAILABLE_SIGNS  # 签到任务对象合集
                         for task in daily_tasks:
                             log.info(f"开始执行{task.name}任务")
@@ -46,19 +45,19 @@ async def main():
                                 log.info(f"任务{task.name}被禁用")
                                 continue
                             token = (
-                                await get_token(cookies["cUserId"])
+                                get_token(cookies["cUserId"])
                                 if task_obj == CheckIn
                                 else None
                             )
-                            status, reason = await task_obj(account, token).sign()
+                            status, reason = task_obj(account, token).sign()
                             if not status and reason == "cookie":
                                 raise ValueError("Cookie失效")
-                        user_info = await sign_obj.user_info()
+                        user_info = sign_obj.user_info()
                         log.info(f"{user_info.title} 成长值: {user_info.point}")
-        except ValueError:
-            ...
+        except ValueError as e:
+            log.error(e)
     notify_me(InterceptHandler.message)
 
 
 if __name__ == "__main__":
-    asyncio.run(main())
+    main()

requirements.txt

@@ -2,8 +2,6 @@ cryptography==42.0.4
 httpx==0.25.2
 loguru==0.7.2
 onepush==1.3.0
-orjson==3.9.15
-pydantic==2.5.2
 PyYAML==6.0.1
 tenacity==8.2.3
 qrcode>=7.4.2

utils/api/login.py

@@ -1,7 +1,7 @@
 """
 Date: 2023-11-12 14:05:06
 LastEditors: Night-stars-1 nujj1042633805@gmail.com
-LastEditTime: 2024-08-20 22:40:54
+LastEditTime: 2025-01-19 16:35:56
 """
 
 import time
@@ -10,7 +10,7 @@
 
 import orjson
 
-from ..config import Account, write_plugin_data
+from ..config import Account, ConfigManager
 from ..data_model import LoginResultHandler
 from ..logger import log
 from ..request import get, post
@@ -29,7 +29,7 @@ def __init__(self, account: Account) -> None:
         self.cookies = account.cookies
 
     # pylint: disable=too-many-return-statements
-    async def login(
+    def login(
         self,
     ) -> Union[Dict[str, str], bool]:
         """登录小米账号"""
@@ -58,21 +58,21 @@ async def login(
                 return False
             if (
                 self.cookies != {}
-                and await BaseSign(self.account).check_daily_tasks(nolog=True) != []
+                and BaseSign(self.account).check_daily_tasks(nolog=True) != []
             ):
                 log.info("Cookie有效，跳过登录")
                 return self.cookies
             elif self.cookies.get("passToken") and (
-                cookies := await self.get_cookies_by_passtk(
+                cookies := 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()
+                ConfigManager.write_plugin_data()
                 return cookies
-            response = await post(
+            response = post(
                 "https://account.xiaomi.com/pass/serviceLoginAuth2",
                 headers=headers,
                 data=data,
@@ -86,22 +86,22 @@ async def login(
                 log.success("小米账号登录成功")
                 self.account.cookies["passToken"] = api_data.pass_token
                 self.account.uid = api_data.user_id
-                if cookies := await self.get_cookies_by_passtk(
+                if cookies := self.get_cookies_by_passtk(
                     api_data.user_id, api_data.pass_token
                 ):
                     self.account.cookies.update(cookies)
-                    write_plugin_data()
+                    ConfigManager.write_plugin_data()
                     return cookies
                 log.error("获取Cookie失败，可能是 login_user_agent 异常")
                 return False
             elif api_data.pwd_wrong:
                 log.error("小米账号登录失败：用户名或密码不正确, 请扫码登录")
-                check_url = await self.qr_login()
-                userid, cookies = await self.check_login(check_url)
+                check_url = self.qr_login()
+                userid, cookies = self.check_login(check_url)
                 self.cookies.update(cookies)
                 self.account.cookies = self.cookies
                 self.account.uid = userid
-                write_plugin_data()
+                ConfigManager.write_plugin_data()
                 return cookies
             elif api_data.need_captcha:
                 log.error("当前账号需要短信验证码, 请尝试修改UA或设备ID")
@@ -112,17 +112,17 @@ async def login(
             log.exception("登录小米账号出错")
             return False
 
-    async def get_cookies(self, url: str) -> Union[Dict[str, str], bool]:
+    def get_cookies(self, url: str) -> Union[Dict[str, str], bool]:
         """获取社区 Cookie"""
         try:
-            response = await get(url, follow_redirects=False)
+            response = get(url, follow_redirects=False)
             log.debug(response.text)
             return dict(response.cookies)
         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):
+    def get_cookies_by_passtk(self, user_id: str, pass_token: str):
         """使用passToken获取签到cookies"""
         try:
             headers = {
@@ -147,23 +147,23 @@ async def get_cookies_by_passtk(self, user_id: str, pass_token: str):
                 "time": round(time.time() * 1000),
             }
             cookies = {"userId": user_id, "passToken": pass_token}
-            response = await get(
+            response = get(
                 "https://api-alpha.vip.miui.com/page/login",
                 params=params,
                 headers=headers,
             )
             url = response.headers.get("location")
 
-            response = await get(url, cookies=cookies, headers=headers)
+            response = get(url, cookies=cookies, headers=headers)
             url = response.headers.get("location")
 
-            response = await get(url, cookies=cookies, headers=headers)
+            response = 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]:
+    def qr_login(self) -> Tuple[str, bytes]:
         """二维码登录"""
         headers = {
             "Accept": "application/json, text/plain, */*",
@@ -182,7 +182,7 @@ async def qr_login(self) -> Tuple[str, bytes]:
             "sec-ch-ua-platform": '"Windows"',
         }
 
-        response = await get(
+        response = 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,
         )
@@ -194,7 +194,7 @@ async def qr_login(self) -> Tuple[str, bytes]:
         generate_qrcode(login_url)
         return check_url
 
-    async def check_login(self, url: str) -> Tuple[Optional[int], Optional[dict]]:
+    def check_login(self, url: str) -> Tuple[Optional[int], Optional[dict]]:
         """检查扫码登录状态"""
         try:
             headers = {
@@ -213,21 +213,19 @@ async def check_login(self, url: str) -> Tuple[Optional[int], Optional[dict]]:
                 "sec-ch-ua-mobile": "?0",
                 "sec-ch-ua-platform": '"Windows"',
             }
-            response = await get(url, headers=headers)
+            response = 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 = 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
 
-    async def checkin_info(self) -> Union[Dict[str, str], bool]:
-        """获取签到+1概率"""
+    def checkin_info(self) -> Union[Dict[str, str], bool]:
+        """获取公告消息"""
         headers = {
             "Accept": "*/*",
             "Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6",
@@ -252,14 +250,14 @@ async def checkin_info(self) -> Union[Dict[str, str], bool]:
                 "miui_vip_a_ph": self.cookies["miui_vip_a_ph"],
             }
 
-            response = await get(
+            response = get(
                 "https://api-alpha.vip.miui.com/mtop/planet/vip/user/getUserCheckinInfoV2",
                 params=params,
                 cookies=self.cookies,
                 headers=headers,
             )
             log.debug(response.text)
             data: dict = response.json()  # pylint: disable=no-member
-            log.info(data.get("entity", {}).get("checkinInfoList", ["", "异常"])[1])
+            log.info(",".join(data.get("entity", {}).get("checkinInfoList", ["异常"])))
         except Exception:  # pylint: disable=broad-exception-caught
             log.exception("获取用户信息失败")