增加了代理

2026-06-21 22:30:30 +08:00
parent 1ce867f3a8
commit 828f9889ed
10 changed files with 317 additions and 49 deletions
@@ -9,12 +9,20 @@ accounts:
    email_imap_server: "mail.bdhg.xyz"
    email_imap_port: 993

-# 代理配置（可选）
+# 代理配置
 proxy:
-  enabled: false
+  enabled: true
+  # 代理API地址（自动获取代理IP）
+  api_url: "http://api.xiequ.cn/VAD/GetIp.aspx?act=get&uid=106015&vkey=97111DB5379E38E3BC2FF09A1B00A0C7&num=1&time=30&plat=1&re=0&type=0&so=1&ow=1&spl=1&addr=&db=1"
+  # 静态代理（如果不使用API获取）
  http: ""
  https: ""

+# 极验配置
+geetest:
+  # 最大重试次数（遇到点选验证码时重试）
+  max_retries: 5
+
 # 日志配置
 log:
  level: "DEBUG"
@@ -3,5 +3,6 @@
 from .login import DouyuLogin
 from .email_verifier import EmailVerifier
 from .config import Config
+from .proxy import ProxyManager

-__all__ = ["DouyuLogin", "EmailVerifier", "Config"]
+__all__ = ["DouyuLogin", "EmailVerifier", "Config", "ProxyManager"]
@@ -21,6 +21,7 @@ class Account:
 class ProxyConfig:
    """代理配置"""
    enabled: bool = False
+    api_url: str = ""
    http: str = ""
    https: str = ""

@@ -59,10 +60,15 @@ class Config:
        proxy = self._config.get('proxy', {})
        return ProxyConfig(
            enabled=proxy.get('enabled', False),
+            api_url=proxy.get('api_url', ''),
            http=proxy.get('http', ''),
            https=proxy.get('https', ''),
        )

+    def get_geetest_config(self) -> dict:
+        """获取极验配置"""
+        return self._config.get('geetest', {'max_retries': 5})
+
    def get_cookie_dir(self) -> str:
        """获取Cookie存储目录"""
        return self._config.get('cookie_dir', 'data/cookies')
@@ -12,6 +12,7 @@ from loguru import logger
 from .config import Account
 from .crypto import encrypt_password, encrypt_nickname_or_phone
 from .email_verifier import EmailVerifier
+from .proxy import ProxyManager, get_proxy_manager

 # 导入geetest滑块模块
 import sys
@@ -49,12 +50,19 @@ class DouyuLogin:
        self,
        account: Account,
        proxy: Optional[str | Mapping[str, str]] = None,
+        proxy_api_url: Optional[str] = None,
        timeout: tuple[float, float] = REQUEST_TIMEOUT,
+        max_geetest_retries: int = 5,
    ):
        self.account = account
        self.proxy = proxy
        self.timeout = timeout
+        self.max_geetest_retries = max_geetest_retries
        self.session = requests.Session()
+
+        # 初始化代理管理器
+        self.proxy_manager = get_proxy_manager(proxy_api_url)
+
        self._setup_session()

    def _setup_session(self) -> None:
@@ -71,7 +79,19 @@ class DouyuLogin:
            'X-Requested-With': 'XMLHttpRequest',
        })

-        if self.proxy:
+        # 设置代理
+        self._apply_proxy()
+
+    def _apply_proxy(self, proxy: str = None) -> None:
+        """应用代理到Session"""
+        if proxy:
+            # 使用指定的代理
+            self.session.proxies = {
+                'http': proxy,
+                'https': proxy,
+            }
+        elif self.proxy:
+            # 使用配置的代理
            if isinstance(self.proxy, str):
                self.session.proxies = {
                    'http': self.proxy,
@@ -83,6 +103,22 @@ class DouyuLogin:
                    for scheme, url in self.proxy.items()
                    if url
                }
+        else:
+            # 从代理管理器获取代理
+            new_proxy = self.proxy_manager.get_proxy()
+            if new_proxy:
+                self.session.proxies = {
+                    'http': new_proxy,
+                    'https': new_proxy,
+                }
+
+    def _refresh_proxy(self) -> Optional[str]:
+        """刷新代理IP"""
+        new_proxy = self.proxy_manager.get_proxy()
+        if new_proxy:
+            self._apply_proxy(new_proxy)
+            logger.info(f"已切换代理: {new_proxy}")
+        return new_proxy

    def _safe_url(self, url: str) -> str:
        """隐藏查询参数，避免日志泄露登录回调 code。"""
@@ -226,62 +262,101 @@ class DouyuLogin:

    def _solve_geetest(self, gt: str, challenge: str) -> Tuple[str, str]:
        """
-        解决极验滑块验证
+        解决极验滑块验证（带重试机制）

        Args:
            gt: 极验gt参数
-            challenge: 极验challenge参数
+            challenge: 极验challenge参数（第一次登录返回的）

        Returns:
            (validate, seccode)
        """
        logger.info("开始极验滑块验证...")

-        # 使用geetest-v3-silde-crack的完整流程
-        str_16 = _generate_seed()
+        for attempt in range(self.max_geetest_retries):
+            try:
+                logger.info(f"极验验证尝试 {attempt + 1}/{self.max_geetest_retries}")

-        # 获取JS地址
-        get_js_address(gt)
+                # 使用geetest-v3-silde-crack的完整流程
+                str_16 = _generate_seed()

-        # 获取第一个w值
-        w1 = get_w1(gt, challenge, str_16)
+                # 获取JS地址
+                get_js_address(gt)

-        # 获取c和s
-        c, s = get_c_s(gt, challenge, w1)
+                # 获取第一个w值
+                w1 = get_w1(gt, challenge, str_16)

-        # 获取第二个w值
-        w2 = get_w2(gt, challenge, c, s, str_16)
+                # 获取c和s
+                c, s = get_c_s(gt, challenge, w1)

-        # 请求滑块
-        req_slide(gt, challenge, w2)
+                # 获取第二个w值
+                w2 = get_w2(gt, challenge, c, s, str_16)

-        # 获取图片
-        bg, fullbg, c, s, slice, challenge_new = get_picture(gt, challenge)
+                # 请求滑块
+                req_slide(gt, challenge, w2)

-        # 识别缺口位置
-        hkjl = download_picture(bg, fullbg, slice)
+                # 获取图片
+                result = get_picture(gt, challenge)

-        # 获取第三个w值
-        w3 = get_w3(str_16, challenge_new, hkjl, c, s, gt)
+                # 检查返回的数据结构
+                if len(result) == 6:
+                    bg, fullbg, c, s, slice, challenge_new = result
+                else:
+                    # 如果返回的数据结构不对，可能是点选验证码，需要重试
+                    logger.warning(f"返回数据结构异常，可能是点选验证码，重试中...")
+                    # 刷新代理
+                    self._refresh_proxy()
+                    time.sleep(1)
+                    continue

-        # 最后验证
-        result = req_end(gt, challenge_new, w3)
+                # 检查是否是滑块验证码（bg和slice不为空）
+                if not bg or not slice:
+                    logger.warning(f"未获取到滑块图片，可能是点选验证码，重试中...")
+                    # 刷新代理
+                    self._refresh_proxy()
+                    time.sleep(1)
+                    continue

-        # 从req_end的响应中提取validate
-        # req_end返回的是验证结果，成功时包含validate
-        if isinstance(result, dict):
-            validate = result.get('validate', '')
-        else:
-            # 如果返回的是字符串，可能是直接的validate值
-            validate = str(result)
+                # 识别缺口位置
+                hkjl = download_picture(bg, fullbg, slice)

-        if not validate:
-            raise ValueError("极验验证失败，未获取到validate")
+                # 获取第三个w值
+                w3 = get_w3(str_16, challenge_new, hkjl, c, s, gt)

-        seccode = f"{validate}|jordan"
+                # 最后验证
+                result = req_end(gt, challenge_new, w3)

-        logger.success(f"极验滑块验证成功! validate={validate[:20]}...")
-        return validate, seccode
+                # 从req_end的响应中提取validate
+                if isinstance(result, dict):
+                    validate = result.get('validate', '')
+                    success = result.get('success', 0)
+                    message = result.get('message', '')
+
+                    if success == 1 and validate:
+                        seccode = f"{validate}|jordan"
+                        logger.success(f"极验滑块验证成功! validate={validate[:20]}...")
+                        return validate, seccode
+                    else:
+                        logger.warning(f"极验验证失败: {message}，重试中...")
+                        # 刷新代理
+                        self._refresh_proxy()
+                        time.sleep(1)
+                        continue
+                else:
+                    validate = str(result)
+                    if validate:
+                        seccode = f"{validate}|jordan"
+                        logger.success(f"极验滑块验证成功! validate={validate[:20]}...")
+                        return validate, seccode
+
+            except Exception as e:
+                logger.warning(f"极验验证异常: {e}，重试中...")
+                # 刷新代理
+                self._refresh_proxy()
+                time.sleep(1)
+                continue
+
+        raise ValueError(f"极验验证失败，已重试 {self.max_geetest_retries} 次")

    def _second_login(self, gt: str, challenge: str, validate: str,
                      seccode: str, code_token: str) -> str:
@@ -0,0 +1,115 @@
+"""代理管理模块"""
+
+import re
+import requests
+from typing import Optional, List
+from loguru import logger
+
+
+class ProxyManager:
+    """代理管理器"""
+
+    def __init__(self, api_url: str = None):
+        self.api_url = api_url or "http://api.xiequ.cn/VAD/GetIp.aspx?act=get&uid=106015&vkey=97111DB5379E38E3BC2FF09A1B00A0C7&num=1&time=30&plat=1&re=0&type=0&so=1&ow=1&spl=1&addr=&db=1"
+        self.current_proxy: Optional[str] = None
+
+    def get_proxy(self) -> Optional[str]:
+        """
+        从代理API获取代理IP
+
+        Returns:
+            代理URL，格式: http://ip:port
+        """
+        try:
+            logger.info("获取代理IP...")
+            response = requests.get(self.api_url, timeout=10)
+            response.raise_for_status()
+
+            text = response.text.strip()
+            logger.debug(f"代理API响应: {text}")
+
+            # 解析IP:Port格式
+            match = re.search(r'(\d+\.\d+\.\d+\.\d+):(\d+)', text)
+            if match:
+                ip = match.group(1)
+                port = match.group(2)
+                proxy = f"http://{ip}:{port}"
+                self.current_proxy = proxy
+                logger.info(f"获取到代理: {proxy}")
+                return proxy
+            else:
+                logger.warning(f"无法解析代理地址: {text}")
+                return None
+
+        except Exception as e:
+            logger.error(f"获取代理失败: {e}")
+            return None
+
+    def get_proxies_dict(self, proxy: str = None) -> dict:
+        """
+        获取requests使用的proxies字典
+
+        Args:
+            proxy: 代理URL，如果不提供则使用当前代理
+
+        Returns:
+            proxies字典
+        """
+        proxy = proxy or self.current_proxy
+        if proxy:
+            return {
+                'http': proxy,
+                'https': proxy,
+            }
+        return {}
+
+    def verify_proxy(self, proxy: str = None) -> bool:
+        """
+        验证代理是否可用
+
+        Args:
+            proxy: 代理URL
+
+        Returns:
+            是否可用
+        """
+        proxy = proxy or self.current_proxy
+        if not proxy:
+            return False
+
+        try:
+            response = requests.get(
+                'https://httpbin.org/ip',
+                proxies={'http': proxy, 'https': proxy},
+                timeout=10
+            )
+            if response.status_code == 200:
+                data = response.json()
+                logger.info(f"代理验证成功，当前IP: {data.get('origin')}")
+                return True
+            return False
+        except Exception as e:
+            logger.error(f"代理验证失败: {e}")
+            return False
+
+
+# 全局代理管理器实例
+_proxy_manager: Optional[ProxyManager] = None
+
+
+def get_proxy_manager(api_url: str = None) -> ProxyManager:
+    """获取全局代理管理器实例"""
+    global _proxy_manager
+    if _proxy_manager is None:
+        _proxy_manager = ProxyManager(api_url)
+    return _proxy_manager
+
+
+def get_proxy() -> Optional[str]:
+    """获取代理URL的便捷函数"""
+    return get_proxy_manager().get_proxy()
+
+
+def get_proxies_dict() -> dict:
+    """获取proxies字典的便捷函数"""
+    return get_proxy_manager().get_proxies_dict()
@@ -132,16 +132,30 @@ def get_c_s(gt:str, challenge:str, w:str) -> Tuple[list[int],str]:
    }

    response = requests.get(
-        'https://apiv6.geetest.com/get.php?gt=' + gt + '&challenge=' + challenge + '&lang=zh-cn&pt=0&client_type=web&w=' + w + '&callback=geetest_' + str(
+        'https://api.geetest.com/get.php?gt=' + gt + '&challenge=' + challenge + '&lang=zh-cn&pt=0&client_type=web&w=' + w + '&callback=geetest_' + str(
            int(round(time.time() * 1000))),
        headers=headers,
        timeout=REQUEST_TIMEOUT,
    )
    response.raise_for_status()
+
+    # 检查是否是错误响应
+    if response.text.startswith('{'):
+        # 直接JSON响应
+        data = json.loads(response.text)
+        if 'status' in data and data['status'] == 'error':
+            raise ValueError(f"极验请求失败: {data.get('user_error', data.get('error', '未知错误'))}")
+
    match = re.search(r'\((.*)\)$', response.text)
    if match:
        json_str = match.group(1)
-        return json.loads(json_str)['data']['c'], json.loads(json_str)['data']['s']
+        data = json.loads(json_str)
+
+        # 检查是否是错误响应
+        if 'status' in data and data['status'] == 'error':
+            raise ValueError(f"极验请求失败: {data.get('user_error', data.get('error', '未知错误'))}")
+
+        return data['data']['c'], data['data']['s']
    else:
        raise ValueError("无法解析 JSONP 响应")

@@ -212,8 +226,30 @@ def get_picture(gt:str, challenge:str) -> tuple[str, str, list[int], str, str, s
    match = re.search(r'\((.*)\)$', response.text)
    if match:
        json_str = match.group(1)
-        return json.loads(json_str)['bg'], json.loads(json_str)['fullbg'], json.loads(json_str)['c'], \
-        json.loads(json_str)['s'], json.loads(json_str)['slice'], json.loads(json_str)['challenge']
+        data = json.loads(json_str)
+
+        # 检查验证码类型
+        if 'data' in data and isinstance(data['data'], dict):
+            # 新版极验格式
+            inner_data = data['data']
+            return (
+                inner_data.get('bg', ''),
+                inner_data.get('fullbg', ''),
+                inner_data.get('c', []),
+                inner_data.get('s', ''),
+                inner_data.get('slice', ''),
+                inner_data.get('challenge', '')
+            )
+        else:
+            # 旧版极验格式
+            return (
+                data.get('bg', ''),
+                data.get('fullbg', ''),
+                data.get('c', []),
+                data.get('s', ''),
+                data.get('slice', ''),
+                data.get('challenge', '')
+            )
    else:
        raise ValueError("无法解析 JSONP 响应")

@@ -245,6 +281,17 @@ def req_end(gt:str, challenge:str, w:str) -> dict:
    if match:
        json_str = match.group(1)
        data = json.loads(json_str)
-        return data
+        print(f"极验验证响应: {data}")
+
+        # 检查验证是否成功
+        if data.get('success') == 1:
+            return data
+        else:
+            # 如果验证失败，返回错误信息
+            return {
+                'success': 0,
+                'message': data.get('message', '验证失败'),
+                'validate': ''
+            }
    else:
        raise ValueError("无法解析 JSONP 响应")
@@ -41,13 +41,29 @@ def login_single(config: Config, account_index: int = 0) -> str:
    # 创建登录器
    proxy = config.get_proxy()
    proxy_url = None
-    if proxy.enabled:
-        proxy_url = {
-            'http': proxy.http or proxy.https,
-            'https': proxy.https or proxy.http,
-        }
+    proxy_api_url = None

-    loginer = DouyuLogin(account, proxy=proxy_url)
+    if proxy.enabled:
+        if proxy.api_url:
+            # 使用代理API
+            proxy_api_url = proxy.api_url
+        elif proxy.http or proxy.https:
+            # 使用静态代理
+            proxy_url = {
+                'http': proxy.http or proxy.https,
+                'https': proxy.https or proxy.http,
+            }
+
+    # 获取极验配置
+    geetest_config = config.get_geetest_config()
+    max_retries = geetest_config.get('max_retries', 5)
+
+    loginer = DouyuLogin(
+        account,
+        proxy=proxy_url,
+        proxy_api_url=proxy_api_url,
+        max_geetest_retries=max_retries,
+    )

    # 执行登录
    result = loginer.login()