[Python] URL 存活探测

本来吧，面板中已经写了存活 URL 批量探测的功能，但是呢，因为是基于 PHP 的，会因为 PHP 内存时间而导致大量的 URL 地址探测时爆 502 错误。因此花了点时间，写了个基于 Python 的探测工具。

注意事项：

1. 脚本 11 行  D:\Desktop\url.txt 是URL 字典，必须一行一个。
2. http 的，不需要加 http:// 开头 ，https 的，需要加上 https:// 协议头，FOFA 搜出来的网站导出后的 URL 就是这样，我也是为了配合 FOFA 的结果，所以写成了这样。

效果：

代码：

import requests
import threading
import queue

success=[]
proxys = 0 # 0 关闭代理 1 启用代理
proxy = {"http": "127.0.0.1:7890"} #代理

def get_urls():
    url = []
    with open(r"D:\Desktop\url.txt") as f:
        for line in f.readlines():
            line = line.strip('\n')
            url.append(line)
    return(url)


class mythread(threading.Thread):
    def __init__(self,url_queue):
        super().__init__()
        self.url_queue=url_queue

    def run(self):
        while True:
            if not self.url_queue.empty():
                url=self.url_queue.get()#从队列里取出一个url
                self.do_work(url)#执行对应业务操作
                self.url_queue.task_done()#通知队列任务执行完成
            else:
                break

    def do_work(self,url):
        global success
        result = 'https://' in url
        if result == True:
            url = url
        else:
            url = 'http://'+url
        try:
            if proxys == 1:
                html = requests.get(url,allow_redirects=False, proxies=proxy)
                html.encoding = 'utf-8'
                html = html.headers['Content-Type']
                print("[√] %s 访问正常！"%(url))
                success.append(url+ '\n')
            else:
                html = requests.get(url,allow_redirects=False)
                html.encoding = 'utf-8'
                html = html.headers['Content-Type']
                print("[√] %s 访问正常！"%(url))
                success.append(url+ '\n')
        except Exception as e:
            print("[x] %s 无法访问！"%(url))

if __name__ == '__main__':
    url_queue=queue.Queue()#存放url的队列
    url_list=get_urls()#获取Url列表
    for url in url_list:#将所有url存入队列
        url_queue.put(url)

    for i in range(10):#创建线程,循环次数等于线程数
        t=mythread(url_queue)
        t.setDaemon(True)
        t.start()

    url_queue.join()
    success = ''.join(success)
    print('访问成功的：')
    print(success)