1.工具介绍

• 目前接入两个免费代理IP网站
• 使用API接口进行调用
• 使用Redis 进行数据存储
• 进行类封装，便于维护阅读
• 自动去重，以及校验可用IP

2. 开发原理

2.1 代理IP爬取

• 使用线程池提高爬取速度
• 存储到redis数据库，并且使用有序集合进行存储，自动去重

# -*- coding:utf-8 -*-
# 这里负责代理IP的采集工作
from proxy_redis import ProxyRedis
from multiprocessing import Process
from concurrent.futures import ThreadPoolExecutor
import requests
from lxml import etree
import time
headers = {
    'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36',
}
def get_kuai_ip(red):
    for i in range(1, 100):
        url = f'https://free.kuaidaili.com/free/intr/{i}/'
        resp = requests.get(url=url, headers=headers)
        tree = etree.HTML(resp.text)
        trs = tree.xpath('//table//tr')
        for tr in trs:
            ip = tr.xpath('./td[1]/text()')  # IP
            port = tr.xpath('./td[2]/text()')  # PORT
            if not ip:
                continue
            ip = ip[0]
            port = port[0]
            proxy_ip = ip + ':' + port
            # print(proxy_ip)
            red.add_proxy_ip(proxy_ip)  # 增加新的ip地址
        time.sleep(20)
def get_buzhidao_ip(red):
    url = 'https://ip.jiangxianli.com/?page=1'
    resp = requests.get(url=url, headers=headers)
    tree = etree.HTML(resp.text)
    trs = tree.xpath('//table//tr')
    for tr in trs:
        ip = tr.xpath('./td[1]/text()')
        port = tr.xpath('./td[2]/text()')
        if not ip:
            continue
        ip = ip[0]
        port = port[0]
        proxy_ip = ip + ':' + port
        # print(proxy_ip)
        red.add_proxy_ip(proxy_ip)  # 增加新的ip地址
def run():  # 启动
    red = ProxyRedis()  # 创建好red存储
    t = ThreadPoolExecutor(2)
    while True:
        try:
            t.submit(get_kuai_ip, red)  # 采集快代理
            t.submit(get_buzhidao_ip, red)  
        except Exception as e:
            print('出现错误 >> ', e)
            continue
        time.sleep(60)  # 每分钟跑一次
if __name__ == '__main__':
    run()

2.2 代理IP校验

• 使用Async异步爬虫提高IP验证速度
• 验证原理
  • 使用代理IP访问百度

# -*- coding:utf-8 -*-
# 负责IP代理的验证工作
import time
from proxy_redis import ProxyRedis
import asyncio
import aiohttp
async def verify_one(ip, sem, red):
    timeout = aiohttp.ClientTimeout(total=10)  # 10s 没回来就报错
    async with sem:
        try:
            async with aiohttp.ClientSession() as session:
                async with session.get("http://www.baidu.com", proxy="http://" + ip, timeout=timeout) as resp:
                    page_source = await resp.text()
                    if resp.status in [200, 302]:
                        print('[*]此IP可用 >> ', ip)
                        red.set_max_score(ip)
                    else:
                        # 有问题，扣分
                        print('[-]此IP不可用 >> ', ip)
                        red.des_incrby(ip)
        except Exception as e:
            print('[-]此IP不可用 >> ', ip)
            red.desc_incrby(ip)
async def main(red):
    # 1.把ip全部取出
    all_proxies = red.get_all_proxy()
    sem = asyncio.Semaphore(30)
    tasks = []
    for ip in all_proxies:
        tasks.append(asyncio.create_task(verify_one(ip, sem, red)))
    await asyncio.wait(tasks)
def run():
    red = ProxyRedis()
    time.sleep(10)
    while True:
        try:
            asyncio.run(main(red))
            time.sleep(100)
        except Exception as e:
            print('出现错误 >> ', e)
            time.sleep(100)
if __name__ == '__main__':
    run()

2.3 提供API接口访问

• 使用 sanic 库，提供Api接口

# -*- coding:utf-8 -*-
# 负责提供代理IP的接口
from proxy_redis import ProxyRedis
from sanic import Sanic, json
from sanic_cors import CORS
from settings import *
red = ProxyRedis()
app = Sanic('IP')
CORS(app)
@app.route('/')
def clhttp(req):
    ip = red.get_keyong_proxy()
    return json({'ip': ip})
def run():
    app.run(host=API_ADDRESS, port=API_PORT)
if __name__ == '__main__':
    run()

以上都是代理池的主要功能代码

3. 如何使用

• 首先安装 Redis，开启Redis服务
• 在settings.py文件中设置必要参数

• 一般来讲，设置这几个就行
• 设置完成之后，直接运行 runner.py 文件即可

• 运行效果

• 访问你设置的API接口，即可拿到代理IP

注意！注意！运行不来看看是不是缺少了库，如果仍有问题，可在论坛私聊

• Github地址