feat: 完善类型注释，增加 mypy 类型检测

media_platform/xhs/client.py

@@ -17,9 +17,10 @@ class XHSClient:
             self,
             timeout=10,
             proxies=None,
-            headers: Optional[Dict] = None,
-            playwright_page: Page = None,
-            cookie_dict: Dict = None
+            *,
+            headers: Dict[str, str],
+            playwright_page: Page,
+            cookie_dict: Dict[str, str],
     ):
         self.proxies = proxies
         self.timeout = timeout
@@ -51,21 +52,21 @@ class XHSClient:
         self.headers.update(headers)
         return self.headers
 
-    async def request(self, method, url, **kwargs):
+    async def request(self, method, url, **kwargs) -> Dict:
         async with httpx.AsyncClient(proxies=self.proxies) as client:
             response = await client.request(
                 method, url, timeout=self.timeout,
                 **kwargs
             )
-        data = response.json()
+        data: Dict = response.json()
         if data["success"]:
-            return data.get("data", data.get("success"))
+            return data.get("data", data.get("success", {}))
         elif data["code"] == self.IP_ERROR_CODE:
             raise IPBlockError(self.IP_ERROR_STR)
         else:
             raise DataFetchError(data.get("msg", None))
 
-    async def get(self, uri: str, params=None):
+    async def get(self, uri: str, params=None) -> Dict:
         final_uri = uri
         if isinstance(params, dict):
             final_uri = (f"{uri}?"
@@ -73,7 +74,7 @@ class XHSClient:
         headers = await self._pre_headers(final_uri)
         return await self.request(method="GET", url=f"{self._host}{final_uri}", headers=headers)
 
-    async def post(self, uri: str, data: dict):
+    async def post(self, uri: str, data: dict) -> Dict:
         headers = await self._pre_headers(uri, data)
         json_str = json.dumps(data, separators=(',', ':'), ensure_ascii=False)
         return await self.request(method="POST", url=f"{self._host}{uri}",
@@ -86,7 +87,7 @@ class XHSClient:
         try:
             note_card: Dict = await self.get_note_by_id(note_id)
             return note_card.get("note_id") == note_id
-        except DataFetchError:
+        except Exception:
             return False
 
     async def update_cookies(self, browser_context: BrowserContext):
@@ -128,7 +129,8 @@ class XHSClient:
         data = {"source_note_id": note_id}
         uri = "/api/sns/web/v1/feed"
         res = await self.post(uri, data)
-        return res["items"][0]["note_card"]
+        res_dict: Dict = res["items"][0]["note_card"]
+        return res_dict
 
     async def get_note_comments(self, note_id: str, cursor: str = "") -> Dict:
         """get note comments

media_platform/xhs/core.py

@@ -21,15 +21,15 @@ from base.proxy_account_pool import AccountPool
 
 
 class XiaoHongShuCrawler(AbstractCrawler):
+    context_page: Page
+    browser_context: BrowserContext
+    xhs_client: XHSClient
+    account_pool: AccountPool
 
     def __init__(self):
-        self.browser_context: Optional[BrowserContext] = None
-        self.context_page: Optional[Page] = None
-        self.user_agent = utils.get_user_agent()
-        self.xhs_client: Optional[XHSClient] = None
         self.index_url = "https://www.xiaohongshu.com"
-        self.command_args: Optional[Namespace] = None
-        self.account_pool: Optional[AccountPool] = None
+        self.command_args: Optional[Namespace] = None # type: ignore
+        self.user_agent = utils.get_user_agent()
 
     def init_config(self, **kwargs):
         for key, value in kwargs.items():
@@ -69,7 +69,7 @@ class XiaoHongShuCrawler(AbstractCrawler):
 
             utils.logger.info("Xhs Crawler finished ...")
 
-    async def search_posts(self):
+    async def search_posts(self) -> None:
         """Search for notes and retrieve their comment information."""
         utils.logger.info("Begin search xiaohongshu keywords")
 
@@ -86,7 +86,7 @@ class XiaoHongShuCrawler(AbstractCrawler):
                 _semaphore = asyncio.Semaphore(config.MAX_CONCURRENCY_NUM)
                 task_list = [
                     self.get_note_detail(post_item.get("id"), _semaphore)
-                    for post_item in posts_res.get("items")
+                    for post_item in posts_res.get("items", {})
                 ]
                 note_details = await asyncio.gather(*task_list)
                 for note_detail in note_details:
@@ -170,18 +170,18 @@ class XiaoHongShuCrawler(AbstractCrawler):
         if config.SAVE_LOGIN_STATE:
             # feat issue #14
             # we will save login state to avoid login every time
-            user_data_dir = os.path.join(os.getcwd(), "browser_data", config.USER_DATA_DIR % self.command_args.platform)
+            user_data_dir = os.path.join(os.getcwd(), "browser_data", config.USER_DATA_DIR % self.command_args.platform) # type: ignore
             browser_context = await chromium.launch_persistent_context(
                 user_data_dir=user_data_dir,
                 accept_downloads=True,
                 headless=headless,
-                proxy=playwright_proxy,
+                proxy=playwright_proxy, # type: ignore
                 viewport={"width": 1920, "height": 1080},
                 user_agent=user_agent
             )
             return browser_context
         else:
-            browser = await chromium.launch(headless=headless, proxy=playwright_proxy)
+            browser = await chromium.launch(headless=headless, proxy=playwright_proxy) # type: ignore
             browser_context = await browser.new_context(
                 viewport={"width": 1920, "height": 1080},
                 user_agent=user_agent

media_platform/xhs/help.py

@@ -92,7 +92,7 @@ def mrc(e):
     ]
     o = -1
 
-    def right_without_sign(num, bit=0) -> int:
+    def right_without_sign(num: int, bit: int=0) -> int:
         val = ctypes.c_uint32(num).value >> bit
         MAX32INT = 4294967295
         return (val + (MAX32INT + 1)) % (2 * (MAX32INT + 1)) - MAX32INT - 1

media_platform/xhs/login.py

@@ -24,8 +24,8 @@ class XHSLogin(AbstractLogin):
                  login_type: str,
                  browser_context: BrowserContext,
                  context_page: Page,
-                 login_phone: str = None,
-                 cookie_str: str = None
+                 login_phone: str = "",
+                 cookie_str: str = ""
                  ):
         self.login_type = login_type
         self.browser_context = browser_context