[限时投稿]-爬取免费代理，拥有自己的代理池- Track 知识社区 - 掌控安全在线教育

很久很久以前，我有个梦想，就是SQL注入不要被ban，于是ip代理成为了首选，但是奈何钱包有限，只能爬取免费代理，于是借鉴了许多文章，形成了今日的ip代理池。

python编写，内容主要涉及多线程问题、以及爬虫。

一、具体实现流程：

包括三个模块：
获取ip代理模块（爬虫获取ip），代理池模块（实现存储爬取的ip，写入有效ip代理），验证模块（对爬取的ip进行验证），获取模块（获取有效ip，并去重！）。

二、代码分析：（按程序执行顺序）代码分析：（按程序执行顺序）

（一）、入口文件：StartApi.py

1.Init: 获取必要的参数。
2.getproxy: 调用module.GetProxy，多线程使用exec函数，实例化3.module.GetProxy下的类，开始爬虫获取代理ip。
4.isproxy：调用module.IsProxy和module.OptPool类，循环判断获取代理池的ip，并进行判断。

startproxy：程序入口函数，三个模式：-c api 可获取已爬取存在ips.txt的有效代理0ip；-t 100 100个多线程各个爬虫爬取一次获取有效代理；-t 50 -m 1 -time 10 50个多线程无限次循环爬取有效ip，并且每10分钟进行循环。
具体代码如下：

# -*- coding:utf-8 -*-
# Author:qiuzishan
import module.GetProxy
from module.OptPool import ProxyPool
from module.IsProxy import IsProxy
from threading import Thread
import time
from queue import Queue
from argparse import ArgumentParser
from module.GetApi import GetApi


class StartApi:
    def __init__(self):
        arg = ArgumentParser(description='baidu_url_collection')
        arg.add_argument('-t', help='线程数量 默认10个线程', type=int, default=10)
        arg.add_argument('-c', help='默认爬取ip进行验证 -c api 即可获取ips.txt的ip值！', type=str, default='getproxy')
        arg.add_argument('-m', help='默认只爬取一次，-m 1 可实现无线循环，默认8分钟爬取一次', type=int, default=0)
        arg.add_argument('-time', help='默认8分钟爬取一次，-m 10 实现每10分钟爬取一次，搭配-m使用，最好不低于5分钟', type=int, default=8)
        args = arg.parse_args()
        self.que = Queue()
        self.c = args.c
        self.t = args.t  # 获取线程！！！这很重要！
        self.m = args.m
        self.time = args.time

    def getproxy(self):
        '''
        有几个代理，就会有几个线程爬取代理！
        :param msg:
        :return: 循环获取代理，这里可以加入获取代理的实例
        '''
        thread = []
        usemodels = ['BeautifulSoup', 'ProxyPool', '__builtins__', '__cached__', '__doc__', '__file__', '__loader__', '__name__', '__package__', '__spec__', 'datetime', 're', 'requests', 'time']
        getmodels = dir(module.GetProxy)
        for i in getmodels:
            if i not in usemodels:
                # 用于测试是否存在代理print(i)
                self.que.put('import module.GetProxy\nmodule.GetProxy.' + i + '()')
        proxyNum = self.que.qsize()  # 有几个爬虫就开启几个线程
        for i in range(0, proxyNum):
            t = Thread(target=exec, args=(self.que.get(),))
            thread.append(t)
        for i in thread:
            i.start()
            # print(f'线程数:{i}开始')
        for i in thread:
            i.join()

    def isproxy(self, msg):
        """
        :return: 循环判断代理
        """
        pp = ProxyPool()
        ip = IsProxy(self.t)
        print(msg)
        while True:
            flag = pp.get_length_pool()
            if flag == 0:
                print("退出循环")
                break  # 退出循环
            ip.startIsProxy()

    def startproxy(self):
        '''
        多进程同时运行
        '''
        if self.c == 'api':
            ga = GetApi()
            ga.start(self.t)
        else:
            if self.m:
                while True:
                    getproxy = Thread(target=self.getproxy)
                    isproxy = Thread(target=self.isproxy, args=('代理判断中',))
                    getproxy.start()
                    time.sleep(5)  # IpPool一开始无数据，因此先爬取加入IpPool列表中，防止出错
                    isproxy.start()
                    time.sleep(self.time*60)  # 每隔多长时间爬取一次
            else:
                getproxy = Thread(target=self.getproxy)
                isproxy = Thread(target=self.isproxy, args=('代理判断中',))
                getproxy.start()
                time.sleep(5)  # IpPool一开始无数据，因此先爬取加入IpPool列表中，防止出错
                isproxy.start()
                getproxy.join()
                isproxy.join()


if __name__ == '__main__':
    print('''

        III                        pppppppp                                      lll
        III                        pp     pp                                     lll
        III                        pp      pp                                    lll
        III  pppppppp              pp     pp        OOOOOOO        OOOOOOO       lll
        III  pp     pp             pppppppp        OO      OO     OO      OO     lll
        III  pp      pp            pp             OO        OO   OO        OO    lll
        III  pp     pp             pp             OO        OO   OO        OO    lll
        III  pppppppp   _________  pp              OO      OO     OO      OO     lll
             pp                    pp               OOOOOOOO       OOOOOOOO      lll
             pp       
             pp         {v 1.0 author:秋紫山 QQ:644976520 欢迎加q交流~~}
             pp                   

    ''')
    start = time.perf_counter()
    sa = StartApi()
    sa.startproxy()
    end = time.perf_counter()
    print("use time:"+str(end - start) + "s")

（二）爬取代理ip: module/GetProxy.py

自定义爬虫类：
1.Init：不需要改变，把print的内容稍微改改即可，实现实例化即可启动该爬虫，并把获取的爬虫加入代理池ProxyPool里面
2.get_lxml_by_url：获取需要爬取页面的内容
3.get_proxy_by_lxml：自定义的爬虫，需要返回ip列表。

具体代码：

例如ip88Proxy

# -*- coding:utf-8 -*-
# Author:qiuzishan
import requests
from bs4 import BeautifulSoup
from module.OptPool import ProxyPool
import time
import re
'''
这里可以定义多个代理类，只要保证以“**.**.**.**:**”形式如池即可，然后在getproxy.py中加入多线程即可。
'''


class demoProxy:  # demo
    '''
    Get Agent from web 89ip
    '''
    def __init__(self):  # 自动爬取并且把ip输入代理池！ 此段函数不需要更改,print 的内容更改一下即可
        self.proxy = []
        try:
            print("demo代理开始爬取")
            self.proxy = self.get_proxy_by_lxml(self.get_lxml_by_url())
        except Exception as e:
            print("出错了：{}\n~aha代理demo凉了，求你了快去看看吧！".format(e))
        if self.proxy != []:
            pp = ProxyPool()
            for p in self.proxy:
                pp.get_into_pool(p)
        else:
            print("demo代理爬取的代理ip为空！代理可能凉了！")

    def get_lxml_by_url(self):
        headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.149 Safari/537.36'
                }
        r = requests.get('', headers=headers)  # 输入需要爬取的网站！
        if r.status_code == 200:
            response = r.text
            return response
        else:
            exit('Error: Module GetAgent Url Error')

    def get_proxy_by_lxml(self, response):
        ip = []
        # 爬取你的代理包括端口噢，例如，123.12.2.33:8080,然后放进ip列表里。
        # 请输入您的爬虫~
        # print(ip)
        return ip


class ip89Proxy:  # 89ip
    '''
    Get Agent from web 89ip
    '''
    def __init__(self):
        self.proxy = []
        try:
            print("89ip代理开始爬取")
            self.proxy = self.get_proxy_by_lxml(self.get_lxml_by_url())
        except Exception as e:
            print("出错了：{}\n~aha代理89ip凉了，求你了快去看看吧！".format(e))
        if self.proxy != []:
            pp = ProxyPool()
            for p in self.proxy:
                pp.get_into_pool(p)
        else:
            print("89ip代理爬取的代理ip为空！代理可能凉了！")

    def get_lxml_by_url(self):
        headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/80.0.3987.149 Safari/537.36'
                }
        r = requests.get('https://www.89ip.cn/tqdl.html?num=200&amp;address=&amp;kill_address=&amp;port=&amp;kill_port=&amp;isp=', headers=headers)
        if r.status_code == 200:
            response = r.text
            return response
        else:
            print('Error: Module GetAgent Url Error')

    def get_proxy_by_lxml(self, response):
        ip = []
        soup = BeautifulSoup(response, 'html5lib')
        soup.find_all('div')
        for div in soup.find_all('div', style="padding-left:20px;"):
            for child in div.children:
                if child.string is not None:
                    if '更好用的代理ip请访问' not in child.string:
                        ip.append(child.string.strip())
        # print(ip)
        return ip

（三）ip加入代理池: module/OptPool.py

相信很容易看懂！

# -*- coding:utf-8 -*-
# Author:qiuzishan
from queue import Queue


# IpPool存储所有未检验的代理，Proxy用于存储可用的代理
IpPool = Queue()
Proxy = Queue()


class ProxyPool:
    '''
    代理池！
    '''

    def get_into_pool(self, args):  # 放进代理池一个ip
        '''
        :param args:
        :return: unknown IP warehousing
        '''
        IpPool.put(args)

    def get_length_pool(self):  # 返回代理池未验证的ip数量
        return IpPool.qsize()

    def get_out_pool(self):  # 取得代理池一个ip
        '''
        :return: unknown IP outgoing
        '''
        if IpPool.qsize():
            getip = IpPool.get()
            return getip
        else:
            print("IpPool变量无代理ip未检测")
            return 0  # 防止报错

    def get_into_proxy(self, args):  # 写入有效ip
        '''
        :param args:
        :return: effective IP warehousing
        '''
        Proxy.put(args)
        ip = Proxy.get()
        print(ip + '有效已写入ips.txt')
        with open('ips.txt', 'a', encoding='utf-8') as file:
            file.write(ip+'\n')

    def get_out_proxy(self):  # 输出有效ip
        '''
        :return: effective IP outgoing
        '''
        proxy_txt = []
        with open('ips.txt', 'r', encoding='utf-8') as file:
            self.pass_isproxy = file.readlines()
            for i in self.pass_isproxy:
                proxy_txt.append(i.replace('\n', ''))
            return proxy_txt

（四）验证代理池ip: module/IsProxy.py

详细的看这个文章吧 https://www.bilibili.com/read/cv13723818 社区的文章不知道有什么bug，改了一下附件就连内容也被修改了。