如何解决undetected_chromedriver设置代理IP无法使用用户名密码的问题
在Python爬虫开发中，代理IP是绕过反爬机制的关键技术之一。但当我们使用undetected_chromedriver进行自动化操作时，会遇到一个非常棘手的问题：undetected_chromedriver本身不支持使用带用户名密码授权的代理IP，本文另辟蹊径，通过"本地端口转发 + 认证信息中转"的方案完美解决这一难题。
我们的解决方案基于一个核心思想：通过本地端口转发服务来解决代理认证问题。
技术架构
浏览器驱动层 (undetected_chromedriver)
↓ 使用无认证的本地代理
本地代理转发层 (SmartProxy服务)
↓ 自动添加认证信息
远程代理服务器 (需要用户名密码)
↓
目标网站代码实现
1. 本地代理转发服务 (smart_proxy.py)
这是解决方案的核心，实现了本地代理转发服务：
import socketserver
import threading
import socket
import base64
import json
import time
import requests
from typing import Optional
from http.server import HTTPServer, BaseHTTPRequestHandler
import logging
import hashlib
import pickle
import os
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
class ProxyConfig:
"""代理配置类"""
def __init__(self, remote_host: str, remote_port: int, username: str = None, password: str = None):
self.remote_host = remote_host
self.remote_port = remote_port
self.username = username
self.password = password
self.auth_header = None
self.local_port = None
if username and password:
auth_str = f"{username}:{password}"
auth_bytes = auth_str.encode('utf-8')
self.auth_header = f"Basic {base64.b64encode(auth_bytes).decode('utf-8')}"
def get_id(self) -> str:
"""获取配置的唯一ID"""
config_str = f"{self.remote_host}:{self.remote_port}:{self.username}:{self.password}"
return hashlib.md5(config_str.encode()).hexdigest()
def __str__(self):
return f"ProxyConfig(host={self.remote_host}, port={self.remote_port}, user={self.username})"
class SmartProxyHandler(BaseHTTPRequestHandler):
"""智能代理处理器"""
def __init__(self, *args, proxy_config: ProxyConfig = None, **kwargs):
self.proxy_config = proxy_config
super().__init__(*args, **kwargs)
def do_CONNECT(self):
"""处理HTTPS CONNECT请求"""
try:
# 连接到目标服务器（远程代理）
remote_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
remote_socket.connect((self.proxy_config.remote_host, self.proxy_config.remote_port))
# 发送CONNECT请求到远程代理
connect_request = f"CONNECT {self.path} HTTP/1.1\r\n"
connect_request += f"Host: {self.path}\r\n"
if self.proxy_config.auth_header:
connect_request += f"Proxy-Authorization: {self.proxy_config.auth_header}\r\n"
connect_request += "\r\n"
remote_socket.sendall(connect_request.encode())
# 读取响应
response = remote_socket.recv(4096)
# 转发响应给客户端
self.wfile.write(response)
# 双向转发数据
self._forward_traffic(self.connection, remote_socket)
except Exception as e:
logger.error(f"HTTPS连接错误: {e}")
self.send_error(502, f"Proxy Error: {e}")
finally:
try:
remote_socket.close()
except:
pass
def do_GET(self):
self._handle_http_request()
def do_POST(self):
self._handle_http_request()
def do_PUT(self):
self._handle_http_request()
def _handle_http_request(self):
"""处理HTTP请求 - 修复版：正确处理代理认证"""
try:
# 1. 获取原始请求信息
url = self.path
# 处理原始请求可能是相对路径的情况
if not url.startswith('http'):
# 从请求头中获取原始请求的完整URL（Host头+路径）
host = self.headers.get('Host', '')
if host:
url = f'http://{host}{url}'
else:
# 如果连Host头都没有，则使用默认的httpbin进行测试
url = 'http://httpbin.org/ip'
# 2. 构建包含认证信息的远程代理URL
if self.proxy_config.username and self.proxy_config.password:
# 格式：http://用户名:密码@代理地址:代理端口
remote_proxy_url = f"http://{self.proxy_config.username}:{self.proxy_config.password}@{self.proxy_config.remote_host}:{self.proxy_config.remote_port}"
else:
remote_proxy_url = f"http://{self.proxy_config.remote_host}:{self.proxy_config.remote_port}"
# 3. 准备requests库使用的proxies字典
proxies = {
'http': remote_proxy_url,
'https': remote_proxy_url,
}
# 4. 正确收集并转发原始请求头
# 注意：我们不应该移除Proxy-Authorization，因为requests库会在底层处理
# 但需要移除一些可能影响代理服务器的头部
headers = {}
for key, value in self.headers.items():
key_lower = key.lower()
# 过滤掉一些连接相关的头部，但保留其他头部
if key_lower not in ['proxy-connection', 'connection']:
headers[key] = value
# 5. 获取请求体数据
content_length = int(self.headers.get('Content-Length', 0))
data = self.rfile.read(content_length) if content_length > 0 else None
# 6. 使用requests库转发请求
# 重要：不需要传递auth参数，认证信息已包含在proxies字典的URL中[citation:10]
response = requests.request(
method=self.command,
url=url,
data=data,
headers=headers,
proxies=proxies,
verify=False, # 跳过SSL验证（仅用于测试）
timeout=30,
allow_redirects=False
)
# 7. 将远程代理的响应返回给客户端
self.send_response(response.status_code)
# 转发响应头
for key, value in response.headers.items():
self.send_header(key, value)
self.end_headers()
# 发送响应体
self.wfile.write(response.content)
except requests.exceptions.ProxyError as e:
logger.error(f"代理服务器连接或认证错误: {e}")
self.send_error(407, "Proxy Authentication Required")
except requests.exceptions.ConnectTimeout as e:
logger.error(f"连接远程代理超时: {e}")
self.send_error(504, "Proxy Timeout")
except Exception as e:
logger.error(f"HTTP请求处理异常: {e}")
self.send_error(500, "Internal Proxy Error")
def _forward_traffic(self, client_socket, remote_socket):
"""双向转发数据"""
import select
sockets = [client_socket, remote_socket]
while True:
try:
readable, _, exceptional = select.select(sockets, [], sockets, 60)
if exceptional:
break
for sock in readable:
if sock is client_socket:
data = client_socket.recv(4096)
if not data:
return
remote_socket.sendall(data)
else:
data = remote_socket.recv(4096)
if not data:
return
client_socket.sendall(data)
except:
break
def log_message(self, format, *args):
"""重写日志方法"""
logger.info(f"Proxy {self.proxy_config.local_port}: {format % args}")
class ThreadingHTTPServer(socketserver.ThreadingMixIn, HTTPServer):
"""支持多线程的HTTP服务器，可被强制停止"""
daemon_threads = True
running = True # 新增：运行状态标志
def __init__(self, server_address, RequestHandlerClass, proxy_config: ProxyConfig):
self.proxy_config = proxy_config
super().__init__(server_address, RequestHandlerClass)
def serve_forever(self, poll_interval=0.5):
"""重写serve_forever，使其能检查running标志"""
while self.running:
try:
self.handle_request() # 只处理一个请求，然后循环检查标志
except (KeyboardInterrupt, SystemExit):
raise
except OSError:
# Socket被关闭时产生的预期错误，直接退出循环
break
except Exception:
# 处理单个请求时的其他错误，忽略并继续
pass
def force_stop(self):
"""强制停止服务器"""
self.running = False # 1. 设置标志，让serve_forever循环退出
time.sleep(0.1) # 2. 给循环一个反应时间
try:
self.server_close() # 3. 关闭服务器Socket，释放端口
self.socket.close() # 4. 确保底层Socket关闭
except:
pass
def finish_request(self, request, client_address):
"""为每个请求处理器传递proxy_config"""
self.RequestHandlerClass(request, client_address, self, proxy_config=self.proxy_config)
class SmartProxyManager:
"""智能代理管理器 - 自动管理端口和配置"""
_instance = None
_lock = threading.Lock()
def __new__(cls):
with cls._lock:
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self):
if self._initialized:
return
self.servers = {} # port -> (server, config)
self.config_map = {} # config_id -> port
self.port_map = {} # port -> config_id
self.min_port = 10000
self.max_port = 20000
self.current_port = self.min_port
self.persistence_file = "proxy_mappings.pkl"
# 加载持久化的映射
self._load_mappings()
self._initialized = True
logger.info("智能代理管理器初始化完成")
def _load_mappings(self):
"""加载持久化的映射关系"""
try:
if os.path.exists(self.persistence_file):
with open(self.persistence_file, 'rb') as f:
data = pickle.load(f)
self.config_map = data.get('config_map', {})
self.port_map = data.get('port_map', {})
logger.info(f"加载了 {len(self.config_map)} 个代理映射")
except Exception as e:
logger.warning(f"加载映射文件失败: {e}")
def _save_mappings(self):
"""保存映射关系"""
try:
with open(self.persistence_file, 'wb') as f:
data = {
'config_map': self.config_map,
'port_map': self.port_map
}
pickle.dump(data, f)
except Exception as e:
logger.warning(f"保存映射文件失败: {e}")
def _find_available_port(self) -> Optional[int]:
"""查找可用端口"""
start_port = self.current_port
for port in range(start_port, self.max_port):
if self._is_port_available(port):
self.current_port = port + 1
return port
# 如果没找到，从头开始找
for port in range(self.min_port, start_port):
if self._is_port_available(port):
self.current_port = port + 1
return port
return None
def _is_port_available(self, port: int) -> bool:
"""检查端口是否可用"""
# 检查是否已经被我们使用
if port in self.servers:
return False
# 检查系统端口是否被占用
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
sock.bind(('127.0.0.1', port))
sock.close()
return True
except OSError:
return False
finally:
try:
sock.close()
except:
pass
def get_proxy_for_config(self, proxy_config: ProxyConfig) -> Optional[int]:
"""
为代理配置获取本地代理端口
如果已经存在对应配置的代理，返回已有端口
否则启动新的代理服务
"""
config_id = proxy_config.get_id()
with self._lock:
# 检查是否已有该配置的代理
if config_id in self.config_map:
port = self.config_map[config_id]
if port in self.servers:
logger.info(f"使用已有的代理服务: {proxy_config} -> 127.0.0.1:{port}")
return port
else:
# 端口存在但服务已停止，清理映射
del self.config_map[config_id]
if port in self.port_map:
del self.port_map[port]
# 启动新的代理服务
port = self._start_proxy_for_config(proxy_config)
if port:
proxy_config.local_port = port
self.config_map[config_id] = port
self.port_map[port] = config_id
self._save_mappings()
return port
def _start_proxy_for_config(self, proxy_config: ProxyConfig) -> Optional[int]:
"""为配置启动代理服务"""
# 查找可用端口
port = self._find_available_port()
if not port:
logger.error("没有可用的端口")
return None
try:
# 创建服务器
server = ThreadingHTTPServer(
('127.0.0.1', port),
SmartProxyHandler,
proxy_config=proxy_config
)
# 启动服务器线程
server_thread = threading.Thread(
target=server.serve_forever,
daemon=True,
name=f"ProxyServer-{port}"
)
server_thread.start()
self.servers[port] = (server, proxy_config)
logger.info(f"启动代理服务成功: {proxy_config} -> 127.0.0.1:{port}")
# 等待服务器完全启动
time.sleep(0.5)
return port
except Exception as e:
logger.error(f"启动代理服务失败 (端口: {port}): {e}")
return None
def stop_proxy(self, port: int):
"""强制停止指定端口的代理服务"""
with self._lock:
if port in self.servers:
server, config = self.servers[port]
logger.info(f"正在强制停止代理服务: 127.0.0.1:{port}")
# 关键：调用我们自定义的强制停止方法
server.force_stop()
# 从管理记录中移除
del self.servers[port]
# 清理映射
if port in self.port_map:
config_id = self.port_map[port]
if config_id in self.config_map:
del self.config_map[config_id]
del self.port_map[port]
logger.info(f"代理服务已强制停止: 127.0.0.1:{port}")
self._save_mappings()
return True
return False
def stop_all(self):
"""强制停止所有代理服务"""
logger.info("正在强制停止所有代理服务...")
ports_to_stop = list(self.servers.keys()) # 先获取键的快照，避免迭代时修改字典
for port in ports_to_stop:
self.stop_proxy(port)
logger.info("所有代理服务已强制停止")
def list_proxies(self):
"""列出所有代理服务"""
logger.info("当前代理服务列表:")
with self._lock:
if not self.servers:
logger.info(" 没有运行的代理服务")
else:
for port, (_, config) in self.servers.items():
logger.info(f" 127.0.0.1:{port} -> {config}")
def get_proxy_url(self, proxy_config: ProxyConfig) -> Optional[str]:
"""获取代理配置的本地代理URL"""
port = self.get_proxy_for_config(proxy_config)
if port:
return f"http://127.0.0.1:{port}"
return None
# 全局代理管理器实例
proxy_manager = SmartProxyManager()
def get_proxy_for_config(remote_host: str, remote_port: int, username: str = None, password: str = None) -> Optional[str]:
"""
获取代理配置的本地代理URL
简化接口，用户只需提供代理配置信息
"""
config = ProxyConfig(remote_host, remote_port, username, password)
return proxy_manager.get_proxy_url(config)
if __name__ == "__main__":
# 测试代码
print("智能代理管理器测试")
print("=" * 50)
# 测试配置1
proxy_url1 = get_proxy_for_config(
remote_host="abc.zdaye.com",
remote_port=16888,
username="20250118216",
password="hayoua"
)
if proxy_url1:
print(f"? 代理配置1获取成功: {proxy_url1}")
else:
print("? 代理配置1获取失败")
# 测试配置2
proxy_url2 = get_proxy_for_config(
remote_host="proxy2.example.com",
remote_port=8080,
username="user2",
password="pass2"
)
if proxy_url2:
print(f"? 代理配置2获取成功: {proxy_url2}")
else:
print("? 代理配置2获取失败")
# 显示所有代理
proxy_manager.list_proxies()
print("\n代理服务运行中，按 Ctrl+C 停止...")
try:
while True:
time.sleep(1)
except KeyboardInterrupt:
proxy_manager.stop_all()
print("代理服务已停止")代码架构
用户应用
↓ (请求本地代理)
SmartProxyManager (全局单例)
↓ (管理代理实例)
ThreadingHTTPServer (多线程服务器)
↓ (处理请求)
SmartProxyHandler (请求处理器)
↓ (转发请求)
远程代理服务器代码说明
这段代码是将远程HTTP/HTTPS代理转换为本地代理服务，提供统一的本地接口。
ProxyConfig 是代理配置类，用来封装远程代理的配置信息，将代理服务器的用户名密码生成基本认证头（Basic Auth），通过MD5生成唯一的配置ID。
SmartProxyHandler 是智能代理处理器，用来处理HTTP请求（GET/POST/PUT等）和HTTPS请求（CONNECT）。
ThreadingHTTPServer 是多线程服务器，组合ThreadingMixIn + HTTPServer，为每个请求传递配置信息，支持线程安全处理并发请求，可支持强制停止功能。
SmartProxyManager 是核心管理器，采用单例模式，确保全局唯一，用来自动分配和管理端口，持久化配置-端口映射，管理代理服务的生命周期。
2. 代理配置接口 (proxy_config.py)
提供简单易用的API接口，可配置多组代理服务器的账号密码：
"""
简化的代理配置接口
"""
from smart_proxy import ProxyConfig, proxy_manager
class SimpleProxy:
"""简化代理接口"""
@staticmethod
def create_proxy(host: str, port: int, username: str = None, password: str = None) -> str:
"""
创建代理并返回本地代理URL
Args:
host: 代理服务器主机
port: 代理服务器端口
username: 用户名（可选）
password: 密码（可选）
Returns:
本地代理URL，如 "http://127.0.0.1:10005"
"""
config = ProxyConfig(host, port, username, password)
proxy_url = proxy_manager.get_proxy_url(config)
if proxy_url:
print(f"? 代理创建成功: {proxy_url}")
return proxy_url
else:
print("? 代理创建失败")
return None
@staticmethod
def stop_all():
"""停止所有代理"""
proxy_manager.stop_all()
print("所有代理已停止")
@staticmethod
def list_proxies():
"""列出所有代理"""
proxy_manager.list_proxies()
# 预定义的代理配置
PROXY_CONFIGS = {
"proxy1": {
"name": "主要代理",
"host": "abc.zdaye.com",
"port":16888,
"username": "202216",
"password": "hyi5a"
},
"proxy2": {
"name": "备用代理1",
"host": "proxy2.example.com",
"port": 8080,
"username": "user2",
"password": "pass2"
},
"proxy3": {
"name": "备用代理2",
"host": "proxy3.example.com",
"port": 8080,
"username": "user3",
"password": "pass3"
}
}
def get_proxy_by_name(name: str) -> str:
"""通过名称获取代理URL"""
if name in PROXY_CONFIGS:
config = PROXY_CONFIGS[name]
return SimpleProxy.create_proxy(
config["host"],
config["port"],
config.get("username"),
config.get("password")
)
else:
print(f"? 代理配置 '{name}' 不存在")
return None
if __name__ == "__main__":
# 使用示例
print("代理配置工具")
print("=" * 50)
# 获取代理1
proxy_url = get_proxy_by_name("proxy1")
if proxy_url:
print(f"\n使用代理URL: {proxy_url}")
# 在requests中使用
import requests
proxies = {
'http': proxy_url,
'https': proxy_url
}
try:
response = requests.get('http://httpbin.org/ip', proxies=proxies)
print(f"代理IP: {response.json()}")
except Exception as e:
print(f"请求失败: {e}")
# 显示所有代理
print("\n当前代理状态:")
SimpleProxy.list_proxies()
# 清理
input("\n按Enter键停止所有代理...")
SimpleProxy.stop_all()代码架构：
├── 外部依赖
│ └── smart_proxy (第三方库)
│ ├── ProxyConfig 类
│ └── proxy_manager 对象
│
├── 核心接口类
│ └── SimpleProxy 类
│ ├── create_proxy() 静态方法
│ ├── stop_all() 静态方法
│ └── list_proxies() 静态方法
│
├── 配置管理
│ └── PROXY_CONFIGS 字典
│ ├── proxy1: 主要代理配置
│ ├── proxy2: 备用代理1
│ └── proxy3: 备用代理2
│
└── 辅助函数
└── get_proxy_by_name() 函数代码说明：
SimpleProxy类用来简化接口，通过create_proxy()实现一键创建代理，stop_all()实现一键停止所有代理，list_proxies()实现查看所有代理状态。
PROXY_CONFIGS字典用来预配置管理，用来集中管理多个代理配置，具有扩展性，可轻松添加新配置。
get_proxy_by_name()函数可通过配置名来获取代理URL。
3. 集成测试 (test_proxy_directly.py)
这里以requests请求为例，可自行替换"undetected_chromedriver"进行测试，亲测有效。
import requests
from smart_proxy import ProxyConfig, proxy_manager
# 1. 获取本地代理URL
config = ProxyConfig(
remote_host="abc.zdaye.com",
remote_port=16888,
username="202216",
password="hyi5a"
)
proxy_url = proxy_manager.get_proxy_url(config)
print(f"本地代理URL: {proxy_url}")
if proxy_url:
# 2. 测试通过本地代理访问
proxies = {
'http': proxy_url,
'https': proxy_url,
}
print("正在测试代理连接...")
try:
# 测试一个简单网站
response = requests.get('http://httpbin.org/ip', proxies=proxies, timeout=10)
print(f"? 成功！响应状态码: {response.status_code}")
print(f"返回内容: {response.text[:200]}...")
except requests.exceptions.ProxyError as e:
print(f"? 代理错误: {e}")
print("这可能表示：1) 远程代理服务器问题；2) 认证失败；3) 网络不通")
except Exception as e:
print(f"? 其他错误: {e}")
# 3. 显示代理状态
print("\n当前代理服务状态:")
proxy_manager.list_proxies()
input("\n按Enter键停止代理服务...")
proxy_manager.stop_all()总结
本文介绍了一种创新的解决方案，通过本地端口转发服务，完美解决了"undetected_chromedriver"无法直接使用需要认证的代理IP的问题。该方案具有以下优点：
1. 无缝集成：无需修改"undetected_chromedriver"的调用方式
2. 稳定可靠：基于HTTP标准协议，兼容性好
3. 易于扩展：支持多代理配置和自动切换
这种方案在实际爬虫项目中已经验证有效，能够显著提高自动化脚本的稳定性和成功率。希望本文的分享能够帮助到遇到类似问题的朋友。
