用Python写一个代理IP池

文章中的代码仅供学习，请勿用于非法操作，包括但不限于非法获取数据、入侵其他计算机系统等

原理

构造一个代理池，最主要的问题就是池中的ip从哪里来，如果你自己有一个一些代理ip，可以将自己的ip放进去。我这里说的是，如果没有自己的ip，如何从其他地方获取ip。我获取ip的网站是https://www.89ip.cn/

IP池代码

import requests, re
from gevent import pool
from lxml import etree

class ProxyPool:
    def __init__(self, max_pool_size=100):
        self.url = "http://api.89ip.cn/tqdl.html?api=1&num=60"
        self.max_pool_size = max_pool_size
        self.pool = pool.Pool(max_pool_size)
        self.proxies = []
        self.refresh_proxies()

    def next(self):
        if not self.proxies:
            self.refresh_proxies()
        return self.proxies.pop()

    def is_valid_ip(self, ip,port):
        print(f'{ip}:{port}')
        pattern = r'^((25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)\.){3}(25[0-5]|2[0-4][0-9]|[01]?[0-9][0-9]?)$'
        flag = bool(re.match(pattern, ip))
        if not flag:
            return False
        else:
            # 测试Ip
            try:
                requests.get("https://baidu.com", proxies={
                    'http': f'http://{ip}:{port}',
                    'https': f'https://{ip}:{port}'
                }, timeout=6)
                print(f"Ip:{ip} is valid")
                return True
            except:
                return False

    def refresh_proxies(self):
        try:
            # todo 获取ip的方法，自定义
            response = requests.get(self.url)
            html = etree.HTML(response.text, etree.HTMLParser())
            result1 = html.xpath('/html/body/text()')  # /表示层级关系，第一个/是根节点
            for res in result1:
                _res = res.strip().replace('\n', '').replace('\r', '').replace('\t', '')
                list = _res.split(":")
                if len(_res) > 0 and self.is_valid_ip(list[0],list[1]):
                    self.proxies.append(_res)
                    if self.size() == self.max_pool_size:
                        break
        except Exception as e:
            print(f"Error refreshing proxies: {e}")
        # 判断下ip池的大小，如果不够，继续获取
        if self.size()<self.max_pool_size:
            self.refresh_proxies()

    def close(self):
        self.pool.join()

    def size(self):
        return self.pool.size

运行代码

if __name__ == "__main__":
    proxy_pool = ProxyPool( max_pool_size=20)
    ip_port = proxy_pool.next()

    # proxy
    req_url = "https://jhacker.cn/"
    proxies = {
        'http': f"http://{ip_port}",
        'https': f"http://{ip_port}"
    }
    res = requests.get(req_url, proxies=proxies)
    print(res)

    # 关闭线程池
    proxy_pool.close()