不使用回调

oneforall/common/request.py

@@ -1,7 +1,6 @@
 # coding=utf-8
 
 import asyncio
-import functools
 import aiohttp
 from aiohttp import ClientSession
 from aiohttp.resolver import AsyncResolver
@@ -65,33 +64,35 @@ async def fetch(session, url, semaphore):
             return resp, text
 
 
-def request_callback(future, index, datas):
-    try:
-        resp, text = future.result()
-    except BaseException as e:
-        logger.log('DEBUG', e.args)
-        datas[index]['reason'] = str(e.args)
-        datas[index]['valid'] = 0
-    else:
-        datas[index]['reason'] = resp.reason
-        datas[index]['status'] = resp.status
-        if resp.status == 400 or resp.status >= 500:
+def deal_results(datas, results):
+    for index, result in enumerate(results):
+        if isinstance(result, Exception):
+            logger.log('DEBUG', result.args)
+            datas[index]['reason'] = str(result.args)
             datas[index]['valid'] = 0
-        else:
-            headers = resp.headers
-            banner = str({'Server': headers.get('Server'),
-                          'Via': headers.get('Via'),
-                          'X-Powered-By': headers.get('X-Powered-By')})
-            datas[index]['banner'] = banner
-            soup = BeautifulSoup(text, 'lxml')
-            title = soup.title
-            head = soup.head
-            if title:
-                datas[index]['title'] = title.text
-            elif head:
-                datas[index]['title'] = head.text
+            continue
+        if isinstance(result, tuple):
+            resp, text = result
+            datas[index]['reason'] = resp.reason
+            datas[index]['status'] = resp.status
+            if resp.status == 400 or resp.status >= 500:
+                datas[index]['valid'] = 0
             else:
-                datas[index]['title'] = text
+                headers = resp.headers
+                banner = str({'Server': headers.get('Server'),
+                              'Via': headers.get('Via'),
+                              'X-Powered-By': headers.get('X-Powered-By')})
+                datas[index]['banner'] = banner
+                soup = BeautifulSoup(text, 'lxml')
+                title = soup.title
+                head = soup.head
+                if title:
+                    datas[index]['title'] = title.text
+                elif head:
+                    datas[index]['title'] = head.text
+                else:
+                    datas[index]['title'] = text
+    return datas
 
 
 async def bulk_get_request(datas, port):
@@ -113,11 +114,11 @@ async def bulk_get_request(datas, port):
         for i, data in enumerate(new_datas):
             url = data.get('url')
             task = asyncio.ensure_future(fetch(session, url, semaphore))
-            task.add_done_callback(functools.partial(request_callback,
-                                                     index=i,
-                                                     datas=new_datas))
             tasks.append(task)
         if tasks:  # 任务列表里有任务不空时才进行解析
-            await asyncio.wait(tasks)  # 等待所有task完成
+            # 等待所有task完成 错误聚合到结果列表里
+            results = await asyncio.gather(*tasks, return_exceptions=True)
+            new_datas = deal_results(new_datas, results)
+
     logger.log('INFOR', f'完成异步进行子域的GET请求')
     return new_datas