proxy_pool/Schedule/ProxyRefreshSchedule.py

# -*- coding: utf-8 -*-
# !/usr/bin/env python
"""
-------------------------------------------------
   File Name：     ProxyRefreshSchedule.py  
   Description :  代理定时刷新
   Author :       JHao
   date：          2016/12/4
-------------------------------------------------
   Change Activity:
                   2016/12/4: 代理定时刷新
-------------------------------------------------
"""
import logging
import random
from multiprocessing import Process

import requests
from apscheduler.schedulers.blocking import BlockingScheduler
import sys
sys.path.append('../')
from DB.DbClient import DbClient
from Manager.ProxyManager import ProxyManager

__author__ = 'JHao'


class ProxyRefreshSchedule(ProxyManager):
    """
    代理定时刷新
    """

    def __init__(self):
        ProxyManager.__init__(self)

    def valid_proxy(self):
        self.db.changeTable(self.raw_proxy_queue)
        raw_proxy = self.db.pop()
        print '[*]check raw proxy {} ...'.format(raw_proxy)
        while raw_proxy:
            proxies = {"http": "http://{proxy}".format(proxy=raw_proxy),
                       "https": "https://{proxy}".format(proxy=raw_proxy)}
            try:
                # 超过30秒的代理就不要了
                r = requests.get('https://www.baidu.com/', proxies=proxies, timeout=30, verify=False)
                if r.status_code == 200:
                    self.db.changeTable(self.useful_proxy_queue)
                    self.db.put(raw_proxy)
            except Exception, e:
                print e
                pass
            self.db.changeTable(self.raw_proxy_queue)
            raw_proxy = self.db.pop()
            print 'validate a  proxy'

    def validate_useful_proxy(self, proxy_list):
        self.db.changeTable(self.useful_proxy_queue)
        for proxy in proxy_list:
            print '[*]validating proxy : {} ...({} remained)'.format(proxy, len(proxy_list))
            proxies = {"http": "http://{proxy}".format(proxy=proxy),
                       "https": "https://{proxy}".format(proxy=proxy)}
            try:
                r = requests.get('https://www.baidu.com/', proxies=proxies, timeout=30, verify=False)
                if r.status_code == 200:
                    continue
            except Exception, e:
                self.db.delete(proxy)
                print '[-]delete proxy {}'.format(proxy)


def refresh_pool():
    pp = ProxyRefreshSchedule()
    pp.valid_proxy()


def validate_user_proxy(proxy_list):
    pp = ProxyRefreshSchedule()
    pp.validate_useful_proxy(proxy_list)


def main(process_num=10):
    p = ProxyRefreshSchedule()
    p.refresh()
    pl = []
    for num in range(process_num):
        proc = Process(target=refresh_pool, args=())
        proc.daemon = True
        pl.append(proc)

    for num in range(process_num):
        pl[num].start()

    print 'All raw_proxy_crawler sub-processes start.'

    for num in range(process_num):
        pl[num].join()


def main_check(process_num=10):
    db = DbClient()
    useful_proxy_queue = 'useful_proxy_queue'
    db.changeTable(useful_proxy_queue)
    proxy_list = db.getAll()

    uncheck_list = [list() for i in xrange(process_num)]
    for proxy in proxy_list:
        uncheck_list[random.randint(0, process_num - 1)].append(proxy)

    pl = []
    for num in range(process_num):
        proc = Process(target=validate_user_proxy, args=(uncheck_list[num],))
        proc.daemon = True
        pl.append(proc)

    for num in range(process_num):
        pl[num].start()

    print 'All proxy validator sub-processes start.'

    for num in range(process_num):
        pl[num].join()


# def main(process_num=100):
#     p = ProxyRefreshSchedule()
#     p.refresh()
#     for num in range(process_num):
#         P = Process(target=refreshPool, args=())
#         P.daemon = True
#         P.start()
#         P.join()
# print '{time}: refresh complete!'.format(time=time.ctime())

if __name__ == '__main__':
    log = logging.getLogger('apscheduler')
    log.setLevel(logging.INFO)  # DEBUG

    fmt = logging.Formatter('%(levelname)s:%(name)s:%(message)s')
    h = logging.StreamHandler()
    h.setFormatter(fmt)
    log.addHandler(h)

    # main()
    sched = BlockingScheduler()
    # sched.add_job(main, 'interval', seconds=10)
    sched.add_job(main_check, 'interval', seconds=15)
    sched.start()