[update] 优化调度程序

Api/ProxyApi.py

@@ -37,11 +37,11 @@ class JsonResponse(Response):
 app.response_class = JsonResponse
 
 api_list = {
-    'get': u'get an usable proxy',
+    'get': u'get an useful proxy',
     # 'refresh': u'refresh proxy pool',
     'get_all': u'get all proxy from proxy pool',
     'delete?proxy=127.0.0.1:8080': u'delete an unable proxy',
-    'get_status': u'proxy statistics'
+    'get_status': u'proxy number'
 }
 
 
@@ -53,7 +53,7 @@ def index():
 @app.route('/get/')
 def get():
     proxy = ProxyManager().get()
-    return proxy if proxy else 'no proxy!'
+    return proxy.info_json if proxy else {"code": 0, "src": "no proxy"}
 
 
 @app.route('/refresh/')
@@ -74,7 +74,7 @@ def getAll():
 def delete():
     proxy = request.args.get('proxy')
     ProxyManager().delete(proxy)
-    return 'success'
+    return {"code": 0, "src": "success"}
 
 
 @app.route('/get_status/')

Config/setting.py

@@ -11,8 +11,24 @@
 -------------------------------------------------
 """
 
-# database config
+import sys
 from os import getenv
+from logging import getLogger
+
+log = getLogger(__name__)
+
+HEADER = """
+    ______                        ______             _
+    | ___ \_                      | ___ \           | |
+    | |_/ / \__ __   __  _ __   _ | |_/ /___   ___  | |
+    |  __/|  _// _ \ \ \/ /| | | ||  __// _ \ / _ \ | |
+    | |   | | | (_) | >  < \ |_| || |  | (_) | (_) || |___
+    \_|   |_|  \___/ /_/\_\ \__  |\_|   \___/ \___/ \_____\
+                           __ / /
+                          /___ /
+"""
+
+PY3 = sys.version_info >= (3,)
 
 
 class ConfigError(BaseException):

DB/DbClient.py

@@ -17,34 +17,37 @@ import os
 import sys
 
 from Config.ConfigGetter import config
-from Util.utilClass import Singleton
+from Util import Singleton
 
 sys.path.append(os.path.dirname(os.path.abspath(__file__)))
 
 
 class DbClient(object):
     """
-    DbClient DB工厂类 提供get/put/pop/delete/getAll/changeTable方法
+    DbClient DB工厂类 提供get/put/update/pop/delete/exists/getAll/clean/getNumber/changeTable方法
 
-    目前存放代理的table/collection/hash有两种：
+    目前存放代理的有两种, 使用changeTable方法切换操作对象：
         raw_proxy： 存放原始的代理；
-        useful_proxy_queue： 存放检验后的代理；
+        useful_proxy： 存放检验后的代理；
+
 
     抽象方法定义：
-        get(proxy): 返回proxy的信息；
-        put(proxy): 存入一个代理；
-        pop(): 弹出一个代理
-        exists(proxy)： 判断代理是否存在
-        getNumber(raw_proxy): 返回代理总数（一个计数器）；
-        update(proxy, num): 修改代理属性计数器的值;
-        delete(proxy): 删除指定代理；
-        getAll(): 返回所有代理；
-        changeTable(name): 切换 table or collection or hash;
+        get(proxy): 返回指定proxy的信息;
+        put(proxy): 存入一个proxy信息;
+        pop(): 返回并删除一个proxy信息;
+        update(proxy): 更新指定proxy信息;
+        delete(proxy): 删除指定proxy;
+        exists(proxy): 判断指定proxy是否存在;
+        getAll(): 列表形式返回所有代理;
+        clean(): 清除所有proxy信息;
+        getNumber(): 返回proxy数据量;
+        changeTable(name): 切换操作对象 raw_proxy/useful_proxy
 
 
         所有方法需要相应类去具体实现：
-            SSDB：SsdbClient.py
-            REDIS:RedisClient.py  停用 统一使用SsdbClient.py
+            ssdb: SsdbClient.py
+            redis: RedisClient.py
+            mongodb: MongodbClient.py
 
     """
 
@@ -98,6 +101,9 @@ class DbClient(object):
     def getAll(self):
         return self.client.getAll()
 
+    def clear(self):
+        return self.client.clear()
+
     def changeTable(self, name):
         self.client.changeTable(name)
 

DB/SsdbClient.py

@@ -15,20 +15,19 @@
 """
 __author__ = 'JHao'
 
-from Util import EnvUtil
+from Config.setting import PY3
 
 from redis.connection import BlockingConnectionPool
 from redis import Redis
-import random
 
 
 class SsdbClient(object):
     """
     SSDB client
 
-    SSDB中代理存放的容器为hash：
-        原始代理存放在name为raw_proxy的hash中，key为代理的ip:port，value为None,以后扩展可能会加入代理属性；
-        验证后的代理存放在name为useful_proxy的hash中，key为代理的ip:port，value为一个计数,初始为1，每校验失败一次减1；
+    SSDB中代理存放的结构为hash：
+        原始代理存放在name为raw_proxy的hash中, key为代理的ip:por, value为代理属性的字典;
+        验证后的代理存放在name为useful_proxy的hash中, key为代理的ip:port, value为代理属性的字典;
 
     """
     def __init__(self, name, **kwargs):
@@ -43,75 +42,100 @@ class SsdbClient(object):
         self.name = name
         self.__conn = Redis(connection_pool=BlockingConnectionPool(**kwargs))
 
-    def get(self, proxy):
+    def get(self, proxy_str):
         """
-        get an item
         从hash中获取对应的proxy, 使用前需要调用changeTable()
-        :param proxy:
+        :param proxy_str: proxy str
         :return:
         """
-        data = self.__conn.hget(name=self.name, key=proxy)
+        data = self.__conn.hget(name=self.name, key=proxy_str)
         if data:
-            return data.decode('utf-8') if EnvUtil.PY3 else data
+            return data.decode('utf-8') if PY3 else data
         else:
             return None
 
-    def put(self, proxy, num=1):
+    def put(self, proxy_obj):
         """
         将代理放入hash, 使用changeTable指定hash name
-        :param proxy:
-        :param num:
+        :param proxy_obj: Proxy obj
         :return:
         """
-        data = self.__conn.hset(self.name, proxy, num)
+        data = self.__conn.hset(self.name, proxy_obj.proxy, proxy_obj.info_json)
         return data
 
-    def delete(self, key):
+    def delete(self, proxy_str):
         """
-        Remove the ``key`` from hash ``name``
-        :param key:
+        移除指定代理, 使用changeTable指定hash name
+        :param proxy_str: proxy str
         :return:
         """
-        self.__conn.hdel(self.name, key)
+        self.__conn.hdel(self.name, proxy_str)
 
-    def update(self, key, value):
-        self.__conn.hincrby(self.name, key, value)
+    def exists(self, proxy_str):
+        """
+        判断指定代理是否存在, 使用changeTable指定hash name
+        :param proxy_str: proxy str
+        :return:
+        """
+        return self.__conn.hexists(self.name, proxy_str)
+
+    def update(self, proxy_obj):
+        """
+        更新 proxy 属性
+        :param proxy_obj:
+        :return:
+        """
+        self.__conn.hset(self.name,  proxy_obj.proxy, proxy_obj.info_json)
 
     def pop(self):
         """
         弹出一个代理
         :return: dict {proxy: value}
         """
-        proxies = self.__conn.hkeys(self.name)
-        if proxies:
-            proxy = random.choice(proxies)
-            value = self.__conn.hget(self.name, proxy)
-            self.delete(proxy)
-            return {'proxy': proxy.decode('utf-8') if EnvUtil.PY3 else proxy,
-                    'value': value.decode('utf-8') if EnvUtil.PY3 and value else value}
+        # proxies = self.__conn.hkeys(self.name)
+        # if proxies:
+        #     proxy = random.choice(proxies)
+        #     value = self.__conn.hget(self.name, proxy)
+        #     self.delete(proxy)
+        #     return {'proxy': proxy.decode('utf-8') if PY3 else proxy,
+        #             'value': value.decode('utf-8') if PY3 and value else value}
         return None
 
-    def exists(self, key):
-        return self.__conn.hexists(self.name, key)
-
     def getAll(self):
+        """
+        列表形式返回所有代理, 使用changeTable指定hash name
+        :return:
+        """
         item_dict = self.__conn.hgetall(self.name)
-        if EnvUtil.PY3:
-            return {key.decode('utf8'): value.decode('utf8') for key, value in item_dict.items()}
+        if PY3:
+            return [value.decode('utf8') for key, value in item_dict.items()]
         else:
-            return item_dict
+            return item_dict.values()
+
+    def clear(self):
+        """
+        清空所有代理, 使用changeTable指定hash name
+        :return:
+        """
+        return self.__conn.execute_command("hclear", self.name)
 
     def getNumber(self):
         """
-        Return the number of elements in hash ``name``
+        返回代理数量
         :return:
         """
         return self.__conn.hlen(self.name)
 
     def changeTable(self, name):
+        """
+        切换操作对象
+        :param name: raw_proxy/useful_proxy
+        :return:
+        """
         self.name = name
 
 
 if __name__ == '__main__':
-    c = SsdbClient(name='useful_proxy', host='127.0.0.1', port=8899, password=None)
-    print(c.getAll())
+    c = SsdbClient(name='raw_proxy', host='120.79.78.193', port=8888, password=None)
+    c.get("103.194.233.188:80810")
+    print(c.clear())

Manager/ProxyManager.py

@@ -15,7 +15,7 @@ __author__ = 'JHao'
 
 import random
 
-from Util import EnvUtil
+from ProxyHelper import Proxy
 from DB.DbClient import DbClient
 from Config.ConfigGetter import config
 from Util.LogHandler import LogHandler
@@ -34,27 +34,36 @@ class ProxyManager(object):
         self.log = LogHandler('proxy_manager')
         self.useful_proxy_queue = 'useful_proxy'
 
-    def refresh(self):
+    def fetch(self):
         """
-        fetch proxy into Db by ProxyGetter/getFreeProxy.py
+        fetch proxy into db by ProxyGetter
         :return:
         """
         self.db.changeTable(self.raw_proxy_queue)
+        proxy_set = set()
+        self.log.info("ProxyFetch : start")
         for proxyGetter in config.proxy_getter_functions:
-            # fetch
+            self.log.info("ProxyFetch - {func}: start".format(func=proxyGetter))
             try:
-                self.log.info("{func}: fetch proxy start".format(func=proxyGetter))
                 for proxy in getattr(GetFreeProxy, proxyGetter.strip())():
-                    # 直接存储代理, 不用在代码中排重, hash 结构本身具有排重功能
                     proxy = proxy.strip()
-                    if proxy and verifyProxyFormat(proxy):
-                        self.log.info('{func}: fetch proxy {proxy}'.format(func=proxyGetter, proxy=proxy))
-                        self.db.put(proxy)
+
+                    if not proxy or not verifyProxyFormat(proxy):
+                        self.log.error('ProxyFetch - {func}: '
+                                       '{proxy} illegal'.format(func=proxyGetter, proxy=proxy.ljust(20)))
+                        continue
+                    elif proxy in proxy_set:
+                        self.log.info('ProxyFetch - {func}: '
+                                      '{proxy} exist'.format(func=proxyGetter, proxy=proxy.ljust(20)))
+                        continue
                     else:
-                        self.log.error('{func}: fetch proxy {proxy} error'.format(func=proxyGetter, proxy=proxy))
+                        self.log.info('ProxyFetch - {func}: '
+                                      '{proxy} success'.format(func=proxyGetter, proxy=proxy.ljust(20)))
+                        self.db.put(Proxy(proxy, source=proxyGetter))
+                        proxy_set.add(proxy)
             except Exception as e:
-                self.log.error("{func}: fetch proxy fail".format(func=proxyGetter))
-                continue
+                self.log.error("ProxyFetch - {func}: error".format(func=proxyGetter))
+                self.log.error(str(e))
 
     def get(self):
         """
@@ -62,23 +71,20 @@ class ProxyManager(object):
         :return:
         """
         self.db.changeTable(self.useful_proxy_queue)
-        item_dict = self.db.getAll()
-        if item_dict:
-            if EnvUtil.PY3:
-                return random.choice(list(item_dict.keys()))
-            else:
-                return random.choice(item_dict.keys())
+        item_list = self.db.getAll()
+        if item_list:
+            random_choice = random.choice(item_list)
+            return Proxy.newProxyFromJson(random_choice)
         return None
-        # return self.db.pop()
 
-    def delete(self, proxy):
+    def delete(self, proxy_str):
         """
         delete proxy from pool
-        :param proxy:
+        :param proxy_str:
         :return:
         """
         self.db.changeTable(self.useful_proxy_queue)
-        self.db.delete(proxy)
+        self.db.delete(proxy_str)
 
     def getAll(self):
         """
@@ -86,10 +92,8 @@ class ProxyManager(object):
         :return:
         """
         self.db.changeTable(self.useful_proxy_queue)
-        item_dict = self.db.getAll()
-        if EnvUtil.PY3:
-            return list(item_dict.keys()) if item_dict else list()
-        return item_dict.keys() if item_dict else list()
+        item_list = self.db.getAll()
+        return [Proxy.newProxyFromJson(_).info_dict for _ in item_list]
 
     def getNumber(self):
         self.db.changeTable(self.raw_proxy_queue)
@@ -101,4 +105,4 @@ class ProxyManager(object):
 
 if __name__ == '__main__':
     pp = ProxyManager()
-    pp.refresh()
+    pp.fetch()

Manager/__init__.py

@@ -10,4 +10,6 @@
                    2016/12/3: 
 -------------------------------------------------
 """
-__author__ = 'JHao'
+__author__ = 'JHao'
+
+from Manager.ProxyManager import ProxyManager

ProxyHelper/Proxy.py

@@ -12,28 +12,39 @@
 """
 __author__ = 'JHao'
 
+import json
+
 
 class Proxy(object):
 
-    def __init__(self, proxy):
-        if isinstance(proxy, str):
-            self._proxy = proxy
-            self._fail_count = 0
-            self._region = ""
-            self._type = ""
-            self._last_status = ""
-            self._last_time = ""
-
-        elif isinstance(proxy, dict):
-            self._proxy = proxy.get("proxy")
-            self._fail_count = proxy.get("fail_count")
-            self._region = proxy.get("region")
-            self._type = proxy.get("type")
-            self._last_status = proxy.get("last_status")
-            self._last_time = proxy.get("last_time")
-
-        else:
-            raise TypeError("proxy arg invalid")
+    def __init__(self, proxy, fail_count=0, region="", proxy_type="",
+                 source="", check_count=0, last_status="", last_time=""):
+        self._proxy = proxy
+        self._fail_count = fail_count
+        self._region = region
+        self._type = proxy_type
+        self._source = source
+        self._check_count = check_count
+        self._last_status = last_status
+        self._last_time = last_time
+
+    @classmethod
+    def newProxyFromJson(cls, proxy_json):
+        """
+        根据proxy属性json创建Proxy实例
+        :param proxy_json:
+        :return:
+        """
+        proxy_dict = json.loads(proxy_json)
+        return cls(proxy=proxy_dict.get("proxy", ""),
+                   fail_count=proxy_dict.get("fail_count", 0),
+                   region=proxy_dict.get("region", ""),
+                   proxy_type=proxy_dict.get("type", ""),
+                   source=proxy_dict.get("source", ""),
+                   check_count=proxy_dict.get("check_count", 0),
+                   last_status=proxy_dict.get("last_status", ""),
+                   last_time=proxy_dict.get("last_time", "")
+                   )
 
     @property
     def proxy(self):
@@ -55,9 +66,19 @@ class Proxy(object):
         """ 透明/匿名/高匿 """
         return self._type
 
+    @property
+    def source(self):
+        """ 代理来源 """
+        return self._source
+
+    @property
+    def check_count(self):
+        """ 代理检测次数 """
+        return self._check_count
+
     @property
     def last_status(self):
-        """ 最后一次检测结果 """
+        """ 最后一次检测结果  1 -> 可用; 0 -> 不可用"""
         return self._last_status
 
     @property
@@ -65,6 +86,23 @@ class Proxy(object):
         """ 最后一次检测时间 """
         return self._last_time
 
+    @property
+    def info_dict(self):
+        """ 属性字典 """
+        return {"proxy": self._proxy,
+                "fail_count": self._fail_count,
+                "region": self._region,
+                "type": self._type,
+                "source": self._source,
+                "check_count": self.check_count,
+                "last_status": self.last_status,
+                "last_time": self.last_time}
+
+    @property
+    def info_json(self):
+        """ 属性json格式 """
+        return json.dumps(self.info_dict, ensure_ascii=False)
+
     # --- proxy method ---
     @fail_count.setter
     def fail_count(self, value):
@@ -78,6 +116,14 @@ class Proxy(object):
     def type(self, value):
         self._type = value
 
+    @source.setter
+    def source(self, value):
+        self._source = value
+
+    @check_count.setter
+    def check_count(self, value):
+        self._check_count = value
+
     @last_status.setter
     def last_status(self, value):
         self._last_status = value
@@ -85,20 +131,3 @@ class Proxy(object):
     @last_time.setter
     def last_time(self, value):
         self._last_time = value
-
-
-def proxy2Json(proxy):
-    return {"proxy": proxy.proxy,
-            "fail_count": proxy.fail_count,
-            "region": proxy.region,
-            "type": proxy.type,
-            "last_status": proxy.last_status,
-            "last_time": proxy.last_time}
-
-
-if __name__ == '__main__':
-    p = Proxy("127.0.0.1:8080")
-
-    import json
-
-    print(json.dumps(p, default=proxy2Json))

ProxyHelper/ProxyHelper.py

@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+"""
+-------------------------------------------------
+   File Name：     ProxyHelper
+   Description :
+   Author :        JHao
+   date：          2019/8/8
+-------------------------------------------------
+   Change Activity:
+                   2019/8/8:
+-------------------------------------------------
+"""
+__author__ = 'JHao'
+
+from ProxyHelper import Proxy
+from Util import validUsefulProxy
+
+from datetime import datetime
+
+
+def checkProxyUseful(proxy_obj):
+    """
+    检测代理是否可用
+    :param proxy_obj: Proxy object
+    :return: Proxy object, status
+    """
+
+    if validUsefulProxy(proxy_obj.proxy):
+        # 检测通过 更新proxy属性
+        proxy_obj.check_count += 1
+        proxy_obj.last_status = 1
+        proxy_obj.last_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        if proxy_obj.fail_count > 0:
+            proxy_obj.fail_count -= 1
+        return proxy_obj, True
+    else:
+        proxy_obj.check_count += 1
+        proxy_obj.last_status = 0
+        proxy_obj.last_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
+        proxy_obj.fail_count += 1
+        return proxy_obj, False
+

ProxyHelper/__init__.py

@@ -10,4 +10,7 @@
                    2019/7/11:
 -------------------------------------------------
 """
-__author__ = 'JHao'
+__author__ = 'JHao'
+
+from ProxyHelper.Proxy import Proxy
+from ProxyHelper.ProxyHelper import checkProxyUseful

Schedule/ProxyCheck.py

@@ -1,64 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
--------------------------------------------------
-   File Name：     ProxyCheck
-   Description :   多线程验证useful_proxy
-   Author :        J_hao
-   date：          2017/9/26
--------------------------------------------------
-   Change Activity:
-                   2017/9/26: 多线程验证useful_proxy
--------------------------------------------------
-"""
-__author__ = 'J_hao'
-
-import sys
-from threading import Thread
-
-
-try:
-    from Queue import Empty  # py3
-except:
-    from queue import Empty  # py2
-
-sys.path.append('../')
-
-from Util.utilFunction import validUsefulProxy
-from Manager.ProxyManager import ProxyManager
-from Util.LogHandler import LogHandler
-
-FAIL_COUNT = 1  # 校验失败次数， 超过次数删除代理
-
-
-class ProxyCheck(ProxyManager, Thread):
-    def __init__(self, queue, item_dict):
-        ProxyManager.__init__(self)
-        Thread.__init__(self)
-        self.log = LogHandler('proxy_check', file=False)  # 多线程同时写一个日志文件会有问题
-        self.queue = queue
-        self.item_dict = item_dict
-
-    def run(self):
-        self.db.changeTable(self.useful_proxy_queue)
-        while True:
-            try:
-                proxy = self.queue.get(block=False)
-            except Empty:
-                break
-            count = self.item_dict[proxy]
-            if validUsefulProxy(proxy):
-                # 验证通过计数器减1
-                if count and int(count) > 0:
-                    self.db.put(proxy, num=int(count) - 1)
-                else:
-                    pass
-                self.log.info('ProxyCheck: {} validation pass'.format(proxy))
-            else:
-                self.log.info('ProxyCheck: {} validation fail'.format(proxy))
-                if count and int(count) + 1 >= FAIL_COUNT:
-                    self.log.info('ProxyCheck: {} fail too many, delete!'.format(proxy))
-                    self.db.delete(proxy)
-                else:
-                    self.db.put(proxy, num=int(count) + 1)
-            self.queue.task_done()
-

Schedule/ProxyRefreshSchedule.py

@@ -1,111 +0,0 @@
-# -*- coding: utf-8 -*-
-# !/usr/bin/env python
-"""
--------------------------------------------------
-   File Name：     ProxyRefreshSchedule.py
-   Description :  代理定时刷新
-   Author :       JHao
-   date：          2016/12/4
--------------------------------------------------
-   Change Activity:
-                   2016/12/4: 代理定时刷新
-                   2017/03/06: 使用LogHandler添加日志
-                   2017/04/26: raw_proxy_queue验证通过但useful_proxy_queue中已经存在的代理不在放入
--------------------------------------------------
-"""
-
-import sys
-import time
-import logging
-from threading import Thread
-from apscheduler.schedulers.background import BackgroundScheduler
-
-sys.path.append('../')
-
-from Util.utilFunction import validUsefulProxy
-from Manager.ProxyManager import ProxyManager
-from Util.LogHandler import LogHandler
-
-__author__ = 'JHao'
-
-logging.basicConfig()
-
-
-class ProxyRefreshSchedule(ProxyManager):
-    """
-    代理定时刷新
-    """
-
-    def __init__(self):
-        ProxyManager.__init__(self)
-        self.log = LogHandler('refresh_schedule')
-
-    def validProxy(self):
-        """
-        验证raw_proxy_queue中的代理, 将可用的代理放入useful_proxy_queue
-        :return:
-        """
-        self.db.changeTable(self.raw_proxy_queue)
-        raw_proxy_item = self.db.pop()
-        self.log.info('ProxyRefreshSchedule: %s start validProxy' % time.ctime())
-        # 计算剩余代理，用来减少重复计算
-        remaining_proxies = self.getAll()
-        while raw_proxy_item:
-            raw_proxy = raw_proxy_item.get('proxy')
-            if isinstance(raw_proxy, bytes):
-                # 兼容Py3
-                raw_proxy = raw_proxy.decode('utf8')
-
-            if (raw_proxy not in remaining_proxies) and validUsefulProxy(raw_proxy):
-                self.db.changeTable(self.useful_proxy_queue)
-                self.db.put(raw_proxy)
-                self.log.info('ProxyRefreshSchedule: %s validation pass' % raw_proxy)
-            else:
-                self.log.info('ProxyRefreshSchedule: %s validation fail' % raw_proxy)
-            self.db.changeTable(self.raw_proxy_queue)
-            raw_proxy_item = self.db.pop()
-            remaining_proxies = self.getAll()
-        self.log.info('ProxyRefreshSchedule: %s validProxy complete' % time.ctime())
-
-
-def refreshPool():
-    pp = ProxyRefreshSchedule()
-    pp.validProxy()
-
-
-def batchRefresh(process_num=30):
-    # 检验新代理
-    pl = []
-    for num in range(process_num):
-        proc = Thread(target=refreshPool, args=())
-        pl.append(proc)
-
-    for num in range(process_num):
-        pl[num].daemon = True
-        pl[num].start()
-
-    for num in range(process_num):
-        pl[num].join()
-
-
-def fetchAll():
-    p = ProxyRefreshSchedule()
-    # 获取新代理
-    p.refresh()
-
-
-def run():
-    scheduler = BackgroundScheduler()
-    # 不用太快, 网站更新速度比较慢, 太快会加大验证压力, 导致raw_proxy积压
-    scheduler.add_job(fetchAll,  'interval', minutes=10, id="fetch_proxy")
-    scheduler.add_job(batchRefresh, "interval", minutes=1)  # 每分钟检查一次
-    scheduler.start()
-
-    fetchAll()
-
-    while True:
-        time.sleep(3)
-
-
-if __name__ == '__main__':
-    run()

Schedule/ProxyScheduler.py

@@ -0,0 +1,54 @@
+# -*- coding: utf-8 -*-
+"""
+-------------------------------------------------
+   File Name：     ProxyScheduler
+   Description :
+   Author :        JHao
+   date：          2019/8/5
+-------------------------------------------------
+   Change Activity:
+                   2019/8/5: ProxyScheduler
+-------------------------------------------------
+"""
+__author__ = 'JHao'
+
+import sys
+from apscheduler.schedulers.blocking import BlockingScheduler
+
+sys.path.append('../')
+
+from Schedule import doRawProxyCheck, doUsefulProxyCheck
+from Manager import ProxyManager
+from Util import LogHandler
+
+
+class DoFetchProxy(ProxyManager):
+    """ fetch proxy"""
+
+    def __init__(self):
+        ProxyManager.__init__(self)
+        self.log = LogHandler('fetch_proxy')
+
+    def main(self):
+        self.log.info("start fetch proxy")
+        self.fetch()
+        self.log.info("finish fetch proxy")
+
+
+def rawProxyScheduler():
+    DoFetchProxy().main()
+    doRawProxyCheck()
+
+
+def usefulProxyScheduler():
+    doUsefulProxyCheck()
+
+
+if __name__ == '__main__':
+    scheduler_log = LogHandler("scheduler_log")
+    scheduler = BlockingScheduler(logger=scheduler_log)
+
+    scheduler.add_job(rawProxyScheduler, 'interval', minutes=5, id="raw_proxy_check", name="raw_proxy定时采集")
+    scheduler.add_job(usefulProxyScheduler, 'interval', minutes=1, id="useful_proxy_check", name="useful_proxy定时检查")
+
+    scheduler.start()

Schedule/ProxyValidSchedule.py

@@ -1,77 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
--------------------------------------------------
-   File Name：     ProxyValidSchedule.py
-   Description :  验证useful_proxy_queue中的代理,将不可用的移出
-   Author :       JHao
-   date：          2017/3/31
--------------------------------------------------
-   Change Activity:
-                   2017/3/31: 验证useful_proxy_queue中的代理
--------------------------------------------------
-"""
-__author__ = 'JHao'
-
-import sys
-import time
-
-try:
-    from Queue import Queue  # py3
-except:
-    from queue import Queue  # py2
-
-sys.path.append('../')
-
-from Schedule.ProxyCheck import ProxyCheck
-from Manager.ProxyManager import ProxyManager
-
-
-class ProxyValidSchedule(ProxyManager, object):
-    def __init__(self):
-        ProxyManager.__init__(self)
-        self.queue = Queue()
-        self.proxy_item = dict()
-
-    def __validProxy(self, threads=20):
-        """
-        验证useful_proxy代理
-        :param threads: 线程数
-        :return:
-        """
-        thread_list = list()
-        for index in range(threads):
-            thread_list.append(ProxyCheck(self.queue, self.proxy_item))
-
-        for thread in thread_list:
-            thread.daemon = True
-            thread.start()
-
-        for thread in thread_list:
-            thread.join()
-
-    def main(self):
-        self.putQueue()
-        while True:
-            if not self.queue.empty():
-                self.log.info("Start valid useful proxy")
-                self.__validProxy()
-            else:
-                self.log.info('Valid Complete! sleep 5 sec.')
-                time.sleep(5)
-                self.putQueue()
-
-    def putQueue(self):
-        self.db.changeTable(self.useful_proxy_queue)
-        self.proxy_item = self.db.getAll()
-        for item in self.proxy_item:
-            self.queue.put(item)
-
-
-def run():
-    p = ProxyValidSchedule()
-    p.main()
-
-
-if __name__ == '__main__':
-    p = ProxyValidSchedule()
-    p.main()

Schedule/RawProxyCheck.py

@@ -0,0 +1,76 @@
+# -*- coding: utf-8 -*-
+"""
+-------------------------------------------------
+   File Name：     RawProxyCheck
+   Description :   check raw_proxy to useful
+   Author :        JHao
+   date：          2019/8/6
+-------------------------------------------------
+   Change Activity:
+                   2019/8/6: check raw_proxy to useful
+-------------------------------------------------
+"""
+__author__ = 'JHao'
+
+from threading import Thread
+
+try:
+    from Queue import Empty, Queue  # py2
+except:
+    from queue import Empty, Queue  # py3
+
+from Util import LogHandler
+from Manager import ProxyManager
+from ProxyHelper import Proxy, checkProxyUseful
+
+
+class RawProxyCheck(ProxyManager, Thread):
+    def __init__(self, queue, thread_name):
+        ProxyManager.__init__(self)
+        Thread.__init__(self, name=thread_name)
+        self.log = LogHandler('raw_proxy_check')
+        self.queue = queue
+
+    def run(self):
+        self.log.info("RawProxyCheck - {}  : start".format(self.name))
+        self.db.changeTable(self.useful_proxy_queue)
+        while True:
+            try:
+                proxy_json = self.queue.get(block=False)
+            except Empty:
+                self.log.info("RawProxyCheck - {}  : exit".format(self.name))
+                break
+
+            proxy_obj = Proxy.newProxyFromJson(proxy_json)
+
+            proxy_obj, status = checkProxyUseful(proxy_obj)
+            if status:
+                self.db.put(proxy_obj)
+                self.log.info('RawProxyCheck - {}  : {} validation pass'.format(self.name, proxy_obj.proxy.ljust(20)))
+            else:
+                self.log.info('RawProxyCheck - {}  : {} validation fail'.format(self.name, proxy_obj.proxy.ljust(20)))
+            self.queue.task_done()
+
+
+def doRawProxyCheck():
+    proxy_queue = Queue()
+
+    pm = ProxyManager()
+    pm.db.changeTable(pm.raw_proxy_queue)
+    for _proxy in pm.db.getAll():
+        proxy_queue.put(_proxy)
+    pm.db.clear()
+
+    thread_list = list()
+    for index in range(20):
+        thread_list.append(RawProxyCheck(proxy_queue, "thread_%s" % index))
+
+    for thread in thread_list:
+        thread.start()
+
+    for thread in thread_list:
+        thread.join()
+
+
+if __name__ == '__main__':
+    doRawProxyCheck()

Schedule/UsefulProxyCheck.py

@@ -0,0 +1,83 @@
+# -*- coding: utf-8 -*-
+"""
+-------------------------------------------------
+   File Name：     UsefulProxyCheck
+   Description :   check useful proxy
+   Author :        JHao
+   date：          2019/8/7
+-------------------------------------------------
+   Change Activity:
+                   2019/8/7: check useful proxy
+-------------------------------------------------
+"""
+__author__ = 'JHao'
+
+from threading import Thread
+
+try:
+    from Queue import Queue, Empty  # py2
+except:
+    from queue import Queue, Empty  # py3
+
+from Util import LogHandler
+from Manager import ProxyManager
+from ProxyHelper import checkProxyUseful, Proxy
+
+FAIL_COUNT = 0
+
+
+class UsefulProxyCheck(ProxyManager, Thread):
+    def __init__(self, queue, thread_name):
+        ProxyManager.__init__(self)
+        Thread.__init__(self, name=thread_name)
+
+        self.queue = queue
+        self.log = LogHandler('useful_proxy_check')
+
+    def run(self):
+        self.log.info("UsefulProxyCheck - {}  : start".format(self.name))
+        self.db.changeTable(self.useful_proxy_queue)
+        while True:
+            try:
+                proxy_str = self.queue.get(block=False)
+            except Empty:
+                self.log.info("UsefulProxyCheck - {}  : exit".format(self.name))
+                break
+
+            proxy_obj = Proxy.newProxyFromJson(proxy_str)
+            proxy_obj, status = checkProxyUseful(proxy_obj)
+            if status or proxy_obj.fail_count < FAIL_COUNT:
+                if self.db.exists(proxy_obj.proxy):
+                    self.log.info('UsefulProxyCheck - {}  : {} validation exists'.format(self.name,
+                                                                                         proxy_obj.proxy.ljust(20)))
+                self.db.put(proxy_obj)
+                self.log.info('UsefulProxyCheck - {}  : {} validation pass'.format(self.name,
+                                                                                   proxy_obj.proxy.ljust(20)))
+            else:
+                self.log.info('UsefulProxyCheck - {}  : {} validation fail'.format(self.name,
+                                                                                   proxy_obj.proxy.ljust(20)))
+                self.db.delete(proxy_obj.proxy)
+            self.queue.task_done()
+
+
+def doUsefulProxyCheck():
+    proxy_queue = Queue()
+
+    pm = ProxyManager()
+    pm.db.changeTable(pm.useful_proxy_queue)
+    for _proxy in pm.db.getAll():
+        proxy_queue.put(_proxy)
+
+    thread_list = list()
+    for index in range(10):
+        thread_list.append(UsefulProxyCheck(proxy_queue, "thread_%s" % index))
+
+    for thread in thread_list:
+        thread.start()
+
+    for thread in thread_list:
+        thread.join()
+
+
+if __name__ == '__main__':
+    doUsefulProxyCheck()

Schedule/__init__.py

@@ -10,4 +10,7 @@
                    2016/12/3: 
 -------------------------------------------------
 """
-__author__ = 'JHao'
+__author__ = 'JHao'
+
+from Schedule.RawProxyCheck import doRawProxyCheck
+from Schedule.UsefulProxyCheck import doUsefulProxyCheck

Test/testProxyClass.py

@@ -0,0 +1,33 @@
+# -*- coding: utf-8 -*-
+"""
+-------------------------------------------------
+   File Name：     testProxyClass
+   Description :
+   Author :        JHao
+   date：          2019/8/8
+-------------------------------------------------
+   Change Activity:
+                   2019/8/8:
+-------------------------------------------------
+"""
+__author__ = 'JHao'
+
+import json
+from ProxyHelper import Proxy
+
+
+def testProxyClass():
+    proxy = Proxy("127.0.0.1:8080")
+
+    print(proxy.info_dict)
+
+    proxy.source = "test"
+
+    proxy_str = json.dumps(proxy.info_dict, ensure_ascii=False)
+
+    print(proxy_str)
+
+    print(Proxy.newProxyFromJson(proxy_str).info_dict)
+
+
+testProxyClass()

Util/EnvUtil.py

@@ -1,17 +0,0 @@
-# -*- coding: utf-8 -*-
-"""
--------------------------------------------------
-   File Name：     EnvUtil
-   Description :   环境相关
-   Author :        J_hao
-   date：          2017/9/18
--------------------------------------------------
-   Change Activity:
-                   2017/9/18: 区分Python版本
--------------------------------------------------
-"""
-__author__ = 'J_hao'
-
-import sys
-
-PY3 = sys.version_info >= (3,)

Util/__init__.py

@@ -9,4 +9,8 @@
    Change Activity:
                    2016/11/25: 
 -------------------------------------------------
-"""
+"""
+
+from Util.utilFunction import validUsefulProxy
+from Util.LogHandler import LogHandler
+from Util.utilClass import Singleton