如何配置属于自己的代理池

骑猪看日落 2022-12-06 12:52 156阅读 0赞

### 代理池 ###

*  （一）准备工作
 *  （二）代理池的目标
 *  （三）代理池的架构
 *  (四)选择db
 *   *  代理会出现的问题
     *  解决
     *  选择数据结构
     *  方法名及其作用
     *  代码实现
 *  (五)校验器
 *   *  代理校验可以通过什么完成
     *  代理的校验为什么用协程
     *  安装aiohttp
     *  代码实现
 *  (六)添加器
 *   *  什么时候开始添加
     *  什么时候结束添加
     *  代码实现
 *  (七)调度器
 *  (八)程序入口
 *   *  安装flask

# （一）准备工作 #

**首先需要成功安装 Redis 数据库并启动服务，另外还需要安装 aiohttp，asyncio，Redis， Flask 库。这些模块安装非常简单，使用 pip 命令就可以直接安装。**

# （二）代理池的目标 #

我们需要做到下面的几个目标，来实现易用高效的代理池。  
基本模块分为四块：**存储模块、获取模块、检测模块、接口模块**

*  存储模块：负责存储抓取下来的代理。存储模块要保证获取的代理能够按照一定的先后 顺序进行排序，同时方便对之前爬取的代理重新校验。基于这两点，我们选择使用 Redis 的 list，因为他可以用来实现队列，而队列可以实现我们的目的。
 *  获取模块：需要定时在各大代理网站抓取代理。代理可以是免费公开代理也可以是付费 代理，代理的形式都是 IP 加端口，此模块尽量从不同来源获取，尽量抓取高匿代理， 抓取成功之后将可用代理保存到数据库中。
 *  检测模块：检测模块的主要作用就是测验代理能否使用，测验的方法也很简单，设置一个目标网站并设置好代理，如果能够返回内容不超时，就说明这个代理是可用的。检测 模块我们使用了异步请求（因为检测代理是一个非常耗时的事情，如果用同步请求，那 么程序的性能就会大大降低）。代理如果检测可用，就将代理添加到代理池中.
 *  添加模块：添加模块主要逻辑是：如果代理池的数量没有达到设置的最大代理数，那么 就会循环调用获取模块获取代理，同时调用检测模块来检测，并将可用的代理添加到代 理池。
 *  接口模块：将代理池获取代理的功能架设到 Flask 上，那么只需要访问一个 url 就可以 获取一个代理，使代理池的使用变得非常方便。
 *  调度器模块：设置两个进程，一个进程循环检测代理的数量，如果代理池数量不足，就 调用添加模块不断的添加内容。另一个进程不断的循环检验代理，这里检验的代理是之 前在代理池中爬下来的可用代理，因为代理不一定永远可用，是有时效性的。这两个进 程都设置一个循环校验时间，避免频繁校验和添加

# （三）代理池的架构 #

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ExMzA5NTI1ODAy_size_16_color_FFFFFF_t_70_pic_center]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ExMzA5NTI1ODAy_size_16_color_FFFFFF_t_70_pic_center 1]

# (四)选择db #

## 代理会出现的问题 ##

代理是有时效性的,有些代理校验成功,一会就可能失效了

## 解决 ##

当添加db组件的有效代理时,应该**定期检查**和**保持顺序**

## 选择数据结构 ##

redis的list,因为list可以实现队列

## 方法名及其作用 ##

**pop() ---->获取代理的方法,从尾部获取新的代理  
get() ---->从头部获取代理,用于定期检查  
put() ------> 在尾部添加代理**  
![在这里插入图片描述][20200913161637519.png_pic_center]

## 代码实现 ##

settings.py

#密码：如果为空，就没有密码
    PASSWORD = ''
    #主机ip和端口号
    HOST = 'localhost'
    PORT = 6379
    
    #代理池的名称
    PROXIES = 'proxies_new'
    #测试代理的网址
    TSET_API = 'https://www.baidu.com/'
    #配置测试网址的请求头
    TEST_REQUEST_HEADERS = { 
            'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.79 Safari/537.36',
    }
    #测试代理的超时时间
    TSET_TIME_OUT = 30
    #循环校验时间
    CYCLE_VALID_TIME = 100
    #代理池代理数量的最小值
    LOWER_THRESHOLD = 10
    #代理池代理的最大值
    UPPER_THRESHOLD = 100
    #循环检查时间
    CYCLE_CHECK_TIME = 100

db.py

import redis
    from proxypool.setttings import PASSWORD,HOST,PORT,PROXIES
    class Reids_Client(object):
        def __init__(self,host=HOST,port = PORT):
            #判断是否有密码：
            if PASSWORD:
                self._db =redis.Redis(host=host,port=port,password=PASSWORD)
            else:
                self._db = redis.Redis(host=host,port=port)
        #将代理添加到代理池的尾部
        def put(self,proxy):
            self._db.rpush(PROXIES,proxy)
    
        #从头部获取count个代理,并将其删除
        def get(self,count=1):
            #lrange：获取指定范围的内容
            #ltrim :保留指定范围内的内容，其他删除
            proxies = self._db.lrange(PROXIES,0,count-1)
            self._db.ltrim(PROXIES,count,-1)
            return proxies
    
        #从尾部获取一个最新代理
        def pop(self):
            #从redis中获取处来的数据是bytes
            return self._db.rpop(PROXIES).decode('utf-8')
    
        #计算代理池长度
        @property
        def queue_len(self):
            return self._db.llen(PROXIES)
    
        def flush(self):
            self._db.flushall()
    
    if __name__ == '__main__':
        r = Reids_Client()
        print(r.queue_len)

# (五)校验器 #

## 代理校验可以通过什么完成 ##

1.  requests(耗时)
2.  aiohttp模块(异步请求)

## 代理的校验为什么用协程 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ExMzA5NTI1ODAy_size_16_color_FFFFFF_t_70_pic_center 2]

## 安装aiohttp ##

pip install aiohttp

## 代码实现 ##

Scheduler.py

import time
    from multiprocessing import Process
    from threading import Thread
    import aiohttp
    from proxypool.setttings import *
    
    import asyncio
    from proxypool.db import Reids_Client
    from proxypool.getter import FreeProxyGetter
    
    class VaildityTester(object):
        def __init__(self):
            #篮子
            self._raw_proxies = []
    
        #向篮子放东西
        def set_raw_proxies(self,proxies):
            self._raw_proxies = proxies
            # 数据的所有连接，用的时候在创建。
            self._conn = Reids_Client()
        #校验代理，要使用异步 请求
        async def test_single_proxy(self,proxy):
            try:
                # 创建一个session对象
                async with aiohttp.ClientSession() as session:
                    # 参数校验,如果proxy参数是一个bytes类型，就给他转成字符串
                    if isinstance(proxy, bytes):
                        proxy = proxy.decode('utf-8')
                    real_proxy = 'http://' + proxy
                    try:
                        async with session.get(TSET_API,
                                               headers=TEST_REQUEST_HEADERS,
                                               proxy=real_proxy,
                                               timeout=TSET_TIME_OUT) as response:
                            if response.status == 200:
                                # 该代理可用
                                # 添加到代理池
                                self._conn.put(proxy)
                                print('有效代理！', proxy)
                    except Exception:
                        print('无效代理！',proxy)
            except Exception as e:
                print(e)
        #校验器的校验方法
        def test(self):
            print('代理池开始启动！')
            #创建一个loop（任务执行链）
            loop = asyncio.get_event_loop()
            #执行任务
            tasks = [self.test_single_proxy(proxy) for proxy in self._raw_proxies]
            #启动loop
            loop.run_until_complete(asyncio.wait(tasks))

**获取免费代理**  
getter.py

import requests
    from lxml import etree
    class ProxyMetaclass(type):
        def __new__(cls, name,bases,attrs):
            # print('元类启动！')
            # print('attr1:',attrs)
            #创建一个属性，这个属性是一个list，里面存放的是所有爬取免费代理方法的方法名字。
            attrs['__CrwalFunc__'] = []
            count = 0
            for k,v in attrs.items():
                if 'crawl_' in k:
                    attrs['__CrwalFunc__'].append(k)
                    count +=1
            attrs['__CrwalCount__'] = count
            # print('attrs2',attrs)
            return type.__new__(cls,name,bases,attrs)
    
    
    class FreeProxyGetter(object,metaclass=ProxyMetaclass):
    
        headers = { 
            'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.79 Safari/537.36',
        }
        def get_raw_proxies(self,callback):
            ''' 通过传入一个字符串的方法名，来调用这个方法，获取代理 :param callback: 方法名的字符串格式--crawl_66ip :return: list-->代理 '''
            proxies = []
            for proxy in eval('self.{}()'.format(callback)):
                proxies.append(proxy)
            return proxies
    
        #爬取代理方法，名称统一为crawl_
        def crawl_66ip(self):
            ''' url:http://www.66ip.cn/ :return:list[proxy] '''
            proxies = []
            base_url = 'http://www.66ip.cn/%s.html'
            for i in range(1, 20):
                response = requests.get(base_url % i, headers=self.headers)
                html = etree.HTML(response.text)
                ips = html.xpath('//tr[position()>1]/td[1]/text()')
                ports = html.xpath('//tr[position()>1]/td[2]/text()')
                if len(ips) == len(ports) and ips and ports:
                    for i, ip in enumerate(ips):
                        port = ports[i]
                        # print(ip,port)
                        # proxies.append(ip.strip()+':'+port.strip())
                        yield ip.strip()+':'+port.strip()
        def crawl_ip3366(self):
            ''' url:http://www.ip3366.net/?stype=1&page=1 :return:list[proxy] '''
            proxies = []
            base_url = 'http://www.ip3366.net/?stype=1&page=%s'
            for i in range(1, 11):
                response = requests.get(base_url % i, headers=self.headers)
                html = etree.HTML(response.text)
                ips = html.xpath('//tr/td[1]/text()')
                ports = html.xpath('//tr/td[2]/text()')
                if len(ips) == len(ports) and ips and ports:
                    for i, ip in enumerate(ips):
                        port = ports[i]
                        # print(ip,port)
                        # proxies.append(ip.strip()+':'+port.strip())
                        yield ip.strip() + ':' + port.strip()
    if __name__ == '__main__':
        f = FreeProxyGetter()
        print(f.__CrwalFunc__)
        # f.crawl_66ip()
        # f.crawl_ip3366()

# (六)添加器 #

## 什么时候开始添加 ##

**假设我们只需要10<proxies<100个代理即可  
所以我们只需要当代理数小于10时则开始添加**

## 什么时候结束添加 ##

**我们只需要当代理数大于100时则结束添加**

## 代码实现 ##

Scheduler.py

....
    #添加器
    class PoolAdder(object):
        #threshold--阈值--代理池最大值
        def __init__(self,threshold):
            self._threshold = threshold
            #校验
            self._tester = VaildityTester()
            #db
            self._conn = Reids_Client()
            self._crawler =FreeProxyGetter()
        #判断代理池数量是否达到最大值
        def is_over_threshold(self):
            ''' :return: True:达到 False不能 '''
            if self._conn.queue_len>=self._threshold:
                return True
            return False
        #添加代理到代理池的方法
        def add_to_queue(self):
            #代理的获取是从getter组件组件中获取的。
            print('添加器开始工作....')
            while True:
                #当添加到代理池数量达到最大值，就不添加了。
                if self.is_over_threshold():
                    break
                proxy_count = 0
                #1、先从网上获取免费代理
                #问题：现在只能调用单一的爬取代理的方法，无法从各个网站都获取代理
                #如何实现？
                # for crawl in [crawl_ip3366,crawl_66ip]:
                # proxies = self._crawler.crawl_ip3366()
                # proxies = self._crawler.crawl_66ip()
                for crawl in self._crawler.__CrwalFunc__:
                    #现在我们拿到了方法的名字：crawl_66ip-->str--->eval()
                    try:
                        raw_proxies = self._crawler.get_raw_proxies(crawl)
                    except Exception:
                        continue
                    #2、用校验器校验（校验的同时在添加）
                    #2.1先把代理放到篮子
                    self._tester.set_raw_proxies(raw_proxies)
                    #2.2test方法就自动从篮子中获取代理校验
                    self._tester.test()
                    proxy_count+=len(raw_proxies)
                if proxy_count==0:
                    print('代理网站全部失效，请添加！')
                    raise RuntimeError('代理网站全部失效！')

# (七)调度器 #

Scheduler.py

.....
    class Scheduler(object):
        @staticmethod
        def vaild_proxy(cycle = CYCLE_VALID_TIME):
            conn = Reids_Client()
            tester = VaildityTester()
            while True:
                print('循环校验器开始启动！')
                count = int(conn.queue_len*0.5)
                if count ==0:
                    print('代理池数量不足！正在添加...')
                    time.sleep(cycle)
                #从代理池头部取出count个代理，进行校验
                raw_proxies = conn.get(count)
                #调用校验器
                # 2.1先把代理放到篮子
                tester.set_raw_proxies(raw_proxies)
                # 2.2test方法就自动从篮子中获取代理校验
                tester.test()
                time.sleep(cycle)
        @staticmethod
        def check_pool_add(lower_threshold = LOWER_THRESHOLD,
                           upper_threshold = UPPER_THRESHOLD,
                           cycle=CYCLE_CHECK_TIME):
            conn = Reids_Client()
            adder = PoolAdder(upper_threshold)
            while True:
                #如果代理池数量小于最小值，调用添加器添加
                if conn.queue_len<lower_threshold:
                    adder.add_to_queue()
                time.sleep(cycle)
        def run(self):
            vaild_process =Process(target=Scheduler.vaild_proxy)
            check_process =Process(target=Scheduler.check_pool_add)
            vaild_process.start()
            check_process.start()

# (八)程序入口 #

## 安装flask ##

pip install flask

**api.py**

from flask import Flask,g
    
    from proxypool.db import Reids_Client
    
    __all__=['app']
    app = Flask(__name__)
    def get_conn():
        if not hasattr(g,'reids_client'):
            g.redis_client = Reids_Client()
        return g.redis_client
    
    @app.route('/')
    def index():
        return '<h1>欢迎来到代理池系统！</h1>'
    @app.route('/get')
    def get():
        conn = get_conn()
        proxy = conn.pop()
        if isinstance(proxy,bytes):
            return proxy.decode('utf-8')
        return proxy
    @app.route('/count')
    def count():
        return str(get_conn().queue_len)

run.py

#代理池
    from proxypool.api import app
    from proxypool.Scheduler import Scheduler
    def main():
        s = Scheduler()
        s.run()
        app.run()
    
    if __name__ == '__main__':
        main()

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ExMzA5NTI1ODAy_size_16_color_FFFFFF_t_70_pic_center 3]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ExMzA5NTI1ODAy_size_16_color_FFFFFF_t_70_pic_center]: /images/20221123/2dfd9c0b76f748c2a76f087620933fa6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ExMzA5NTI1ODAy_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221123/d66609a0411746609428c5144a5ed57d.png
[20200913161637519.png_pic_center]: /images/20221123/a5c7d720a1b543e7a9384364abae4341.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ExMzA5NTI1ODAy_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221123/967c14155c5b4d6fa4c3ad1a7375d83a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2ExMzA5NTI1ODAy_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221123/06a2029e428b4a74aefcb78fd368d27c.png