diff --git a/.bandit b/.bandit new file mode 100644 index 00000000..2e084b3b --- /dev/null +++ b/.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"] diff --git a/.pylintrc b/.pylintrc index 6b51c0c4..82feaf6e 100644 --- a/.pylintrc +++ b/.pylintrc @@ -2,4 +2,4 @@ max-line-length=200 [MESSAGES CONTROL] -disable=no-else-return, too-few-public-methods \ No newline at end of file +disable=no-else-return, too-few-public-methods, invalid-name \ No newline at end of file diff --git a/miuitask.py b/miuitask.py index 9326081c..140f8bd8 100644 --- a/miuitask.py +++ b/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,7 +18,7 @@ _conf = ConfigManager.data_obj -async def main(): +def main(): """启动签到""" print_info() for account in _conf.accounts: @@ -27,10 +26,10 @@ async def main(): 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() diff --git a/requirements.txt b/requirements.txt index e2a9b28c..689edfff 100644 --- a/requirements.txt +++ b/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 diff --git a/utils/api/login.py b/utils/api/login.py index 36ef0c83..33dcd252 100644 --- a/utils/api/login.py +++ b/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,7 +250,7 @@ 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, @@ -260,6 +258,6 @@ async def checkin_info(self) -> Union[Dict[str, str], bool]: ) 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("获取用户信息失败") diff --git a/utils/api/sign.py b/utils/api/sign.py index 4e2f4ef0..48b1cda7 100644 --- a/utils/api/sign.py +++ b/utils/api/sign.py @@ -1,17 +1,22 @@ -''' +""" Date: 2023-12-03 02:07:29 LastEditors: Night-stars-1 nujj1042633805@gmail.com -LastEditTime: 2023-12-31 01:28:23 -''' +LastEditTime: 2025-01-19 17:03:47 +""" + import time -from typing import Any, Dict, List, Optional, Tuple, Type, Union +from typing import Any, Dict, List, Optional, Tuple, Type from requests_toolbelt import MultipartEncoder from tenacity import RetryError, Retrying, stop_after_attempt from ..config import Account -from ..data_model import (ApiResultHandler, DailyTasksResult, - SignResultHandler, UserInfoResult) +from ..data_model import ( + ApiResultHandler, + DailyTasksResult, + SignResultHandler, + UserInfoResult, +) from ..logger import log from ..request import get, post from ..utils import get_random_chars_as_string, is_incorrect_return @@ -21,6 +26,7 @@ class BaseSign: """ 签到基类 """ + NAME = "" """任务名字""" @@ -41,72 +47,94 @@ def __init__(self, account: Account, token: Optional[str] = None): self.token = token self.user_agent = account.user_agent self.headers = { - 'Host': 'api-alpha.vip.miui.com', - 'Connection': 'keep-alive', - 'Pragma': 'no-cache', - 'Cache-Control': 'no-cache', - 'sec-ch-ua': 'Not_A', - 'Accept': 'application/json', - 'sec-ch-ua-mobile': '?1', - 'User-Agent': self.user_agent, - 'sec-ch-ua-platform': 'Android', - 'Origin': 'https://web-alpha.vip.miui.com', - 'X-Requested-With': 'com.xiaomi.vipaccount', - 'Sec-Fetch-Site': 'same-site', - 'Sec-Fetch-Mode': 'cors', - 'Sec-Fetch-Dest': 'empty', - 'Referer': 'https://web-alpha.vip.miui.com/', - 'Accept-Language': 'zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7', + "Host": "api-alpha.vip.miui.com", + "Connection": "keep-alive", + "Pragma": "no-cache", + "Cache-Control": "no-cache", + "sec-ch-ua": "Not_A", + "Accept": "application/json", + "sec-ch-ua-mobile": "?1", + "User-Agent": self.user_agent, + "sec-ch-ua-platform": "Android", + "Origin": "https://web-alpha.vip.miui.com", + "X-Requested-With": "com.xiaomi.vipaccount", + "Sec-Fetch-Site": "same-site", + "Sec-Fetch-Mode": "cors", + "Sec-Fetch-Dest": "empty", + "Referer": "https://web-alpha.vip.miui.com/", + "Accept-Language": "zh-CN,zh;q=0.9,en-US;q=0.8,en;q=0.7", } self.params = { - 'ref': 'vipAccountShortcut', - 'pathname': '/mio/detail', - 'version': 'dev.20231205', - 'miui_version': 'V816.0.23.12.11.DEV', - 'android_version': '14', - 'oaid': {get_random_chars_as_string(16, "0123456789abcdefghijklmnopqrstuvwxyz")}, - 'device': account.device, - 'restrict_imei': '', - 'miui_big_version': 'V816', - 'model': account.device_model, - 'androidVersion': '14', - 'miuiBigVersion': 'V816', - 'miui_vip_a_ph': None, + "ref": "vipAccountShortcut", + "pathname": "/mio/detail", + "version": "dev.20231205", + "miui_version": "V816.0.23.12.11.DEV", + "android_version": "14", + "oaid": { + get_random_chars_as_string(16, "0123456789abcdefghijklmnopqrstuvwxyz") + }, + "device": account.device, + "restrict_imei": "", + "miui_big_version": "V816", + "model": account.device_model, + "androidVersion": "14", + "miuiBigVersion": "V816", + "miui_vip_a_ph": None, } - async def check_daily_tasks(self, nolog: bool = False) -> Union[List[DailyTasksResult], List[None]]: + def check_daily_tasks(self, nolog: bool = False): """获取每日任务状态""" try: + task_status: List[DailyTasksResult] = [] for attempt in Retrying(stop=stop_after_attempt(3)): with attempt: - response = await get('https://api-alpha.vip.miui.com/mtop/planet/vip/member/getCheckinPageCakeList', - cookies=self.cookies) + response = get( + "https://api-alpha.vip.miui.com/mtop/planet/vip/member/getCheckinPageCakeList", + cookies=self.cookies, + ) log.debug(response.text) result = response.json() api_data = ApiResultHandler(result) if api_data.success: - task_status = [] - tasks: List[Dict[str, List[Dict[str, Any]]]] = list(filter( - lambda x: x['head']['title'] in ["每日任务", "其他任务"], api_data.data)) + tasks: List[Dict[str, List[Dict[str, Any]]]] = list( + filter( + lambda x: x["head"]["title"] + in ["每日任务", "其他任务"], + api_data.data, + ) + ) for task in tasks: - for daily_task in task['data']: - task_name = daily_task['title'] - task_desc = daily_task.get('desc', '') - show_type = True if daily_task['showType'] == 0 else False # pylint: disable=simplifiable-if-expression - task_status.append(DailyTasksResult(name=task_name, showType=show_type, desc=task_desc)) + for daily_task in task["data"]: + task_name = daily_task["title"] + task_desc = daily_task.get("desc", "") + show_type = daily_task["showType"] == 0 + task_status.append( + DailyTasksResult( + name=task_name, + showType=show_type, + desc=task_desc, + ) + ) + task_status.append( + DailyTasksResult( + name=WxSign.NAME, + showType=False, + desc=WxSign.NAME, + ) + ) return task_status else: if not nolog: log.error(f"获取每日任务状态失败：{api_data.message}") - return [] + return task_status except RetryError as error: if is_incorrect_return(error): log.exception(f"每日任务 - 服务器没有正确返回 {response.text}") else: log.exception("获取每日任务异常") - return [] + return task_status - async def sign(self) -> Tuple[bool, str]: # pylint: disable=too-many-branches + def sign(self) -> Tuple[bool, str]: # pylint: disable=too-many-branches """ 每日任务处理器 """ @@ -114,28 +142,32 @@ async def sign(self) -> Tuple[bool, str]: # pylint: disable=too-many-branches for attempt in Retrying(stop=stop_after_attempt(3)): with attempt: params = self.PARAMS.copy() - if 'miui_vip_a_ph' in self.cookies: - params['miui_vip_a_ph'] = self.cookies['miui_vip_a_ph'] - if 'token' in params: - params['token'] = self.token + if "miui_vip_a_ph" in self.cookies: + params["miui_vip_a_ph"] = self.cookies["miui_vip_a_ph"] + if "token" in params: + params["token"] = self.token self.params.update(params) self.params["version"] = self.user_agent.split("/")[-1] data = self.DATA.copy() - if 'miui_vip_a_ph' in self.cookies: - data['miui_vip_a_ph'] = self.cookies['miui_vip_a_ph'] - if 'token' in data: + if "miui_vip_a_ph" in self.cookies: + data["miui_vip_a_ph"] = self.cookies["miui_vip_a_ph"] + if "token" in data: if self.token: - data['token'] = self.token + data["token"] = self.token else: log.info(f"未获取到token, 跳过{self.NAME}") return False, "None" - boundary=f'----WebKitFormBoundaryZ{get_random_chars_as_string(16, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")}' + boundary = f'----WebKitFormBoundaryZ{get_random_chars_as_string(16, "0123456789abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ")}' data = MultipartEncoder(fields=data, boundary=boundary) - self.headers['Content-Type'] = data.content_type - response = await post(self.URL_SIGN, - params=self.params, data=data.to_string(), - cookies=self.cookies, headers=self.headers) + self.headers["Content-Type"] = data.content_type + response = post( + self.URL_SIGN, + params=self.params, + data=data.to_string(), + cookies=self.cookies, + headers=self.headers, + ) log.debug(response.text) result = response.json() api_data = SignResultHandler(result) @@ -158,21 +190,21 @@ async def sign(self) -> Tuple[bool, str]: # pylint: disable=too-many-branches log.exception("{self.NAME}出错") return False, "None" - async def user_info(self) -> UserInfoResult: + def user_info(self) -> UserInfoResult: """获取用户信息""" try: for attempt in Retrying(stop=stop_after_attempt(3)): with attempt: headers = { - 'Content-Type': 'application/x-www-form-urlencoded', - 'User-Agent': self.user_agent, - 'Request-Container-Mark': 'android', - 'Host': 'api-alpha.vip.miui.com', - 'Connection': 'Keep-Alive', + "Content-Type": "application/x-www-form-urlencoded", + "User-Agent": self.user_agent, + "Request-Container-Mark": "android", + "Host": "api-alpha.vip.miui.com", + "Connection": "Keep-Alive", } - response = await get( - 'https://api-alpha.vip.miui.com/mtop/planet/vip/homepage/mineInfo', + response = get( + "https://api-alpha.vip.miui.com/mtop/planet/vip/homepage/mineInfo", cookies=self.cookies, headers=headers, ) @@ -180,7 +212,7 @@ async def user_info(self) -> UserInfoResult: result = response.json() api_data = ApiResultHandler(result) if api_data.success: - return UserInfoResult.model_validate(api_data.data) + return UserInfoResult(**api_data.data) else: log.error(f"获取用户信息失败：{api_data.message}") return UserInfoResult() @@ -190,162 +222,174 @@ async def user_info(self) -> UserInfoResult: else: log.exception("获取用户信息异常") return UserInfoResult() -#pylint: disable=trailing-whitespace + + +# pylint: disable=trailing-whitespace class CheckIn(BaseSign): """ 每日签到 """ + NAME = "每日签到" PARAMS = { - 'ref': 'vipAccountShortcut', - 'pathname': '/mio/checkIn', - 'version': 'dev.231026', - 'miui_vip_a_ph': "{miui_vip_a_ph}" + "ref": "vipAccountShortcut", + "pathname": "/mio/checkIn", + "version": "dev.231026", + "miui_vip_a_ph": "{miui_vip_a_ph}", } - DATA = { - 'miui_vip_a_ph': "{miui_vip_a_ph}", - 'token': "{token}" - } - URL_SIGN = 'https://api-alpha.vip.miui.com/mtop/planet/vip/user/checkinV2' + DATA = {"miui_vip_a_ph": "{miui_vip_a_ph}", "token": "{token}"} + URL_SIGN = "https://api-alpha.vip.miui.com/mtop/planet/vip/user/checkinV2" class BrowsePost(BaseSign): """ 浏览帖子超过10秒 """ + NAME = "浏览帖子超过10秒" PARAMS = { - 'ref': 'vipAccountShortcut', - 'pathname': '/mio/detail', - 'version': 'dev.231026', - 'miui_vip_a_ph': "{miui_vip_a_ph}" + "ref": "vipAccountShortcut", + "pathname": "/mio/detail", + "version": "dev.231026", + "miui_vip_a_ph": "{miui_vip_a_ph}", } - DATA = { - 'action': 'BROWSE_POST_10S', - 'miui_vip_a_ph': "{miui_vip_a_ph}" - } - URL_SIGN = 'https://api-alpha.vip.miui.com/mtop/planet/vip/member/addCommunityGrowUpPointByActionV2' + DATA = {"action": "BROWSE_POST_10S", "miui_vip_a_ph": "{miui_vip_a_ph}"} + URL_SIGN = "https://api-alpha.vip.miui.com/mtop/planet/vip/member/addCommunityGrowUpPointByActionV2" class BrowseUserPage(BaseSign): """ 浏览个人主页10s """ + NAME = "浏览个人主页10s" PARAMS = { - 'ref': 'vipAccountShortcut', - 'pathname': '/mio/detail', - 'version': 'dev.231026', - 'miui_vip_a_ph': "{miui_vip_a_ph}" + "ref": "vipAccountShortcut", + "pathname": "/mio/detail", + "version": "dev.231026", + "miui_vip_a_ph": "{miui_vip_a_ph}", } DATA = { - 'action': 'BROWSE_SPECIAL_PAGES_USER_HOME', - 'miui_vip_a_ph': "{miui_vip_a_ph}" + "action": "BROWSE_SPECIAL_PAGES_USER_HOME", + "miui_vip_a_ph": "{miui_vip_a_ph}", } - URL_SIGN = 'https://api-alpha.vip.miui.com/mtop/planet/vip/member/addCommunityGrowUpPointByActionV2' + URL_SIGN = "https://api-alpha.vip.miui.com/mtop/planet/vip/member/addCommunityGrowUpPointByActionV2" class BrowseSpecialPage(BaseSign): """ 浏览指定专题页 """ + NAME = "浏览指定专题页" PARAMS = { - 'ref': 'vipAccountShortcut', - 'pathname': '/mio/detail', - 'version': 'dev.231026', - 'miui_vip_a_ph': "{miui_vip_a_ph}" + "ref": "vipAccountShortcut", + "pathname": "/mio/detail", + "version": "dev.231026", + "miui_vip_a_ph": "{miui_vip_a_ph}", } DATA = { - 'action': 'BROWSE_SPECIAL_PAGES_SPECIAL_PAGE', - 'miui_vip_a_ph': "{miui_vip_a_ph}" + "action": "BROWSE_SPECIAL_PAGES_SPECIAL_PAGE", + "miui_vip_a_ph": "{miui_vip_a_ph}", } - URL_SIGN = 'https://api-alpha.vip.miui.com/mtop/planet/vip/member/addCommunityGrowUpPointByActionV2' + URL_SIGN = "https://api-alpha.vip.miui.com/mtop/planet/vip/member/addCommunityGrowUpPointByActionV2" class BrowseVideoPost(BaseSign): """ 浏览指定页面的多个视频超过5分钟（中途退出重新计算时间） """ + NAME = "浏览指定页面的多个视频超过5分钟（中途退出重新计算时间）" PARAMS = { - 'ref': 'vipAccountShortcut', - 'pathname': '/mio/detail', - 'version': 'dev.231026', - 'miui_vip_a_ph': "{miui_vip_a_ph}" - } - DATA = { - 'action': 'BROWSE_VIDEO_POST', - 'miui_vip_a_ph': "{miui_vip_a_ph}" + "ref": "vipAccountShortcut", + "pathname": "/mio/detail", + "version": "dev.231026", + "miui_vip_a_ph": "{miui_vip_a_ph}", } - URL_SIGN = 'https://api-alpha.vip.miui.com/mtop/planet/vip/member/addCommunityGrowUpPointByActionV2' + DATA = {"action": "BROWSE_VIDEO_POST", "miui_vip_a_ph": "{miui_vip_a_ph}"} + URL_SIGN = "https://api-alpha.vip.miui.com/mtop/planet/vip/member/addCommunityGrowUpPointByActionV2" class BoardFollow(BaseSign): """ 加入小米圈子 """ + NAME = "加入小米圈子" PARAMS = { - 'pathname': '/mio/allboard', - 'version': 'dev.20051', - 'boardId': '558495', - 'miui_vip_a_ph': "{miui_vip_a_ph}" + "pathname": "/mio/allboard", + "version": "dev.20051", + "boardId": "558495", + "miui_vip_a_ph": "{miui_vip_a_ph}", } - URL_SIGN = 'https://api-alpha.vip.miui.com/api/community/board/follow' + URL_SIGN = "https://api-alpha.vip.miui.com/api/community/board/follow" class BoardUnFollow(BaseSign): """ 退出小米圈子 """ + NAME = "退出小米圈子" PARAMS = { - 'pathname': '/mio/allboard', - 'version': 'dev.20051', - 'boardId': '558495', - 'miui_vip_a_ph': "{miui_vip_a_ph}" + "pathname": "/mio/allboard", + "version": "dev.20051", + "boardId": "558495", + "miui_vip_a_ph": "{miui_vip_a_ph}", } - URL_SIGN = 'https://api-alpha.vip.miui.com/api/community/board/unfollow' + URL_SIGN = "https://api-alpha.vip.miui.com/api/community/board/unfollow" class ThumbUp(BaseSign): """ 点赞他人帖子 """ + NAME = "点赞他人帖子" DATA = { - 'postId': '36625780', - 'sign': '36625780', - 'timestamp': int(round(time.time() * 1000)) + "postId": "36625780", + "sign": "36625780", + "timestamp": int(round(time.time() * 1000)), } - URL_SIGN = 'https://api-alpha.vip.miui.com/mtop/planet/vip/content/announceThumbUp' + URL_SIGN = "https://api-alpha.vip.miui.com/mtop/planet/vip/content/announceThumbUp" class CarrotPull(BaseSign): """ 参与拔萝卜获得奖励 """ + NAME = "参与拔萝卜获得奖励" - DATA = { - 'miui_vip_a_ph': "{miui_vip_a_ph}" - } - URL_SIGN = 'https://api-alpha.vip.miui.com/api/carrot/pull' + DATA = {"miui_vip_a_ph": "{miui_vip_a_ph}"} + URL_SIGN = "https://api-alpha.vip.miui.com/api/carrot/pull" + + +class WxSign(BaseSign): + """ + 微信签到 + """ + + NAME = "微信小程序签到" + PARAMS = {"miui_vip_a_ph": "{miui_vip_a_ph}"} + DATA = {"action": "WECHAT_CHECKIN_TASK"} + URL_SIGN = "https://api.vip.miui.com/mtop/planet/vip/member/addCommunityGrowUpPointByActionV2" # 注册签到任务 +BaseSign.AVAILABLE_SIGNS[WxSign.NAME] = WxSign BaseSign.AVAILABLE_SIGNS[CheckIn.NAME] = CheckIn BaseSign.AVAILABLE_SIGNS[BrowsePost.NAME] = BrowsePost BaseSign.AVAILABLE_SIGNS[BrowseVideoPost.NAME] = BrowseVideoPost diff --git a/utils/captcha.py b/utils/captcha.py index c85ffc45..4fe3f4f3 100644 --- a/utils/captcha.py +++ b/utils/captcha.py @@ -21,7 +21,7 @@ def find_key(data: dict, key: str): find_key(dvalue, key) return None -async def get_validate(gt: str, challenge: str) -> GeetestResult: # pylint: disable=invalid-name +def get_validate(gt: str, challenge: str) -> GeetestResult: # pylint: disable=invalid-name """获取人机验证结果""" try: validate = None @@ -30,7 +30,7 @@ async def get_validate(gt: str, challenge: str) -> GeetestResult: # pylint: dis params = json.loads(json.dumps(params).replace("{gt}", gt).replace("{challenge}", challenge)) data = _conf.preference.geetest_data.copy() data = json.loads(json.dumps(data).replace("{gt}", gt).replace("{challenge}", challenge)) - response = await post( + response = post( _conf.preference.geetest_url, params=params, json=data, diff --git a/utils/config.py b/utils/config.py index 2440d40a..baed42f4 100644 --- a/utils/config.py +++ b/utils/config.py @@ -1,14 +1,12 @@ """"配置文件""" + +import json import os import platform from hashlib import md5 -from json import JSONDecodeError from pathlib import Path -from typing import Dict, List, Optional, Union -import orjson -import yaml # pylint: disable=wrong-import-order -from pydantic import BaseModel, ValidationError, field_validator # pylint: disable=no-name-in-module +import yaml # pylint: disable=wrong-import-order from .logger import log @@ -20,7 +18,11 @@ CONFIG_TYPE = "json" if os.path.isfile(DATA_PATH / "config.json") else "yaml" """数据文件类型""" -CONFIG_PATH = DATA_PATH / f"config.{CONFIG_TYPE}" if os.getenv("MIUITASK_CONFIG_PATH") is None else Path(os.getenv("MIUITASK_CONFIG_PATH")) +CONFIG_PATH = ( + DATA_PATH / f"config.{CONFIG_TYPE}" + if os.getenv("MIUITASK_CONFIG_PATH") is None + else Path(os.getenv("MIUITASK_CONFIG_PATH")) +) """数据文件默认路径""" os.makedirs(DATA_PATH, exist_ok=True) @@ -28,7 +30,7 @@ def md5_crypto(passwd: str) -> str: """MD5加密""" - return md5(passwd.encode('utf8')).hexdigest().upper() + return md5(passwd.encode("utf8")).hexdigest().upper() def cookies_to_dict(cookies: str): @@ -36,166 +38,197 @@ def cookies_to_dict(cookies: str): cookies_dict = {} if not cookies or "=" not in cookies: return cookies_dict - for cookie in cookies.split(';'): - key, value = cookie.strip().split('=', 1) # 分割键和值 + for cookie in cookies.split(";"): + key, value = cookie.strip().split("=", 1) # 分割键和值 cookies_dict[key] = value return cookies_dict + def get_platform() -> str: """获取当前运行平台""" - if os.path.exists('/.dockerenv'): - if os.environ.get('QL_DIR') and os.environ.get('QL_BRANCH'): + if os.path.exists("/.dockerenv"): + if os.environ.get("QL_DIR") and os.environ.get("QL_BRANCH"): return "qinglong" else: return "docker" return platform.system().lower() -class Account(BaseModel): +# pylint: disable=too-many-instance-attributes +class Account: """账号处理器""" - uid: str = "100000" - """账户ID 非账户用户名或手机号""" - password: str = "" - """账户密码或其MD5哈希""" - cookies: Union[dict, str] = {} - """账户登录后的cookies""" - login_user_agent: str = "" - """登录账户时所用浏览器的 User-Agent""" - user_agent: str = 'Mozilla/5.0 (Linux; Android 13) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/116.0.0.0 Safari/537.36' - """登录社区时所用浏览器的 User-Agent""" - device: str = "" - """设备代号""" - device_model: str = "" - """设备名称""" - CheckIn: bool = False - """社区成长值签到，启用功能意味着你愿意自行承担相关风险""" - BrowseUserPage: bool = False - """社区浏览个人主页10秒，启用功能意味着你愿意自行承担相关风险""" - BrowsePost: bool = False - """社区浏览帖子10秒，启用功能意味着你愿意自行承担相关风险""" - BrowseVideoPost: bool = False - """社区浏览视频帖子5分钟，启用功能意味着你愿意自行承担相关风险""" - ThumbUp: bool = False - """点赞帖子，启用功能意味着你愿意自行承担相关风险""" - BrowseSpecialPage: bool = False - """社区在活动期间可能会出现限时的“浏览指定专题页”任务，启用功能意味着你愿意自行承担相关风险""" - BoardFollow: bool = False - """社区可能会出现限时的“加入圈子”任务，启用功能意味着你愿意自行承担相关风险""" - CarrotPull: bool = False - """社区拔萝卜，启用功能意味着你愿意自行承担相关风险""" - - @field_validator("password") - @classmethod - def _password(cls, value: Optional[str]): # pylint: disable=no-self-argument - if len(value) == 32: - return value - return md5_crypto(value) - @field_validator("cookies") - @classmethod - def _cookies(cls, value: Union[dict, str]): # pylint: disable=no-self-argument - if isinstance(value, str): - return cookies_to_dict(value) - return value + # pylint: disable=too-many-arguments, too-many-positional-arguments, too-many-locals + def __init__( + self, + uid="100000", + password="", + cookies=None, + login_user_agent="", + user_agent="Mozilla/5.0 (Linux; Android 13) AppleWebKit/537.36 (KHTML, like Gecko) Version/4.0 Chrome/116.0.0.0 Safari/537.36", + device="", + device_model="", + CheckIn=False, + BrowseUserPage=False, + BrowsePost=False, + BrowseVideoPost=False, + ThumbUp=False, + BrowseSpecialPage=False, + BoardFollow=False, + CarrotPull=False, + WxSign=False, + ): + self.uid = uid + """账户ID 非账户用户名或手机号""" + self.password = password + """账户密码或其MD5哈希""" + self.cookies = cookies or {} + """账户登录后的cookies""" + self.login_user_agent = login_user_agent + """登录账户时所用浏览器的 User-Agent""" + self.user_agent = user_agent + """登录社区时所用浏览器的 User-Agent""" + self.device = device + """设备代号""" + self.device_model = device_model + """设备名称""" + self.CheckIn = CheckIn + """社区成长值签到，启用功能意味着你愿意自行承担相关风险""" + self.BrowseUserPage = BrowseUserPage + """社区浏览个人主页10秒，启用功能意味着你愿意自行承担相关风险""" + self.BrowsePost = BrowsePost + """社区浏览帖子10秒，启用功能意味着你愿意自行承担相关风险""" + self.BrowseVideoPost = BrowseVideoPost + """社区浏览视频帖子5分钟，启用功能意味着你愿意自行承担相关风险""" + self.ThumbUp = ThumbUp + """点赞帖子，启用功能意味着你愿意自行承担相关风险""" + self.BrowseSpecialPage = BrowseSpecialPage + """社区在活动期间可能会出现限时的“浏览指定专题页”任务，启用功能意味着你愿意自行承担相关风险""" + self.BoardFollow = BoardFollow + """社区可能会出现限时的“加入圈子”任务，启用功能意味着你愿意自行承担相关风险""" + self.CarrotPull = CarrotPull + """社区拔萝卜，启用功能意味着你愿意自行承担相关风险""" + self.WxSign = WxSign + """微信小程序签到，启用功能意味着你愿意自行承担相关风险""" + + def _password(self): + if len(self.password) == 32: + return self.password + return md5_crypto(self.password) + + def _cookies(self): + if isinstance(self.cookies, str): + return cookies_to_dict(self.cookies) + return self.cookies + + +class OnePush: + """推送配置""" + def __init__(self, notifier="", params=None): + self.notifier = notifier + self.params = params or { + "title": "", + "markdown": False, + "token": "", + "userid": "", + } -class OnePush(BaseModel): - """推送配置""" - notifier: Union[str, bool] = "" - """是否开启消息推送""" - params: Dict = { - "title": "", - "markdown": False, - "token": "", - "userid": "" - } - """推送参数""" - - -class Preference(BaseModel): + +class Preference: """偏好设置""" - geetest_url: str = "" - """极验验证URL""" - geetest_params: Dict = {} - """极验自定义params参数""" - geetest_data: Dict = {} - """极验自定义data参数""" - -class Config(BaseModel): + + def __init__(self, geetest_url="", geetest_params=None, geetest_data=None): + self.geetest_url = geetest_url + self.geetest_params = geetest_params or {} + self.geetest_data = geetest_data or {} + + +class Config: """插件数据""" - preference: Preference = Preference() - """偏好设置""" - accounts: List[Account] = [Account()] - """账号设置""" - ONEPUSH: OnePush = OnePush() - """消息推送""" - -def write_plugin_data(data: Config = None): - """ - 写入插件数据文件 - - :param data: 配置对象 - """ - try: - if data is None: - data = ConfigManager.data_obj - try: - if CONFIG_TYPE == "json": - str_data = orjson.dumps(data.model_dump(), option=orjson.OPT_PASSTHROUGH_DATETIME | orjson.OPT_SERIALIZE_NUMPY | orjson.OPT_INDENT_2) - with open(CONFIG_PATH, "wb") as file: - file.write(str_data) - else: - str_data = yaml.dump(data.model_dump(), indent=4, allow_unicode=True, sort_keys=False) - with open(CONFIG_PATH, "w", encoding="utf-8") as file: - file.write(str_data) - return True - except (AttributeError, TypeError, ValueError): - log.exception("数据对象序列化失败，可能是数据类型错误") - return False - except OSError: - return False + + def __init__(self, preference=None, accounts=None, onepush=None): + self.preference = preference or Preference() + self.accounts = accounts or [Account()] + self.ONEPUSH = onepush or OnePush() + + def to_dict(self): + """将 Config 转换为字典""" + return { + "preference": vars(self.preference), + "accounts": [vars(account) for account in self.accounts], + "ONEPUSH": vars(self.ONEPUSH), + } + + @classmethod + def from_dict(cls, data): + """从字典创建 Config 实例""" + preference = Preference(**data.get("preference", {})) + accounts = [Account(**account) for account in data.get("accounts", [])] + onepush = OnePush(**data.get("ONEPUSH", {})) + return cls(preference, accounts, onepush) class ConfigManager: """配置管理器""" + data_obj = Config() - """加载出的插件数据对象""" - platform = get_platform() - """运行平台""" + platform = "platform_example" # 示例平台 @classmethod def load_config(cls): """ 加载插件数据文件 """ - if os.path.exists(DATA_PATH) and os.path.isfile(CONFIG_PATH): + if os.path.exists(CONFIG_PATH) and os.path.isfile(CONFIG_PATH): try: - with open(CONFIG_PATH, 'r', encoding="utf-8") as file: + with open(CONFIG_PATH, "r", encoding="utf-8") as file: if CONFIG_TYPE == "json": - data = orjson.loads(file.read()) + data = json.load(file) else: data = yaml.safe_load(file) - new_model = Config.model_validate(data) - for attr in new_model.model_fields: - # ConfigManager.data_obj.__setattr__(attr, new_model.__getattribute__(attr)) - setattr(ConfigManager.data_obj, attr, getattr(new_model, attr)) - write_plugin_data(ConfigManager.data_obj) # 同步配置 - except (ValidationError, JSONDecodeError): - log.exception(f"读取数据文件失败，请检查数据文件 {CONFIG_PATH} 格式是否正确") - raise - except Exception: - log.exception( - f"读取数据文件失败，请检查数据文件 {CONFIG_PATH} 是否存在且有权限读取和写入") + # 从文件数据创建 Config 对象 + cls.data_obj = Config.from_dict(data) + cls.write_plugin_data(cls.data_obj) # 同步配置 + except Exception as e: + log.exception(f"读取数据文件失败，请检查文件格式或权限: {e}") raise else: try: if not os.path.exists(DATA_PATH): os.mkdir(DATA_PATH) - write_plugin_data() - except (AttributeError, TypeError, ValueError, PermissionError): - log.exception(f"创建数据文件失败，请检查是否有权限读取和写入 {CONFIG_PATH}") + cls.write_plugin_data(cls.data_obj) # 创建并写入默认数据 + except Exception as e: + log.exception(f"创建数据文件失败，请检查权限: {e}") raise log.info(f"数据文件 {CONFIG_PATH} 不存在，已创建默认数据文件。") + @classmethod + def write_plugin_data(cls, data: Config = None): + """ + 写入插件数据文件 + :param data: 配置对象 + """ + try: + if data is None: + data = cls.data_obj + if CONFIG_TYPE == "json": + with open(CONFIG_PATH, "w", encoding="utf-8") as file: + json.dump(data.to_dict(), file, indent=4, ensure_ascii=False) + else: + with open(CONFIG_PATH, "w", encoding="utf-8") as file: + yaml.dump( + data.to_dict(), + file, + indent=4, + allow_unicode=True, + sort_keys=False, + ) + return True + except (json.JSONDecodeError, yaml.YAMLError) as e: + log.exception(f"写入数据文件失败: {e}") + return False + +# 加载配置 ConfigManager.load_config() diff --git a/utils/data_model.py b/utils/data_model.py index a170a4e3..b880ac0d 100644 --- a/utils/data_model.py +++ b/utils/data_model.py @@ -1,24 +1,25 @@ -"""数据处理模型""" -from typing import (Any, Dict, NamedTuple, Optional) -from pydantic import BaseModel # pylint: disable=no-name-in-module +""" +数据模型 +""" +from typing import Any, Dict, NamedTuple, Optional -class ApiResultHandler(BaseModel): +class ApiResultHandler: """ API返回的数据处理器 """ - content: Dict[str, Any] - """API返回的JSON对象序列化以后的Dict对象""" - data: Dict[str, Any] = {} - """API返回的数据体""" - message: str = "" - """API返回的消息内容""" - status: Optional[int] = None - """API返回的状态码""" - def __init__(self, content: Dict[str, Any]): - super().__init__(content=content) + """ + 初始化处理器，解析API返回的数据。 + :param content: API返回的原始JSON对象 + """ + self.content = content + self.data = self.content.get("data", {}) + self.message = self.content.get("message", "") + self.status = self.content.get("status") + + # 尝试从其他键获取数据 for key in ["data", "entity"]: if self.data == {}: self.data = self.content.get(key, {}) @@ -49,14 +50,6 @@ class LoginResultHandler(ApiResultHandler): """ 登录API返回的数据处理器 """ - pwd: Optional[int] = None - """登录状态""" - pass_token: Optional[str] = None - """登录成功后的passToken""" - location: Optional[str] = None - """登录成功后的跳转地址""" - user_id: Optional[str] = None - def __init__(self, content: Dict[str, Any]): super().__init__(content=content) @@ -97,9 +90,6 @@ class SignResultHandler(ApiResultHandler): 签到API返回的数据处理器 """ - growth: Optional[str] = None - """签到成功后的成长值""" - def __init__(self, content: Dict[str, Any]): super().__init__(content=content) self.growth = self.content.get("entity", {}) @@ -110,7 +100,6 @@ def __init__(self, content: Dict[str, Any]): else: self.growth = None - # pylint: disable=trailing-whitespace def __bool__(self): """ 签到是否成功 @@ -129,11 +118,8 @@ class TokenResultHandler(ApiResultHandler): """ TOKEN数据处理器 """ - token: str = "" - def __init__(self, content: Dict[str, Any]): super().__init__(content=content) - self.token = self.data.get("token", "") @property @@ -152,16 +138,19 @@ class GeetestResult(NamedTuple): validate: str challenge: str -class UserInfoResult(BaseModel): - """用户信息数据""" - title: str = "未知" - """等级名称""" - point: int = 0 - """积分""" +class UserInfoResult: + """用户信息数据""" def __init__(self, **kwargs): + """ + 初始化用户信息对象 + + :param kwargs: 用户信息字典，包含了 'userInfo' 和 'userGrowLevelInfo' + """ if isinstance(kwargs, dict) and kwargs: - kwargs = kwargs.get("userInfo", {}).get("userGrowLevelInfo") - super().__init__(**kwargs) + kwargs = kwargs.get("userInfo", {}).get("userGrowLevelInfo", {}) + self.title = kwargs.get("title", "未知") + self.point = kwargs.get("point", 0) else: - super().__init__() + self.title = "未知" + self.point = 0 diff --git a/utils/request.py b/utils/request.py index 39e551b0..3f02e5f8 100644 --- a/utils/request.py +++ b/utils/request.py @@ -1,11 +1,12 @@ """ Date: 2023-11-12 14:05:06 LastEditors: Night-stars-1 nujj1042633805@gmail.com -LastEditTime: 2023-11-18 00:32:53 +LastEditTime: 2025-01-17 22:42:51 """ + from typing import Any, Dict, Optional -import httpx +import requests from onepush import notify from .config import ConfigManager @@ -14,15 +15,17 @@ _conf = ConfigManager.data_obj -async def get(url: str, - *, - headers: Optional[Dict[str, str]] = None, - params: Optional[Dict[str, Any]] = None, - timeout: Optional[int] = 20, - **kwargs) -> httpx.Response: +def get( + url: str, + *, + headers: Optional[Dict[str, str]] = None, + params: Optional[Dict[str, Any]] = None, + timeout: Optional[int] = 20, + **kwargs, +): """ 说明： - httpx的get请求封装 + get请求封装 参数： :param url: url :param headers: 请求头 @@ -31,23 +34,20 @@ async def get(url: str, :param json: json :param timeout: 超时时间 """ - async with httpx.AsyncClient() as client: - return await client.get(url, - headers=headers, - params=params, - timeout=timeout, - **kwargs) + return requests.get(url, headers=headers, params=params, timeout=timeout, **kwargs) -async def post(url: str, - *, - headers: Optional[Dict[str, str]] = None, - params: Optional[Dict[str, Any]] = None, - timeout: Optional[int] = 20, - **kwargs) -> httpx.Response: +def post( + url: str, + *, + headers: Optional[Dict[str, str]] = None, + params: Optional[Dict[str, Any]] = None, + timeout: Optional[int] = 20, + **kwargs, +): """ 说明： - httpx的post请求封装 + post请求封装 参数： :param url: url :param headers: 请求头 @@ -56,12 +56,7 @@ async def post(url: str, :param json: json :param timeout: 超时时间 """ - async with httpx.AsyncClient() as client: - return await client.post(url, - headers=headers, - params=params, - timeout=timeout, - **kwargs) + return requests.post(url, headers=headers, params=params, timeout=timeout, **kwargs) def notify_me(content=""): @@ -71,6 +66,6 @@ def notify_me(content=""): notifier = _conf.ONEPUSH.notifier params = _conf.ONEPUSH.params if not notifier or not params: - log.error('未配置推送或未正确配置推送') + log.error("未配置推送或未正确配置推送") return False return notify(notifier, content=content, **params) diff --git a/utils/system_info.py b/utils/system_info.py index a0d9b1d8..5eb66b19 100644 --- a/utils/system_info.py +++ b/utils/system_info.py @@ -1,36 +1,44 @@ """ -Date: 2023-11-13 20:29:19 +Author: Night-stars-1 nujj1042633805@gmail.com +Date: 2024-08-20 22:17:31 LastEditors: Night-stars-1 nujj1042633805@gmail.com -LastEditTime: 2024-04-05 22:36:46 +LastEditTime: 2025-01-19 16:48:47 """ + import platform from urllib.request import getproxies + from utils.logger import log def print_info(): """打印系统信息""" log.info("MIUI-AUTO-TASK v1.7.5") - log.info('---------- 系统信息 -------------') + log.info("---------- 系统信息 -------------") system_info() - log.info('---------- 项目信息 -------------') + log.info("---------- 项目信息 -------------") log.info("这是一个免费且开源的项目，如果你是付费购买获得请务必退款") log.info("项目地址：https://github.com/0-8-4/miui-auto-tasks") log.info("欢迎 star，感谢所有项目贡献者，已经提交issues的人，帮助项目发展的人") - log.info('---------- 脚本日志 -------------') + log.info("---------- 脚本日志 -------------") def system_info(): """系统信息""" - log.info(show_info('操作系统平台', platform.platform())) - log.info(show_info('操作系统版本', platform.version())) - log.info(show_info('操作系统名称', platform.system())) - log.info(show_info('操作系统位元', platform.architecture())) - log.info(show_info('操作系统类型', platform.machine())) - log.info(show_info('处理器信息', platform.processor())) - log.info(show_info('Python 版本', str(platform.python_version()) + ' ' + str(platform.python_build()))) + log.info(show_info("操作系统平台", platform.platform())) + log.info(show_info("操作系统版本", platform.version())) + log.info(show_info("操作系统名称", platform.system())) + log.info(show_info("操作系统位元", platform.architecture())) + log.info(show_info("操作系统类型", platform.machine())) + log.info(show_info("处理器信息", platform.processor())) + log.info( + show_info( + "Python 版本", + str(platform.python_version()) + " " + str(platform.python_build()), + ) + ) if getproxies(): - log.info(show_info('系统代理', getproxies())) + log.info(show_info("系统代理", getproxies())) def show_info(tip: str, info: str): diff --git a/utils/utils.py b/utils/utils.py index a589ba5e..5d594a44 100644 --- a/utils/utils.py +++ b/utils/utils.py @@ -12,7 +12,6 @@ from cryptography.hazmat.primitives import padding, serialization from cryptography.hazmat.primitives.asymmetric.padding import PKCS1v15 from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes -from pydantic import ValidationError from tenacity import RetryError, Retrying, stop_after_attempt from .captcha import get_validate @@ -78,7 +77,7 @@ def rsa_encrypt(public_key_pem: str, data: str) -> base64: return base64.b64encode(ciphertext).decode('utf-8') -IncorrectReturn = (KeyError, TypeError, AttributeError, IndexError, ValidationError) +IncorrectReturn = (KeyError, TypeError, AttributeError, IndexError) """API返回数据无效会触发的异常组合""" @@ -92,14 +91,14 @@ def is_incorrect_return(exception: Exception, *addition_exceptions: Type[Excepti return isinstance(exception, exceptions) or isinstance(exception.__cause__, exceptions) -async def get_token_by_captcha(url: str) -> str | bool: +def get_token_by_captcha(url: str) -> str | bool: """通过人机验证码获取TOKEN""" try: parsed_url = urlparse(url) query_params = dict(parse_qsl(parsed_url.query)) # 解析URL参数 gt = query_params.get("c") # pylint: disable=invalid-name challenge = query_params.get("l") - geetest_data = await get_validate(gt, challenge) + geetest_data = get_validate(gt, challenge) params = { 'k': '3dc42a135a8d45118034d1ab68213073', 'locale': 'zh_CN', @@ -112,7 +111,7 @@ async def get_token_by_captcha(url: str) -> str | bool: 'seccode': f'{geetest_data.validate}|jordan', } - response = await post('https://verify.sec.xiaomi.com/captcha/v2/gt/dk/verify', params=params, headers=headers, + response = post('https://verify.sec.xiaomi.com/captcha/v2/gt/dk/verify', params=params, headers=headers, data=data) log.debug(response.text) result = response.json() @@ -131,7 +130,7 @@ async def get_token_by_captcha(url: str) -> str | bool: # pylint: disable=trailing-whitespace -async def get_token(uid: str) -> str | bool: +def get_token(uid: str) -> str | bool: """获取TOKEN""" try: for attempt in Retrying(stop=stop_after_attempt(3)): @@ -215,7 +214,7 @@ async def get_token(uid: str) -> str | bool: 'd': aes_encrypt(key, str(data)), 'a': 'GROW_UP_CHECKIN', } - response = await post('https://verify.sec.xiaomi.com/captcha/v2/data', params=params, headers=headers, + response = post('https://verify.sec.xiaomi.com/captcha/v2/data', params=params, headers=headers, data=data) log.debug(response.text) result = response.json() @@ -225,7 +224,7 @@ async def get_token(uid: str) -> str | bool: elif api_data.need_verify: log.error("遇到人机验证码, 尝试调用解决方案") url = api_data.data.get("url") - if token := await get_token_by_captcha(url): + if token := get_token_by_captcha(url): return token else: raise ValueError("人机验证失败") @@ -258,4 +257,3 @@ def generate_qrcode(url): line = "".join(chaes[pixel] for pixel in row) print(line) log.debug(line) - \ No newline at end of file