fix

Author: DevAlone

processor.py

@@ -77,18 +77,18 @@ async def consumer(self):
         while True:
             await asyncio.sleep(0.1)
             i = 0
-            async with aiohttp.ClientSession(
-                    connector=ProxyConnector(remote_resolve=False),
-                    request_class=ProxyClientRequest
-            ) as aiohttp_proxy_check_session:
-                while not self.queue.empty() and i <= settings.CONCURRENT_TASKS_COUNT:
-                    proxy_data = self.queue.get_nowait()
-                    tasks.append(self.process_proxy(*proxy_data, aiohttp_proxy_check_session))
-                    self.queue.task_done()
-
-                if tasks:
-                    await asyncio.wait(tasks)
-                    tasks.clear()
+            # async with aiohttp.ClientSession(
+            #         connector=ProxyConnector(remote_resolve=False),
+            #         request_class=ProxyClientRequest
+            # ) as aiohttp_proxy_check_session:
+            while not self.queue.empty() and i <= settings.CONCURRENT_TASKS_COUNT:
+                proxy_data = self.queue.get_nowait()
+                tasks.append(self.process_proxy(*proxy_data))
+                self.queue.task_done()
+
+            if tasks:
+                await asyncio.wait(tasks)
+                tasks.clear()
 
     async def producer(self):
         while True:
@@ -200,7 +200,7 @@ async def process_raw_proxies(self, proxies, collector_id):
                     ))
 
     async def process_proxy(self, raw_protocol: int, auth_data: str, domain: str, port: int, collector_id: int,
-                            aiohttp_proxy_check_session):
+                            aiohttp_proxy_check_session=None):
         self.logger.debug("start processing proxy {}://{}@{}:{} with collector id {}".format(
             raw_protocol, auth_data, domain, port, collector_id))
 
@@ -253,6 +253,8 @@ async def process_proxy(self, raw_protocol: int, auth_data: str, domain: str, po
             session.rollback()
             self.logger.error("Error during processing proxy")
             self.logger.exception(ex)
+            print("Exception proxy_url = {}".format(proxy_url))
+            exit(1)
 
     @staticmethod
     def create_or_update_proxy(raw_protocol: Proxy.PROTOCOLS, auth_data, domain, port,

proxy_utils.py

@@ -1,5 +1,7 @@
 import traceback
 
+import sys
+
 from proxy_py import settings
 import async_requests
 
@@ -9,18 +11,21 @@
 import aiosocks
 import os
 
+from aiosocks.connector import ProxyConnector, ProxyClientRequest
+
 
 # TODO: add multiple checks with several sites
-async def check_proxy(proxy_url: str, session):
+async def check_proxy(proxy_url: str, session=None):
     try:
         res = await _request(
             'get',
             'https://pikagraphs.d3d.info/OK/',
+            # 'https://wtfismyip.com/text',
             proxy_url,
             settings.PROXY_CHECKING_TIMEOUT,
             session
         )
-        
+
         if res.status == 200 and res.text == "OK":
             return True
     except (aiohttp.client_exceptions.ServerDisconnectedError,
@@ -35,16 +40,22 @@ async def check_proxy(proxy_url: str, session):
         message = str(ex)
         if "file" in message:
             print(message)
-        # TODO: check to "Too many open files"
+        # TODO: check for "Too many open files"
         return False
 
+    return False
 
-async def _request(method, url, proxy_url, timeout, session : aiohttp.ClientSession):
+
+async def _request(method, url, proxy_url, timeout, session: aiohttp.ClientSession=None):
     headers = {
         'User-Agent': async_requests.get_random_user_agent()
     }
 
-    async with session.request(method, url, timeout=timeout, headers=headers, proxy=proxy_url) as response:
-        status = response.status
-        text = await response.text()
-        return async_requests.Response(status, text)
+    conn = ProxyConnector(remote_resolve=True)
+
+    with aiohttp.ClientSession(connector=conn, request_class=ProxyClientRequest) as session:
+        async with session.request(method, url, proxy=proxy_url, timeout=timeout) as response:
+            # async with session.request(method, url, timeout=timeout, headers=headers, proxy=proxy_url) as response:
+            status = response.status
+            text = await response.text()
+            return async_requests.Response(status, text)