有效IP

# 获取代理IP def getProxyIPs(): url = 'https://proxy/get_ip.php' response = requests.get(url, params=params) data = response.json() # print(data) listIPs = [] for val in data['data']['proxies']: url = "https://" + str(val) try: response = requests.get("https://web.com/api", verify=False, timeout=15, allow_redirects=True, proxies={"https": url}) listIPs.append(url) except: continue return listIPs

IP池

def genIPPool(maxCount): unique_list = [] while True: # 重新拉取新的 try: my_list = getProxyIPs() [unique_list .append(item) for item in my_list if item not in unique_list ] except: continue if len(ipHttpsProxys) > maxCount: return unique_list

均衡更新

ipHttpsProxys = [] def reqUrl(url): failRemove = False #请求失败是否从代理池移除，触发ip池更新 if 0 == len(ipHttpsProxys): ipHttpsProxys = genIPPool(10) proxyUrl = ipHttpsProxys[0] proxies = { "https": proxyUrl } del ipHttpsProxys[0] if not failRemove: ipHttpsProxys.append(proxyUrl) ssl._create_default_https_context = ssl._create_unverified_context json_response = requests.get(url, proxies=proxies).json() if failRemove: ipHttpsProxys.append(proxyUrl)

拥塞调控

def getDatas(url, retryMaxCount): sleepTime = 0.5 for id in range(1, 100): retryCount = 0 while (retryCount < retryMaxCount): time.sleep(sleepTime) try: reqUrl(url + f"?id={id}") sleepTime /= 2 if sleepTime < 0.5: sleepTime = 0.5 break except Exception as e: sleepTime *= 2 if(sleepTime > 60): sleepTime = 60 retryCount += 1 if retryCount == retryMaxCount: print(f"{id} fail!") continue

创作不易，小小的支持一下吧！