yml/live-hub-py
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化代理配置

Browse Source
This commit is contained in:
yml2213
2026-06-22 13:59:03 +08:00
parent 63c1ffa046
commit b08f0a647d
10 changed files with 577 additions and 197 deletions
Download Patch File Download Diff File Expand all files Collapse all files
core/douyu/proxy.py
+139 -43
View File
@@ -1,16 +1,17 @@
"""代理管理模块"""
import re
import time
import requests
from typing import Optional, List
from typing import Optional
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"
def __init__(self, api_url: str = ""):
self.api_url = api_url
self.current_proxy: Optional[str] = None
def get_proxy(self) -> Optional[str]:
@@ -37,46 +38,26 @@ class ProxyManager:
self.current_proxy = proxy
logger.info(f"获取到代理: {proxy}")
return proxy
else:
logger.warning(f"无法解析代理地址: {text}")
return None
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字典
"""
"""获取requests使用的proxies字典"""
proxy = proxy or self.current_proxy
if proxy:
return {
'http': proxy,
'https': 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',
@@ -93,23 +74,138 @@ class ProxyManager:
return False
# 全局代理管理器实例
_proxy_manager: Optional[ProxyManager] = None
def get_proxy_manager(api_url: str = "") -> ProxyManager:
"""获取代理管理器实例（每次传入 api_url 时创建新实例，避免全局状态污染）"""
return ProxyManager(api_url)
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 parse_proxy_response(text: str) -> tuple[Optional[str], Optional[str]]:
"""
解析代理API响应。
Returns:
(proxy_url, whitelist_ip)
- proxy_url: 解析到的代理地址（http://ip:port），无法解析时为 None
- whitelist_ip: 需要添加到白名单的IP（当API返回白名单错误时），无错误时为 None
"""
text = text.strip()
# 正常代理地址
match = re.search(r'(\d+\.\d+\.\d+\.\d+):(\d+)', text)
if match:
ip, port = match.group(1), match.group(2)
# 排除 "请先添加白名单：1.2.3.4" 中误匹配到 ip:port 的情况
if '白名单' not in text:
return f"http://{ip}:{port}", None
# 白名单错误：DB1.请先添加白名单：39.144.114.76
if '添加白名单' in text or '白名单' in text:
ip_match = re.search(r'(\d+\.\d+\.\d+\.\d+)', text)
if ip_match:
return None, ip_match.group(1)
return None, None
def get_proxy() -> Optional[str]:
"""获取代理URL的便捷函数"""
return get_proxy_manager().get_proxy()
def verify_proxy_url(proxy_url: str, timeout: tuple = (4, 6)) -> tuple[bool, str]:
"""
验证代理是否可用。
Returns:
(是否可用, 消息)
"""
proxies = {'http': proxy_url, 'https': proxy_url}
targets = [
'https://qifu-api.baidubce.com/ip/local/geo/v1/district',
'https://myip.ipip.net',
'https://4.ipw.cn',
]
for url in targets:
try:
response = requests.get(
url, proxies=proxies, timeout=timeout,
headers={'User-Agent': 'Mozilla/5.0'},
)
response.raise_for_status()
return True, f'代理可用: {url.split("/")[2]}'
except Exception as e:
err_msg = str(e)
if 'Tunnel connection failed' in err_msg or '503' in err_msg:
detail = '代理拒绝连接（白名单可能未生效）'
elif 'timed out' in err_msg.lower():
detail = '连接超时'
else:
detail = type(e).__name__
logger.debug(f"代理验证 {url} 失败: {detail}")
continue
return False, '代理验证失败（所有目标不可达）'
def get_proxies_dict() -> dict:
"""获取proxies字典的便捷函数"""
return get_proxy_manager().get_proxies_dict()
def resolve_working_proxy(
api_url: str,
whitelist_uid: str = "",
whitelist_ukey: str = "",
max_attempts: int = 5,
log_func=None,
) -> tuple[Optional[str], str]:
"""
从代理API获取可用代理，自动处理白名单同步。
Args:
api_url: 代理API地址
whitelist_uid: 白名单UID（启用白名单时传入）
whitelist_ukey: 白名单UKEY
max_attempts: 最大尝试次数
log_func: 日志回调函数 (level, message)
Returns:
(代理URL, 消息)
"""
def log(level, msg):
if log_func:
log_func(level, msg)
else:
getattr(logger, level if level in ('info', 'warning', 'error', 'success') else 'info', logger.info)(msg)
synced_whitelist = False
for attempt in range(1, max_attempts + 1):
log('info', f'代理预检 {attempt}/{max_attempts}: 正在获取代理')
try:
response = requests.get(api_url, timeout=10)
response.raise_for_status()
text = response.text.strip()
proxy_url, whitelist_ip = parse_proxy_response(text)
if proxy_url:
# 验证代理
ok, msg = verify_proxy_url(proxy_url)
if ok:
log('success', f'代理预检成功: {proxy_url}')
return proxy_url, msg
log('warning', f'代理预检 {attempt}/{max_attempts}: {msg}')
continue
# 代理API返回白名单错误
if whitelist_ip and not synced_whitelist and whitelist_uid and whitelist_ukey:
log('warning', f'代理需要白名单IP: {whitelist_ip}，自动同步...')
from core.douyu.whitelist import WhitelistManager
manager = WhitelistManager(whitelist_uid, whitelist_ukey)
ok, sync_msg = manager.sync_ip(whitelist_ip)
log('success' if ok else 'error', f'白名单同步: {sync_msg}')
if ok:
synced_whitelist = True
log('info', '白名单已更新，等待2秒后重试...')
time.sleep(2)
continue
return None, f'白名单同步失败: {sync_msg}'
log('warning', f'代理预检 {attempt}/{max_attempts}: 代理API响应无法解析: {text[:80]}')
except Exception as exc:
log('warning', f'代理预检 {attempt}/{max_attempts}: 代理API请求失败: {exc}')
return None, f'代理预检失败，已尝试 {max_attempts} 次'
core/douyu/whitelist.py
+34 -12
View File
@@ -2,6 +2,7 @@
import json
import re
import time
from typing import Optional
from urllib.parse import urlencode
@@ -65,12 +66,12 @@ class WhitelistManager:
logger.error(f"获取白名单失败: {e}")
return []
def add_ip(self, ip: str) -> bool:
def add_ip(self, ip: str, retry: bool = True) -> tuple[bool, str]:
"""
添加IP到白名单
Args:
ip: 要添加的IP地址
Returns:
(是否成功, API原始响应)
"""
try:
url = self._build_url(act="add", ip=ip, meno=self._memo)
@@ -80,22 +81,40 @@ class WhitelistManager:
text = response.text.strip()
logger.debug(f"添加白名单响应: {text}")
# 成功通常返回 "ok" 或类似信息
if "ok" in text.lower() or "success" in text.lower() or "添加成功" in text:
logger.info(f"白名单添加成功: {ip} (备注: {self._memo})")
return True
return True, text
# 检查是否已存在
if "已存在" in text or "exist" in text.lower():
if "已存在" in text or "exist" in text.lower() or "IpRep" in text:
logger.info(f"白名单已存在: {ip}")
return True
return True, text
# 频率限制，等待后重试一次
if retry and ("频率过快" in text or "稍后" in text):
wait = 5
match = re.search(r'(\d+)\s*秒', text)
if match:
wait = int(match.group(1))
logger.info(f"白名单添加被限流，等待 {wait} 秒后重试...")
time.sleep(wait)
return self.add_ip(ip, retry=False)
# UKEY 错误
if "Err:Key" in text:
logger.error("白名单UKEY错误，请检查配置")
return False, "UKEY错误，请检查白名单配置"
# 超出白名单数量限制
if "Err:Max" in text or "超过" in text or "上限" in text:
logger.error(f"白名单数量超限: {text}")
return False, f"白名单数量超限: {text}"
logger.warning(f"白名单添加结果: {text}")
return False
return False, text
except Exception as e:
logger.error(f"添加白名单失败: {e}")
return False
return False, str(e)
def delete_ip(self, ip: str) -> bool:
"""
@@ -175,15 +194,18 @@ class WhitelistManager:
if existing_ip:
logger.info(f"白名单IP变化: {existing_ip} -> {current_ip}")
self.delete_ip(existing_ip)
time.sleep(1)
# IP已存在但备注不同（或备注为空），先删除后重新添加
records = self.get_whitelist_json()
if any(r.get('IP') == current_ip for r in records):
logger.info(f"白名单IP {current_ip} 已存在但备注不同，删除后重新添加")
self.delete_ip(current_ip)
time.sleep(1)
# 添加新IP
if self.add_ip(current_ip):
ok, resp = self.add_ip(current_ip)
if ok:
if existing_ip:
msg = f"白名单IP已更新: {existing_ip} -> {current_ip}"
else:
@@ -191,7 +213,7 @@ class WhitelistManager:
logger.info(msg)
return True, msg
return False, "白名单添加失败"
return False, f"白名单添加失败，API响应: {resp}"
except Exception as e:
msg = f"白名单同步失败: {e}"