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

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

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

目 录CONTENT

文章目录

python实现代理IP池

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

用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()
1

评论区