(self)
| 334 | return None |
| 335 | |
| 336 | def configure_proxy(self): |
| 337 | proxy_api = os.environ.get("UNICOM_PROXY_API") |
| 338 | if not proxy_api: |
| 339 | return |
| 340 | proxy_type = os.environ.get("UNICOM_PROXY_TYPE", "socks5").lower() |
| 341 | max_retries = 5 |
| 342 | for attempt in range(1, max_retries + 1): |
| 343 | try: |
| 344 | if attempt > 1: |
| 345 | self.log(f"🔄 [第{attempt}次] 重试获取代理IP ({proxy_type})...") |
| 346 | time.sleep(2) |
| 347 | else: |
| 348 | self.log(f"正在获取代理IP (模式: {proxy_type})...") |
| 349 | res = requests.get(proxy_api, timeout=10) |
| 350 | if res.status_code != 200: |
| 351 | self.log(f"⚠️ 获取代理失败: HTTP {res.status_code}") |
| 352 | continue |
| 353 | proxy_info = self._parse_proxy_response(res.text) |
| 354 | if not proxy_info: |
| 355 | preview = res.text[:100] + "..." if len(res.text) > 100 else res.text |
| 356 | self.log(f"❌ 提取失败: 无法识别代理格式 (内容: {preview})") |
| 357 | continue |
| 358 | ip, port = proxy_info['ip'], proxy_info['port'] |
| 359 | user, pwd = proxy_info['user'], proxy_info['pass'] |
| 360 | if user and pwd: |
| 361 | proxy_url = f"{proxy_type}://{quote(user)}:{quote(pwd)}@{ip}:{port}" |
| 362 | log_msg = f"{proxy_type}://***:***@{ip}:{port}" |
| 363 | else: |
| 364 | proxy_url = f"{proxy_type}://{ip}:{port}" |
| 365 | log_msg = proxy_url |
| 366 | self.log(f"🔍 提取成功: {log_msg}") |
| 367 | test_proxies = {"http": proxy_url, "https": proxy_url} |
| 368 | try: |
| 369 | requests.get("https://www.baidu.com", proxies=test_proxies, timeout=3) |
| 370 | self.session.proxies.update(test_proxies) |
| 371 | self.log("✅ 代理连通性测试通过") |
| 372 | return |
| 373 | except Exception as te: |
| 374 | self.log(f"⚠️ 代理测试失败: {te}") |
| 375 | except Exception as e: |
| 376 | self.log(f"❌ 请求代理API异常: {e}") |
| 377 | self.log(f"🚫 重试{max_retries}次均失败，回退至本地IP") |
| 378 | |
| 379 | def failover_proxy(self): |
| 380 | proxy_api = os.environ.get("UNICOM_PROXY_API") |
no test coverage detected