Merge branch 'master' of github.com:DevAlone/proxy_py

DevAlone · DevAlone · commit 2dd56b5bd42d · 2018-08-30T23:00:46.000+03:00
diff --git a/checkers/base_checker.py b/checkers/base_checker.py
@@ -35,6 +35,7 @@ def set_attr_if_is_not_none(attribute_name, first_obj, second_obj):
 
 
 class BaseChecker:
+    # TODO: rewrite using HttpClient
     aiohttp_connector = None
 
     def __init__(self, url=None, request_type="GET", timeout=None):
diff --git a/collectors/web/cn/89ip/collector.py b/collectors/web/cn/89ip/collector.py
@@ -0,0 +1,23 @@
+from collectors.abstract_collector import AbstractCollector
+import http_client
+
+
+class Collector(AbstractCollector):
+    __collector__ = True
+
+    def __init__(self):
+        super(Collector, self).__init__()
+        self.processing_period = 30 * 60  # 30 minutes
+        '''
+        floating period means proxy_py will be changing 
+        period to not make extra requests and handle 
+        new proxies in time, you don't need to change 
+        it in most cases
+        '''
+        # self.floating_processing_period = False
+
+    async def collect(self):
+        url = 'http://www.89ip.cn/tqdl.html?num=9999&address=&kill_address=&port=&kill_port=&isp='
+        html = await http_client.get_text(url)
+
+        return []
diff --git a/docs/source/guides/how_to_create_collector.rst b/docs/source/guides/how_to_create_collector.rst
@@ -1,14 +1,21 @@
 proxy_py How to create collector
 ================================
 
-Collector is a class which is used to parse proxies from web page or another source. All collectors are inherited from `collectors.abstract_collector.AbstractCollector`, also there is `collectors.pages_collector.PagesCollector` which is used for paginated sources. It's better to understand through examples.
+Collector is a class which is used to parse proxies from web page or another source.
+All collectors are inherited from `collectors.abstract_collector.AbstractCollector`,
+also there is `collectors.pages_collector.PagesCollector` which is used for paginated sources.
+It's better to understand through examples.
 
 Simple collector
 ****************
 
-Let's start with the simplest collector we can imagine, it will be from the page http://www.89ip.cn/ti.html as you can see, it sends form as GET request to this url http://www.89ip.cn/tqdl.html?num=9999&address=&kill_address=&port=&kill_port=&isp= 
+Let's start with the simplest collector we can imagine,
+it will be from the page http://www.89ip.cn/ti.html as you can see,
+it sends form as GET request to this url
+http://www.89ip.cn/tqdl.html?num=9999&address=&kill_address=&port=&kill_port=&isp=
 
-Firstly we can try to check that these proxies are really good. Just copy and paste list of proxies to file say /tmp/proxies and run this command inside virtual environment
+Firstly we can try to check that these proxies are really good.
+Just copy and paste list of proxies to file say /tmp/proxies and run this command inside virtual environment
 
 .. code-block:: bash
 
@@ -18,10 +25,45 @@ You're gonna get something like this:
 
 `++++++++++++++++++++++-+++++-+++++++++++++++++++++++++++-++++++-++++-+++++++++++++++++++++++++++++++--+++++++-+++++++-++-+-+++-+++++++++-+++++++++++++++++++++--++--+-++++++++++++++++-+++--+++-+-+++++++++++++++++--++++++++++++-+++++-+++-++++++++-+++++-+-+++++++-++-+--++++-+++-++++++++++-++++--+++++++-+++++++-++--+++++-+-+++++++++++++++++++++-++-+++-+++--++++--+++-+++++++-+++++++-+++++++++++++++---+++++-+++++++++-+++++-+-++++++++++++-+--+++--+-+-+-++-+++++-+++--++++++-+++++++++++--+-+++-+-++++--+++++--+++++++++-+-+-++++-+-++++++++++++++-++-++++++--+--++++-+-++--++--+++++-++-+++-++++--++--+---------+--+--++--------+++-++-+--++++++++++++++++-+++++++++-+++++++--+--+--+-+-+++---++------------------+--+----------+-+-+--++-+----------+-------+--+------+----+-+--+--++----+--+-++++++-++-+++`
 
-+ means working proxy with at leat one protocol, - means not working, the result above is perfect, so many good proxies.
++ means working proxy with at least one protocol,
+- means not working, the result above is perfect, so many good proxies.
 
 Note: working means proxy respond with timeout set in settings, if you increase it you'll get more proxies.
 
+Alright, let's code it!
+
+We need to place our collector inside `collectors/web/` directory using reversed domain path,
+it will be `collectors/web/cn/89ip/collector.py`
+
+To make class be collector we need to declare variable `__collector__`
+
+Note: name of file and name of class don't make sense,
+you can declare as many files and classes in each file per domain as you want
+
+.. code-block:: python
+
+    from collectors.abstract_collector import AbstractCollector
+
+
+    class Collector(AbstractCollector):
+        __collector__ = True
+
+We can override default processing period in constructor like this:
+
+.. code-block:: python
+
+    def __init__(self):
+        super(Collector, self).__init__()
+        self.processing_period = 30 * 60  # 30 minutes
+        '''
+        floating period means proxy_py will be changing
+        period to not make extra requests and handle
+        new proxies in time, you don't need to change
+        it in most cases
+        '''
+        # self.floating_processing_period = False
+
+
 Paginated collector
 *******************
 
diff --git a/http_client.py b/http_client.py
@@ -0,0 +1,100 @@
+import json
+
+from proxy_py import settings
+from fake_useragent import UserAgent
+from aiosocks.connector import ProxyConnector, ProxyClientRequest
+
+import aiohttp
+
+
+class HttpClientResult:
+    text = None
+    aiohttp_response = None
+
+    @staticmethod
+    async def make(aiohttp_response):
+        obj = HttpClientResult()
+        obj.aiohttp_response = aiohttp_response
+        obj.text = await obj.aiohttp_response.text()
+
+        return obj
+
+    def as_text(self):
+        return self.text
+
+    def as_json(self):
+        return json.loads(self.text)
+
+
+# TODO: complete cookies saving
+class HttpClient:
+    """
+    Simple class for making http requests,
+    user-agent is set to random one in constructor
+    """
+    _aiohttp_connector = None
+
+    def __init__(self):
+        self.user_agent = UserAgent().random
+        self.timeout = 60
+        if HttpClient._aiohttp_connector is None:
+            HttpClient._aiohttp_connector = ProxyConnector(
+                remote_resolve=True,
+                limit=settings.NUMBER_OF_SIMULTANEOUS_REQUESTS,
+                limit_per_host=settings.NUMBER_OF_SIMULTANEOUS_REQUESTS_PER_HOST,
+            )
+        self.proxy_address = None
+
+    async def get(self, url):
+        """
+        send HTTP GET request
+
+        :param url:
+        :return:
+        """
+        return await self.request('GET', url, None)
+
+    async def post(self, url, data):
+        """
+        send HTTP POST request
+
+        :param url:
+        :param data:
+        :return:
+        """
+        return await self.request('POST', url, data)
+
+    async def request(self, method, url, data) -> HttpClientResult:
+        headers = {
+            'User-Agent': self.user_agent,
+        }
+
+        async with aiohttp.ClientSession(connector=HttpClient._aiohttp_connector,
+                                         connector_owner=False,
+                                         request_class=ProxyClientRequest
+                                         ) as session:
+            async with session.request(method,
+                                       url=url,
+                                       data=data,
+                                       proxy=self.proxy_address,
+                                       timeout=self.timeout,
+                                       headers=headers) as response:
+                return await HttpClientResult.make(response)
+
+    @staticmethod
+    async def clean():
+        HttpClient._aiohttp_connector.close()
+
+
+async def get_text(url):
+    """
+    fast method for sending get response without creating extra objects
+
+    :param url:
+    :return:
+    """
+    return (await HttpClient().get(url)).as_text()
+
+
+async def get_json(url):
+    return (await HttpClient().get(url)).as_json()
diff --git a/processor.py b/processor.py
@@ -68,6 +68,7 @@ def __init__(self):
 
         self.queue = asyncio.Queue(maxsize=settings.PROXY_QUEUE_SIZE)
         self.proxies_semaphore = asyncio.BoundedSemaphore(settings.NUMBER_OF_CONCURRENT_TASKS)
+        self.good_proxies_are_processed = False
 
     async def worker(self):
         await asyncio.gather(*[
@@ -93,6 +94,13 @@ async def consumer(self):
                 await asyncio.sleep(settings.SLEEP_AFTER_ERROR_PERIOD)
 
     async def producer(self):
+        while True:
+            await asyncio.gather(*[
+                self.process_proxies(),
+                self.process_collectors(),
+            ])
+
+    async def process_proxies(self):
         while True:
             await asyncio.sleep(0.00001)
             try:
@@ -103,28 +111,15 @@ async def producer(self):
                         Proxy.last_check_time < time.time() - Proxy.checking_period,
                         ).order_by(Proxy.last_check_time).limit(settings.NUMBER_OF_CONCURRENT_TASKS)
                 )
+                if proxies:
+                    self.good_proxies_are_processed = False
 
                 await self.add_proxies_to_queue(proxies)
 
-                if len(proxies) > settings.NUMBER_OF_CONCURRENT_TASKS / 2:
+                if proxies:
                     continue
 
-                # check collectors
-                collector_states = await db.execute(
-                    CollectorState.select().where(
-                        CollectorState.last_processing_time < time.time() - CollectorState.processing_period
-                    ).order_by(peewee.fn.Random()).
-                    limit(settings.NUMBER_OF_CONCURRENT_COLLECTORS)
-                )
-
-                tasks = [
-                    self.process_collector_of_state(collector_state)
-                    for collector_state in collector_states
-                ]
-
-                if tasks:
-                    await asyncio.gather(*tasks)
-                    continue
+                self.good_proxies_are_processed = True
 
                 # check bad proxies
                 proxies = await db.execute(
@@ -159,6 +154,31 @@ async def producer(self):
 
                 await asyncio.sleep(settings.SLEEP_AFTER_ERROR_PERIOD)
 
+    async def process_collectors(self):
+        while True:
+            try:
+                await asyncio.sleep(0.000001)
+
+                # check collectors
+                collector_states = await db.execute(
+                    CollectorState.select().where(
+                        CollectorState.last_processing_time < time.time() - CollectorState.processing_period
+                    ).order_by(peewee.fn.Random()).limit(settings.NUMBER_OF_CONCURRENT_COLLECTORS)
+                )
+
+                await asyncio.gather(*[
+                    self.process_collector_of_state(collector_state)
+                    for collector_state in collector_states
+                ])
+            except KeyboardInterrupt as ex:
+                raise ex
+            except BaseException as ex:
+                self.collectors_logger.exception(ex)
+                if settings.DEBUG:
+                    raise ex
+
+                await asyncio.sleep(settings.SLEEP_AFTER_ERROR_PERIOD)
+
     def is_queue_free(self):
         return self.queue.qsize() < settings.NUMBER_OF_CONCURRENT_TASKS
 
@@ -183,15 +203,15 @@ async def process_collector_of_state(self, collector_state):
             )
             proxies = await collector._collect()
 
-            if not proxies:
-                self.collectors_logger.warning(
-                    "got 0 proxies from collector of type \"{}\"".format(type(collector))
-                )
-            else:
+            if proxies:
                 self.logger.debug(
                     "got {} proxies from collector of type \"{}\"".format(len(proxies), type(collector))
                 )
                 await self.process_raw_proxies(proxies, collector_state.id)
+            else:
+                self.collectors_logger.warning(
+                    "got 0 proxies from collector of type \"{}\"".format(type(collector))
+                )
         except KeyboardInterrupt as ex:
             raise ex
         except BaseException as ex:
@@ -208,6 +228,7 @@ async def process_raw_proxies(self, proxies, collector_id):
         tasks = []
 
         for proxy in proxies:
+            # TODO: refactor it
             tasks.append(self.process_raw_proxy(proxy, collector_id))
             if len(tasks) > settings.NUMBER_OF_CONCURRENT_TASKS:
                 await asyncio.gather(*tasks)
@@ -261,6 +282,9 @@ async def process_raw_proxy(self, proxy, collector_id):
                 pass
 
             for raw_protocol in range(len(Proxy.PROTOCOLS)):
+                while not self.good_proxies_are_processed:
+                    await asyncio.sleep(0.01)
+
                 await self.queue.put((
                     raw_protocol,
                     auth_data,
diff --git a/requirements.txt b/requirements.txt
@@ -10,3 +10,4 @@ recommonmark
 sphinx_rtd_theme
 py_mini_racer
 pytest
+pytest-asyncio
diff --git a/tests/test_http_client.py b/tests/test_http_client.py
@@ -0,0 +1,13 @@
+import http_client
+import pytest
+
+
+@pytest.mark.asyncio
+async def test_fast_methods():
+    resp = await http_client.get_json('https://ipinfo.io/json')
+    assert 'ip' in resp
+
+
+def test_saving_state():
+    # TODO: implement
+    pass
