78 lines
2.8 KiB
Python
78 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
from __future__ import annotations
|
||
|
||
import re
|
||
import time
|
||
from typing import Optional
|
||
|
||
import requests
|
||
|
||
from app_logger import get_logger
|
||
|
||
logger = get_logger("proxy")
|
||
|
||
|
||
def validate_ip_port(ip_port_str: str) -> bool:
|
||
"""验证IP:PORT格式是否有效（含范围校验）。"""
|
||
pattern = re.compile(r"^(\d{1,3})\.(\d{1,3})\.(\d{1,3})\.(\d{1,3}):(\d{1,5})$")
|
||
match = pattern.match(ip_port_str or "")
|
||
if not match:
|
||
return False
|
||
|
||
for i in range(1, 5):
|
||
octet = int(match.group(i))
|
||
if octet < 0 or octet > 255:
|
||
return False
|
||
|
||
port = int(match.group(5))
|
||
return 1 <= port <= 65535
|
||
|
||
|
||
def get_proxy_from_api(api_url: str, max_retries: int = 3) -> Optional[str]:
|
||
"""从API获取代理IP（支持重试）。"""
|
||
ip_port_pattern = re.compile(r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}:\d{1,5}$")
|
||
max_retries = max(1, int(max_retries or 1))
|
||
|
||
for attempt in range(max_retries):
|
||
try:
|
||
response = requests.get(api_url, timeout=10)
|
||
if response.status_code == 200:
|
||
text = response.text.strip()
|
||
|
||
# 尝试解析JSON响应
|
||
try:
|
||
import json
|
||
|
||
data = json.loads(text)
|
||
if isinstance(data, dict):
|
||
if data.get("status") not in (200, 0, None):
|
||
error_msg = data.get("msg", data.get("message", "未知错误"))
|
||
logger.warning(f"代理API返回错误: {error_msg} (尝试 {attempt + 1}/{max_retries})")
|
||
if attempt < max_retries - 1:
|
||
time.sleep(1)
|
||
continue
|
||
ip_port = data.get("ip") or data.get("proxy") or data.get("data")
|
||
if ip_port:
|
||
text = str(ip_port).strip()
|
||
except Exception:
|
||
pass
|
||
|
||
if ip_port_pattern.match(text) and validate_ip_port(text):
|
||
proxy_server = f"http://{text}"
|
||
logger.info(f"获取代理成功: {proxy_server} (尝试 {attempt + 1}/{max_retries})")
|
||
return proxy_server
|
||
|
||
logger.warning(f"代理格式或范围无效: {text[:50]} (尝试 {attempt + 1}/{max_retries})")
|
||
else:
|
||
logger.warning(f"获取代理失败: HTTP {response.status_code} (尝试 {attempt + 1}/{max_retries})")
|
||
except Exception as e:
|
||
logger.warning(f"获取代理异常: {str(e)} (尝试 {attempt + 1}/{max_retries})")
|
||
|
||
if attempt < max_retries - 1:
|
||
time.sleep(1)
|
||
|
||
logger.warning(f"获取代理失败，已重试 {max_retries} 次，将不使用代理继续")
|
||
return None
|
||
|