5分钟学会搭建一个自己的代理IP池

可能是世界上最简单的代理IP池，主要思路是1.>>从免费代理网站爬取代理，2.>>然后通过requests get请求验证，如果返回状态码200则保留，3.>>由于验证过程属于IO密集型操作，故采用多线程提高速度，最后返回可用代理。

import requests   
from fake_useragent import UserAgent  
import re  
import threading
import time
  
# 从两个免费代理网站获取代理，参数n默认为10，即在第一个网站中获取n-1个页面的代理  
# 更多网站请bing一下
def get_prox(n=10):  
    ua = UserAgent()  # 创建一个UserAgent对象，用于生成随机的User-Agent  
    prox_pool = []  # 初始化一个空的列表，用于存储获取到的代理  
      
    # 从第一个免费代理网站获取代理  
    for i in range(1, n):  # 循环从1到n-1，相当于手动翻页ing
        url1 = f'https://www.kuaidaili.com/free/fps/{i}/' 
        print(f'正在从{url1}获取代理...')  
        headers = {'User-Agent': ua.random}  # 设置请求头，其中User-Agent为随机生成的字符串，用于模拟真实浏览器访问  
        response1 = requests.get(url1, headers=headers)  # 发送GET请求获取代理列表页面内容  
        pattern = re.compile('<td data-title="IP">(.*?)</td>.*?<td data-title="PORT">(.*?)</td>', re.S)  # 定义正则表达式，用于从响应内容中提取IP地址和端口号
        plist1 = pattern.findall(response1.text)  # 使用正则表达式在响应内容中查找所有匹配的IP地址和端口号对  
        for pp in plist1:  
            prox = pp[0] + ':' + pp[1]  # 将IP地址和端口号拼接成完整的代理字符串形式，如"127.0.0.1:8080"  
            prox_pool.append(prox)  # 将拼接好的代理字符串添加到代理池中  
      
    # 从第二个免费代理网站获取代理  
    url2 = 'https://vpnoverview.com/privacy/anonymous-browsing/free-proxy-servers/' 
    print(f'正在从{url2}获取代理...') 
    headers = {'User-Agent': ua.random}
    response2 = requests.get(url2, headers=headers)   
    pattern2 = re.compile('<td><strong>(\d+\.\d+\.\d+\.\d+)</strong></td><td>(\d+)</td><td>', re.S)    
    plist2 = pattern2.findall(response2.text)     
    for pp in plist2:  
        prox = pp[0] + ':' + pp[1]  # 拼接代理字符串  
        prox_pool.append(prox) 
      
    print(f'共获取代理{len(prox_pool)}个{prox_pool}') 
    return prox_pool 

def test_single_prox(prox, idx, url='http://httpbin.org/get', timeout=8, max_retries=2):  
    retries = 0  
    # 当重试次数小于最大重试次数时，继续循环  
    while retries < max_retries:  
        time.sleep(0.5) # 小睡一下
        # 构建代理字典，用于http和https请求  
        pr = {  
            'http': 'http://' + prox,  
            'https': 'https://' + prox,  
        }  
        try:  
            # 使用requests库发起GET请求，设置代理和超时时间  
            response = requests.get(url, proxies=pr, timeout=timeout)  
            # 如果响应状态码是200，表示请求成功  
            if response.status_code == 200:  
                # 将工作的代理添加到工作代理列表中  
                working_proxies.append(prox)  
                # 打印出当前工作的代理信息  
                print(f"==> 撒花！Proxy {prox} #{idx} is working.")  
                # 如果代理工作正常，则无需继续重试，跳出循环  
                break  
            # 如果响应状态码不是200，则打印出代理失败的信息  
            else:  
                print(f"==> Proxy {prox} #{idx} failed with status code {response.status_code}.")  
                # 重试次数加1  
                retries += 1  
        # 捕获异常  
        except Exception as e:  
            # 重试次数加1  
            retries += 1  
            # 如果重试次数还未达到最大重试次数，打印超时信息，并继续重试  
            if retries < max_retries:  
                print(f"==> Proxy {prox} #{idx} filed. Retrying ({retries}/{max_retries})...")  
            # 如果达到最大重试次数仍然失败，打印异常信息
            else:  
                print(f"==> Proxy {prox} #{idx} failed with exception: {e}.")  

def run():
    prox_pool = get_prox(3)  # 从网站获取代理并保存到prox_pool变量中 (前两页，这个自己根据需要调整) 
    # 存储将要启动的线程  
    thread_list = []  
    # 遍历代理服务器列表，为每个代理服务器创建一个线程  
    for idx, prox in enumerate(prox_pool):    
        # 创建一个新的线程，目标函数是 test_single_prox
        t = threading.Thread(target=test_single_prox, args=(prox,idx,))  
        
        # 将新创建的线程添加到线程列表中  
        thread_list.append(t)  
    
    # 启动线程列表中的所有线程  
    for t in thread_list:  
        t.start()  
    # 等待所有线程完成  
    for t in thread_list:  
        t.join()

    print(f"Working proxies: {working_proxies}, total {len(working_proxies)} proxies.")  


if __name__=='__main__':
    # 存储验证通过的代理服务器
    working_proxies = []
    run()