侧边栏壁纸
博主头像
小鱼吃猫博客博主等级

你所热爱的,便是你的生活。

  • 累计撰写 112 篇文章
  • 累计创建 47 个标签
  • 累计收到 9 条评论

目 录CONTENT

文章目录

python实现代理IP池

小鱼吃猫
2023-07-29 / 0 评论 / 1 点赞 / 31 阅读 / 2834 字

用Python写一个代理IP池

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

原理

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

IP池代码

import requests, re
from gevent import pool
from lxml import etree
<p>class ProxyPool:
def <strong>init</strong>(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()</p>
<pre><code>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) &gt; 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()&lt;self.max_pool_size:
        self.refresh_proxies()

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

def size(self):
    return self.pool.size</code></pre><h2 id="heading-4">运行代码</h2><pre><code class="language-python">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()</code></pre>
1

评论区