🔥码云GVP开源项目 12k star Uniapp+ElementUI 功能强大 支持多语言、二开方便! 广告
**1. 在`settings.py`做如下配置** ```python DOWNLOADER_MIDDLEWARES = { "proxies_pro.middlewares.ProxiesProDownloaderMiddleware": 543, # 启用自己定义的中间件 "proxies_pro.middlewares.RandomIpPoolMiddleware": 125, } IPPOOL = [ # 免费代理,无密码和用户名 {"ipaddr": "101.236.21.22:8866"}, {"ipaddr": "101.126.18.101:8866"}, {"ipaddr": "118.31.220.3:8080"}, {"ipaddr": "35.214.170.66:3128"}, # 付费代理,有密码和用户名 {"ipaddr": "56.214.180.69:3990", "user_passwd":"username:password"}, ] ``` <br/> **2. 在`middlewares.py`中自定义 RandomIpPoolMiddleware 中间件** ```python import random import logging import base64 from proxies_pro.settings import IPPOOL myLogger = logging.getLogger(__name__) class RandomIpPoolMiddleware: """ 如果某一个ip无效或者超时,process_request会重复其它ip 在我的测试中重复了4次,如果重复了4次依然没有一个ip有效 则抛出异常: twisted.internet.error.TCPTimedOutError: TCP connection timed out: 10060: 由于连接方在一段时间后没有正确答复或连接的主机没有反应,连接尝试失败。. """ def process_request(self, request, spider): proxy = random.choice(IPPOOL) # 付费的ip代理需要进行认证 if 'user_passwd' in proxy: # 对账号密码进行编码 b64_up = base64.b64encode(proxy['user_passwd']) # 设置认证 request.headers['Proxy-Authorization'] = 'Basic '+ b64_up.decode() # 设置代理 request.meta['proxy'] = proxy["ipaddr"] else: # 设置代理 request.meta['proxy'] = proxy['ipaddr'] return None def process_response(self, request, response, spider): myLogger.warning("proxy:{}".format(request.meta["proxy"])) return response ```