403绕过.py

@@ -142,6 +142,8 @@ if 命令.parse_args().域名:
         except:
             print(命令.format_help())
             raise TypeError("参数格式错误!")
+    else:
+        格式化参数 = None
     绕过403限制(命令.parse_args().域名, 命令.parse_args().请求方式, 格式化参数).运行()
 elif 命令.parse_args().文件名:
     with open(命令.parse_args().文件名) as 读取数据:
@@ -149,5 +151,3 @@ elif 命令.parse_args().文件名:
             绕过403限制(i).运行()
 else:
     print(命令.format_help())
-
-

edu.cn专用脚本.py

+from 数据连接 import 读取数据, 连接数据库
+from 爬虫模板 import 爬虫常用配置
+import requests
+import datetime
+# 弄个进度条库
+from tqdm import tqdm, trange
+# 线程池
+from concurrent.futures import ThreadPoolExecutor
+
+
+当前日期 = datetime.datetime.now().date()
+# 查询语句
+# "SELECT * FROM `cn.edu数据` WHERE `id` = '1'"
+# 添加语句
+# UPDATE `cn.edu数据` SET `存活状态` = 1 WHERE `id` = 1
+# 添加日期的日期格式
+# UPDATE `网安数据记录`.`cn.edu数据` SET `存活检查日期` = 'edu无法访问网站数据(2021-10-10).txt' WHERE `id` = 1
+# 获取表长度
+# select count(*) from A
+
+
+# 存活状态 1存活, 0不存货(教育网存活,非教育网不存活为1, 反之为0)
+
+class 数据库操作:
+    def __init__(self, SQL语句=None):
+        self.数据库 = 连接数据库(游标=False)
+        self.数据库游标 = 连接数据库()
+        self.SQL语句 = SQL语句
+        self.数据库游标.execute("SELECT count(*) FROM `cn.edu数据`")
+        self.数据库长度 = self.数据库游标.fetchone()[0]
+
+    def 所有数据(self):
+        for i in trange(self.数据库长度, desc='网站存活率检测进度: '):
+            self.数据库游标.execute(f"SELECT * FROM `cn.edu数据` WHERE `id` = '{i + 1}'")
+
+            数据 = self.数据库游标.fetchone()
+            # 小记录: 这里原本返回值用的是`self.数据库游标`, 但是这样`self.数据库游标`都会指向同一个地方,会导致一些数据丢失!
+            yield 数据
+
+    def 存活网站地址(self, 后缀='/'):
+        for i in trange(self.数据库长度, desc='检测进度: '):
+            self.数据库游标.execute(f"SELECT * FROM `cn.edu数据` WHERE `id` = '{i + 1}' and `存活状态` = '1'")
+            数据 = self.数据库游标.fetchone()
+            if 数据:
+                if 数据[7] and 数据[7] != "无":
+                    url = 数据[7].rstrip("/")
+                    目标 = url + 后缀
+                    # yield {"数据库id": 数据[0], "学校名称": 数据[1]}, 数据[7]
+                    yield 目标
+                if 数据[6] and 数据[6] != "无":
+                    url = 数据[6].rstrip("/")
+                    目标 = url + 后缀
+                    # yield {"数据库id": 数据[0], "学校名称": 数据[1]}, 数据[6]
+                    yield 目标
+
+    def 存活信息保存(self, sql或id, 存活状态=1):
+        数据库游标 = self.数据库.cursor()
+        try:
+            try:
+                int(sql或id)
+                sql存储语句 = [
+                    "UPDATE `cn.edu数据` SET `存活状态` = '%s' WHERE `id` = '%s'" % (存活状态, sql或id),
+                    "UPDATE `网安数据记录`.`cn.edu数据` SET `存活检查日期` = '%s' WHERE `id` = '%s'" % (当前日期, sql或id)
+                ]
+            except:
+                if isinstance(sql或id, list):
+                    sql存储语句 = sql或id
+                else:
+                    sql存储语句 = [sql或id]
+            for i in sql存储语句:
+                数据库游标.execute(i)
+            self.数据库.commit()
+        except Exception as 异常:
+            print('[数据库]保存数据失败!', "错误信息", {repr(异常)})
+
+
+def 网站存活检测(网址):
+    try:
+        爬虫数据 = 爬虫常用配置().运行(网址)
+        if 爬虫数据["状态码"] != "200":
+            return True, {"网址": 网址, "错误信息": f"网站标题: {爬虫数据['网页标题']}", "备注": 爬虫数据.get("备注")}
+        else:
+            return False, {"网址": 网址, "错误信息": f"状态码异常: {爬虫数据['状态码']}", "备注": 爬虫数据.get("备注")}
+    except Exception as 异常:
+        return False, {"网址": 网址, "错误信息": {repr(异常)}}
+
+
+class 多线程网站存活记录:
+    # 推荐线程数范围为[cpu数量*2:cpu数量*5]
+    def __init__(self, 线程数=32):
+        self.问题网站记录 = []
+        self._线程数 = 线程数
+
+    def __信息输出(self, 网站信息):
+        正常_网站 = {"数据库id": 网站信息[0], "学校名称": 网站信息[1]}
+        数据 = []
+        if 网站信息[7] and 网站信息[7] != "无":
+            ok, 抓取信息 = 网站存活检测(网站信息[7])
+            if ok:
+                正常_网站["非教育网地址"] = 网站信息[7]
+            else:
+                数据.append(抓取信息)
+        if 网站信息[6] and 网站信息[6] != "无":
+            ok, 抓取信息 = 网站存活检测(网站信息[6])
+            if ok:
+                正常_网站["教育网地址"] = 网站信息[6]
+            else:
+                数据.append(抓取信息)
+        if ok:
+            数据库操作().存活信息保存(正常_网站['数据库id'])
+        else:
+            数据库操作().存活信息保存(正常_网站['数据库id'], 存活状态=0)
+        if 数据:
+            问题_网站 = {"数据库id": 网站信息[0], "学校名称": 网站信息[1], "问题信息": 数据}
+            self.问题网站记录.append(问题_网站)
+        return 正常_网站
+
+    def 运行(self, 保存_错误信息=False):
+        print("扫描时请关闭梯子,以免影响结果!")
+        线程池_存活测试 = ThreadPoolExecutor(max_workers=self._线程数)
+        线程池_存活测试.map(self.__信息输出, 数据库操作().所有数据())
+        print('存活扫描数据包发生完成,发现{}个失效网站!'.format(len(self.问题网站记录)))
+        if 保存_错误信息:
+            with open("edu无法访问网站数据({}).txt".format(当前日期), "a+") as 写入:
+                保存记录 = tqdm([str(i) for i in self.问题网站记录])
+                for i in 保存记录:
+                    保存记录.set_description('保存进度: ' + i)
+                    写入.write(i)
+            print('生成文件中!!!')
+        return True
+
+
+def 多线程域名访问测试(地址, 访问方式='post', 数据=None, 线程数=16):
+    # 默认使用所有存活的edu.cn网站
+    线程池_域名访问 = ThreadPoolExecutor(线程数)
+    线程池_域名访问.map(爬虫常用配置(访问方式, 数据, 网页内容=True, 数据保存=True).运行, 数据库操作().存活网站地址(地址))
+
+
+if __name__ == '__main__':
+    # 多线程网站存活记录().运行(保存_错误信息=True)
+    多线程域名访问测试("/JSON-RPC")
+    # 多线程域名访问测试()
+    pass
+
+
+
+
+