feat: 快代理实现

config/base_config.py

@@ -13,7 +13,7 @@ ENABLE_IP_PROXY = False
 IP_PROXY_POOL_COUNT = 2
 
 # 代理IP提供商名称
-IP_PROXY_PROVIDER_NAME = "jishuhttp"
+IP_PROXY_PROVIDER_NAME = "kuaidaili"
 
 # 设置为True不会打开浏览器（无头浏览器），设置False会打开一个浏览器（小红书如果一直扫码登录不通过，打开浏览器手动过一下滑动验证码）
 HEADLESS = True

proxy/providers/kuaidl_proxy.py

@@ -2,14 +2,120 @@
 # @Author  : relakkes@gmail.com
 # @Time    : 2024/4/5 09:43
 # @Desc    : 快代理HTTP实现，官方文档：https://www.kuaidaili.com/?ref=ldwkjqipvz6c
+import os
+import re
 from typing import Dict, List
 
+import httpx
+from pydantic import BaseModel, Field
+
 from proxy import IpGetError, IpInfoModel, ProxyProvider, RedisDbIpCache
+from proxy.types import ProviderNameEnum
+from tools import utils
+
+
+class KuaidailiProxyModel(BaseModel):
+    ip: str = Field("ip")
+    port: int = Field("端口")
+    expire_ts: int = Field("过期时间")
+
+
+def parse_kuaidaili_proxy(proxy_info: str) -> KuaidailiProxyModel:
+    """
+    解析快代理的IP信息
+    Args:
+        proxy_info:
+
+    Returns:
+
+    """
+    proxies: List[str] = proxy_info.split(":")
+    if len(proxies) != 2:
+        raise Exception("not invalid kuaidaili proxy info")
+
+    pattern = r'(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}):(\d{1,5}),(\d+)'
+    match = re.search(pattern, proxy_info)
+    if not match.groups():
+        raise Exception("not match kuaidaili proxy info")
+
+    return KuaidailiProxyModel(
+        ip=match.groups()[0],
+        port=int(match.groups()[1]),
+        expire_ts=int(match.groups()[2])
+    )
 
 
 class KuaiDaiLiProxy(ProxyProvider):
-    async def get_proxies(self, num: int) -> List[Dict]:
+    def __init__(self, kdl_user_name: str, kdl_user_pwd: str):
-        pass
+        """
+
+        Args:
+            kdl_user_name:
+            kdl_user_pwd:
+        """
+        self.kdl_user_name = kdl_user_name
+        self.kdl_user_pwd = kdl_user_pwd
+        self.api_base = "https://dps.kdlapi.com/"
+        self.ip_cache = RedisDbIpCache()
+        self.proxy_brand_name = ProviderNameEnum.KUAI_DAILI_PROVIDER.value
+        self.params = {
+            "secret_id": "onl8885qqpz1r85uhmud",
+            "signature": "kq929uc9w1awnyevd5v5njnnu8kgl6pw",
+            "pt": 1,
+            "format": "json",
+            "sep": 1,
+            "f_et": 1,
+        }
+
+    async def get_proxies(self, num: int) -> List[IpInfoModel]:
+        """
+        快代理实现
+        Args:
+            num:
+
+        Returns:
+
+        """
+        uri = "/api/getdps/"
+
+        # 优先从缓存中拿 IP
+        ip_cache_list = self.ip_cache.load_all_ip(proxy_brand_name=self.proxy_brand_name)
+        if len(ip_cache_list) >= num:
+            return ip_cache_list[:num]
+
+        # 如果缓存中的数量不够，从IP代理商获取补上，再存入缓存中
+        need_get_count = num - len(ip_cache_list)
+        self.params.update({"num": need_get_count})
+
+        ip_infos: List[IpInfoModel] = []
+        async with httpx.AsyncClient() as client:
+            response = await client.get(self.api_base + uri, params=self.params)
+
+            if response.status_code != 200:
+                utils.logger.error(f"[KuaiDaiLiProxy.get_proxies] statuc code not 200 and response.txt:{response.text}")
+                raise Exception("get ip error from proxy provider and status code not 200 ...")
+
+            ip_response: Dict = response.json()
+            if ip_response.get("code") != 0:
+                utils.logger.error(f"[KuaiDaiLiProxy.get_proxies]  code not 0 and msg:{ip_response.get('msg')}")
+                raise Exception("get ip error from proxy provider and  code not 0 ...")
+
+            proxy_list: List[str] = ip_response.get("data", {}).get("proxy_list")
+            for proxy in proxy_list:
+                proxy_model = parse_kuaidaili_proxy(proxy)
+                ip_info_model = IpInfoModel(
+                    ip=proxy_model.ip,
+                    port=proxy_model.port,
+                    user=self.kdl_user_name,
+                    password=self.kdl_user_pwd,
+                    expired_time_ts=proxy_model.expire_ts,
+
+                )
+                ip_key = f"{self.proxy_brand_name}_{ip_info_model.ip}_{ip_info_model.port}"
+                self.ip_cache.set_ip(ip_key, ip_info_model.model_dump_json(), ex=ip_info_model.expired_time_ts)
+                ip_infos.append(ip_info_model)
+
+        return ip_cache_list + ip_infos
 
 
 def new_kuai_daili_proxy() -> KuaiDaiLiProxy:
@@ -18,4 +124,7 @@ def new_kuai_daili_proxy() -> KuaiDaiLiProxy:
     Returns:
 
     """
-    return KuaiDaiLiProxy()
+    return KuaiDaiLiProxy(
+        kdl_user_name=os.getenv("kdl_user_name", "你的快代理用户名"),
+        kdl_user_pwd=os.getenv("kdl_user_pwd", "你的快代理密码"),
+    )