chore: add test rules & type hint

ayugespidertools/scraper/http/request/form.py

@@ -1,4 +1,4 @@
-from typing import TYPE_CHECKING, Optional, Union
+from typing import TYPE_CHECKING, Any, Callable, Iterable, List, Optional, Tuple, Union
 
 from scrapy import FormRequest
 
@@ -11,20 +11,23 @@
 if TYPE_CHECKING:
     from ayugespidertools.common.typevars import AiohttpRequestArgs
 
+FormdataKVType = Tuple[str, Union[str, Iterable[str]]]
+FormdataType = Optional[Union[dict, List[FormdataKVType]]]
+
 
 class AiohttpFormRequest(AiohttpRequest, FormRequest):
     """使用 aiohttp 发送 FormRequest 请求"""
 
     def __init__(
         self,
-        url=None,
-        callback=None,
-        method=None,
-        formdata=None,
-        body=None,
+        url: str = None,
+        callback: Optional[Callable] = None,
+        method: Optional[str] = None,
+        formdata: FormdataType = None,
+        body: Optional[Union[bytes, str]] = None,
         args: Optional[Union["AiohttpRequestArgs", dict]] = None,
-        **kwargs
-    ):
+        **kwargs: Any
+    ) -> None:
         # First init FormRequest to get url, body and method
         if formdata:
             FormRequest.__init__(self, url=url, method=method, formdata=formdata)

ayugespidertools/scraper/middlewares/headers/ua.py

@@ -37,7 +37,7 @@ def from_crawler(cls, crawler: "Crawler") -> "Self":
         crawler.signals.connect(s.spider_opened, signal=signals.spider_opened)
         return s
 
-    def spider_opened(self, spider: "AyuSpider"):
+    def spider_opened(self, spider: "AyuSpider") -> None:
         # 带权重的 ua 列表，这里是根据 fake_useragent 库中的打印信息来规划权重的。
         ua_arr = [
             {"explorer": "opera", "weight": 16},

ayugespidertools/scraper/middlewares/proxy/dynamic.py

@@ -1,10 +1,19 @@
 import base64
+from typing import TYPE_CHECKING
 
 from scrapy import signals
 
 from ayugespidertools.common.multiplexing import ReuseOperation
 from ayugespidertools.common.params import Param
 
+if TYPE_CHECKING:
+    from scrapy import Request
+    from scrapy.crawler import Crawler
+    from scrapy.settings import Settings
+    from typing_extensions import Self
+
+    from ayugespidertools.spiders import AyuSpider
+
 
 class DynamicProxyDownloaderMiddleware:
     """动态隧道代理中间件"""
@@ -15,12 +24,12 @@ def __init__(self):
         self.password = None
 
     @classmethod
-    def from_crawler(cls, crawler):
+    def from_crawler(cls, crawler: "Crawler") -> "Self":
         s = cls()
         crawler.signals.connect(s.spider_opened, signal=signals.spider_opened)
         return s
 
-    def process_request(self, request, spider):
+    def process_request(self, request: "Request", spider: "AyuSpider") -> None:
         # TODO: 根据权重来随机获取一个账号 DYNAMIC_PROXY_CONFIG
         # account = ReuseOperation.random_weight(self.account_arr)
         if request.url.startswith("https://"):
@@ -41,7 +50,7 @@ def process_request(self, request, spider):
         # 采用 gzip 压缩加速访问
         request.headers["Accept-Encoding"] = "gzip"
 
-    def spider_opened(self, spider):
+    def spider_opened(self, spider: "AyuSpider") -> None:
         spider.slog.info(
             f"动态隧道代理中间件: DynamicProxyDownloaderMiddleware 已开启，生效脚本为: {spider.name}"
         )
@@ -54,7 +63,7 @@ def spider_opened(self, spider):
 class AbuDynamicProxyDownloaderMiddleware:
     """阿布云动态代理 - 隧道验证方式（其实和快代理的写法一致）"""
 
-    def __init__(self, settings):
+    def __init__(self, settings: "Settings") -> None:
         dynamic_proxy_conf = settings.get("DYNAMIC_PROXY_CONFIG", None)
         # 查看动态隧道代理配置是否符合要求
         is_match = ReuseOperation.is_dict_meet_min_limit(
@@ -70,17 +79,17 @@ def __init__(self, settings):
         self.password = dynamic_proxy_conf["password"]
 
     @classmethod
-    def from_crawler(cls, crawler):
+    def from_crawler(cls, crawler: "Crawler") -> "Self":
         s = cls(crawler.settings)
         crawler.signals.connect(s.spider_opened, signal=signals.spider_opened)
         return s
 
-    def spider_opened(self, spider):
+    def spider_opened(self, spider: "AyuSpider") -> None:
         spider.slog.info(
             f"阿布云动态隧道代理中间件: AbuDynamicProxyDownloaderMiddleware 已开启，生效脚本为: {spider.name}"
         )
 
-    def process_request(self, request, spider):
+    def process_request(self, request: "Request", spider: "AyuSpider") -> None:
         if request.url.startswith("https://"):
             request.meta["proxy"] = f"https://{self.proxy_url}"
         elif request.url.startswith("http://"):

ayugespidertools/scraper/middlewares/proxy/exclusive.py

@@ -1,4 +1,5 @@
 import base64
+from typing import TYPE_CHECKING
 
 import requests
 from scrapy import signals
@@ -7,6 +8,13 @@
     "ExclusiveProxyDownloaderMiddleware",
 ]
 
+if TYPE_CHECKING:
+    from scrapy import Request
+    from scrapy.crawler import Crawler
+    from typing_extensions import Self
+
+    from ayugespidertools.spiders import AyuSpider
+
 
 class ExclusiveProxyDownloaderMiddleware:
     """独享代理中间件"""
@@ -20,7 +28,7 @@ def __init__(self):
         self.proxy = None
 
     @classmethod
-    def from_crawler(cls, crawler):
+    def from_crawler(cls, crawler: "Crawler") -> "Self":
         s = cls()
         crawler.signals.connect(s.spider_opened, signal=signals.spider_opened)
         return s
@@ -39,7 +47,7 @@ def get_proxy_ip(self, proxy_url: str, index: int) -> str:
         except Exception:
             raise Exception("获取独享代理时失败，请查看独享配置及网络是否正常。")
 
-    def process_request(self, request, spider):
+    def process_request(self, request: "Request", spider: "AyuSpider") -> None:
         if request.url.startswith("https://"):
             request.meta["proxy"] = f"https://{self.proxy}"
         elif request.url.startswith("http://"):
@@ -53,7 +61,7 @@ def process_request(self, request, spider):
         ).decode("utf8")
         request.headers["Proxy-Authorization"] = encoded_user_pass
 
-    def spider_opened(self, spider):
+    def spider_opened(self, spider: "AyuSpider") -> None:
         spider.slog.info(
             f"独享代理中间件: ExclusiveProxyDownloaderMiddleware 已开启，生效脚本为: {spider.name}"
         )

pyproject.toml

@@ -125,6 +125,7 @@ exclude_lines = [
     "pragma: no cover",
     "raise AssertionError",
     "raise NotImplementedError",
+    "if TYPE_CHECKING:",
     "__repr__",
     "__str__",
 ]