代理池3.0版本——基于tornado和redis实现动态代理池

一开始我是学习崔大神的flask和redis动态代理池，可惜里面的异步检查以及请求，还有元类的编程（看了整整一天，实在没有完全搞懂），我就算照着写也报错不断，于是今天自己写了一个代理池，也实现了一样的功能，先说一下代理池主要的思路，首先是启动一个主进程，然后分别创建两个子进程，一个是定期测试IP，一个是定期检查IP数量，符合要求则不用进行抓取，不符合要求则进行抓取，同时记得启动tornado框架，让我们可以在浏览器访问，然后获得代理。楼主对于代理的测试用的是线程池，用协程实在没有信心写好，git_hub地址是https://github.com/xiaobeibei26/dynamic_ip_pool
代码下载之后，大家只要运行了里面的run文件就不用管了，代理池会自己启动，自己刷新，你可用在别的脚本或者浏览器提取用就好了
这里先看一看关键的两个子进程的代码

from ip_pool.database import RedisConnect#导入数据库链接
import requests
import re
from ip_pool.ip_request import FreeProxyGetter
from ip_pool.Thread_pool import ThreadPool
import time
from multiprocessing import Process
check_time = 50#多久检查一次IP的有效性
count_time =50#多久严查一次IP数量是否够
lower_num_ip = 30#最少的IP数量
max_num_ip = 70#最多的IP数量

class Test_ip(object):#用于测试IP
    def __init__(self):
        self.url='http://ip.chinaz.com/getip.aspx'#用于测试的url
        self._conn=RedisConnect()
        self._raw_proxies =None
        self.Thread_pool =ThreadPool(10)
    def set_raw_proxies(self,proxies):#接收以备测试的代理
        self._raw_proxies =proxies

    def _test(self,proxy):#测试IP
        if isinstance(proxy,bytes):
            proxy=proxy.decode('utf-8')
        real_proxy= {'http':'http://{}'.format(proxy),
                    'https':'http://{}'.format(proxy),}
        print('正在测试，',proxy)
        try:
            html = requests.get(self.url, proxies=real_proxy, timeout=1)
            status_number = re.findall(r'\d\d\d', str(html))[0]  # 提取网页返回码
            re_ip = re.findall(r'\{ip', html.text)  # 有些 ip极其恶心，虽然返回的是200数字，表示正常，实则是bad request，这里去除掉
            if status_number == str(200):
                if re_ip:
                        # 检验代理是否能正常使用
                    self._conn.put(proxy)
                    print('网页返回状态码:', html, proxy, '代理有效,地址是：', html.text)
        except Exception as e:
            print('移除无效代理，',proxy)
    def Thread_test_ip(self,proxies):#传进去一个列表
        for proxy in proxies:
            self.Thread_pool.run(func=self._test, args=proxy)  # 用多线程测试
        print('本次测试用了多少线程',len(self.Thread_pool.generate_list))

class Get_ip(object):
    def __init__(self,max_ip_count=max_num_ip):#默认IP池里面最多的IP上面设置好了
        self._conn=RedisConnect()
        self.max_ip_count=max_ip_count
        self.crawl=FreeProxyGetter()
        self.Test=Test_ip()
        self.Thread_pool=ThreadPool(10)#用于多线程测试，这里设置最多10个线程
    def is_ip_enough(self):
        if self._conn.ip_count >= self.max_ip_count:#如果池里IP数大于规定最大的数量，则返回False
            return False
        return True

    def catch_ip(self):
        while self.is_ip_enough():
            print('代理数量不足，代理抓取中')
            for callback_lable in range(self.crawl.__CrawlFuncCount__):#该方法在元类里面添加了
                callback = self.crawl.__CrawlFunc__[callback_lable]
                raw_proxies = self.crawl.get_raw_proxies(callback)#这是接收抓取的代理
                if raw_proxies:
                    self.Test.Thread_test_ip(raw_proxies)
                else:
                    print('该源头没有代理')


class schedule(object):
    @staticmethod
    def check_ip(cycle_time=check_time):
        conn = RedisConnect()
        tester = Test_ip()
        while True:
            print('IP检查程序启动')
            count= int(0.5 * conn.ip_count)
            if count == 0:
                time.sleep(cycle_time)
                continue
            raw_proxies = conn.get_to_test(count)
            tester.Thread_test_ip(raw_proxies)#将列表传进去
            time.sleep(cycle_time)
    @staticmethod
    def catch_ip(cycle_time = count_time,max_ip=max_num_ip,min_ip=lower_num_ip):#时间多久检查一次IP数量是否足够
        conn= RedisConnect()
        Ip_catch = Get_ip()
        while True:
            if conn.ip_count <min_ip:#小于最少IP数量，启动抓取
                Ip_catch.catch_ip()
            print('代理暂时充足，不用抓取')
            time.sleep(cycle_time)

    def run(self):
        print('代理池启动')
        check_process =Process(target=schedule.check_ip)
        catch_process = Process(target=schedule.catch_ip)
        check_process.start()
        catch_process.start()

这么长的代码，我也懒得再看了，其主要功能就是同事运行两个进程，一个检查IP数量，一个测试质量
数据库的操作也很简单

import redis


class RedisConnect(object):
    def __init__(self):
        self.db=redis.Redis(host='localhost',port=6379)

    def get_to_test(self,count=1):#获取代理以备检查
        proxy = self.db.lrange('proxies',0,count-1)
        self.db.ltrim('proxies',count,-1)
        return proxy


    def put(self,proxy):#放进一个代理
        self.db.rpush('proxies',proxy)


    def pop(self):#获取一个代理，然后提出
        try:
            return self.db.rpop('proxies')
        except Exception as e:
            return  '没有代理可用'
    @property
    def ip_count(self):#获取代理数量
        return self.db.llen("proxies")

这里值得一提的是元类的编程，如图这是代理抓取的代码

from .html_request import MyRequest
from pyquery import PyQuery as pq
from bs4 import BeautifulSoup

class ProxyMetaclass(type):
    """
        __CrawlFunc__和__CrawlFuncCount__
        两个参数，分别表示爬虫函数，和爬虫函数的数量。
    """
    def __new__(cls, name, bases, attrs):#这几个是固定参数
        count = 0
        attrs['__CrawlFunc__'] = []
        for k, v in attrs.items():
            if 'crawl_' in k:#检查每个方法里面的key值是否有crawl字符，然后运行，方便以后添加代理的抓取进去
                attrs['__CrawlFunc__'].append(k)
                count += 1
        attrs['__CrawlFuncCount__'] = count
        return type.__new__(cls, name, bases, attrs)


class FreeProxyGetter(object, metaclass=ProxyMetaclass):

    def get_raw_proxies(self, callback):
        proxies = []
        print('Callback', callback)
        for proxy in eval("self.{}()".format(callback)):#eval这里用于执行字符串
            print('Getting', proxy, 'from', callback)
            proxies.append(proxy)
        return proxies

    def crawl_daili66(self, page_count=4):
        start_url = 'http://www.66ip.cn/{}.html'
        urls = [start_url.format(page) for page in range(1, page_count + 1)]
        for url in urls:
            print('Crawling', url)
            try:
                html = MyRequest.get(url,3)
                if html:
                    doc = pq(html)
                    trs = doc('.containerbox table tr:gt(0)').items()
                    for tr in trs:
                        ip = tr.find('td:nth-child(1)').text()
                        port = tr.find('td:nth-child(2)').text()
                        yield ':'.join([ip, port])
            except Exception as e:
                print('Crawling faild', url)

    def crawl_proxy360(self):
        start_url = 'http://www.proxy360.cn/Region/China'
        print('Crawling', start_url)
        try:
            html = MyRequest.get(start_url,3)
            if html:
                doc = pq(html)
                lines = doc('div[name="list_proxy_ip"]').items()
                for line in lines:
                    ip = line.find('.tbBottomLine:nth-child(1)').text()
                    port = line.find('.tbBottomLine:nth-child(2)').text()
                    yield ':'.join([ip, port])
        except:
            print('Crawling faild', start_url)

    def crawl_goubanjia(self):
        start_url = 'http://www.goubanjia.com/free/gngn/index.shtml'
        try:
            html = MyRequest.get(start_url,3)
            if html:
                doc = pq(html)
                tds = doc('td.ip').items()
                for td in tds:
                    td.find('p').remove()
                    yield td.text().replace(' ', '')
        except:
            print('Crawling faild', start_url)

    def crawl_haoip(self):
        start_url = 'http://haoip.cc/tiqu.htm'
        try:
            html =MyRequest.get(start_url,3)
            if html:
                doc = pq(html)
                results = doc('.row .col-xs-12').html().split('<br/>')
                for result in results:
                    if result: yield result.strip()
        except:
            print('Crawling faild', start_url)


    def crawl_xici(self):#西刺网的爬取
        start_url ='http://www.xicidaili.com/nn/1'
        try:
            data = MyRequest.get(start_url, 3)
            all_data = BeautifulSoup(data, 'lxml')
            all_ip = all_data.find_all('tr', class_='odd')
            for i in all_ip:
                ip = i.find_all('td')[1].get_text()  # ip
                port = i.find_all('td')[2].get_text()  # 端口
                proxy = (ip + ':' + port).strip()  # 组成成proxy代理
                if proxy:
                    yield proxy
        except:
            print('Crawling faild', start_url)

这里封装了多个网站的代理抓取，元类起的作用是根据这个类创建的对象，都具有了自定义new方法里面的属性，我们这里定义的是检查自身所定义的方法以及数量这个属性，后面要添加什么网站的代理抓取，直接添加进去就好了，不需要做任何改动，上两张结果图，代理不足的时候就会自动去抓取，也会定时测试可用代理，这就实现了动态代理池的功能

Paste_Image.png Paste_Image.png

我们在浏览器打开，一个显示的是当前代理可用的数量，一个是提取的代理，每次是提取一个

Paste_Image.png Paste_Image.png