这是我的程序的源代码,但对我来说,输出看起来有点有趣,但它有效,所以我仍然对它感到满意。但如果我没有正确使用线程,我想知道这样我就可以得到它 以正确的方式工作。
import os
import time
import threading
import urllib.request
max_threads = 10
RED = '\033[31m'
GREEN = '\033[32m'
ESC = '\033[0m'
def check(proxy):
proxy_support = urllib.request.ProxyHandler({'https':proxy})
opener = urllib.request.build_opener(proxy_support)
urllib.request.install_opener(opener)
print(end='\r' + time.strftime('[%H:%M:%S]')+" ~ Trying => " +proxy)
try:
urllib.request.urlopen("https://www.google.com", timeout=5)
time.sleep(1)
print(end='\r'+time.strftime('[%H:%M:%S]')+" ~"+GREEN+" Good [!] "+ESC +proxy)
time.sleep(1)
with open("CheckedProxies.txt", "a") as appe:
appe.write(proxy.replace("\n","") + "\n")
except:
time.sleep(1)
print(end='\r'+time.strftime('[%H:%M:%S]')+" ~"+RED+" Bad [!] "+ESC +proxy)
time.sleep(1)
pass
try:
proxies = open("/home/zion/Desktop/proxies.txt", "r").readlines()
except:
print("File Empty Exiting!")
exit()
if proxies == "":
print("File Empty, Enter Proxies In proxies.txt File")
newtxt = open("CheckedProxies.txt","w")
print("Loading "+ str(len(proxies)) +" Proxies From Text File[!]")
time.sleep(3)
for proxy in proxies:
threading.Thread(target=check, args=(proxy,)).start()
while threading.activeCount() >= max_threads:
time.sleep(1)
os.exit()
这是我的程序的输出......
[02:28:02] ~ Trying => 1.0.135.34:8080
[02:28:02] ~ Trying => 1.10.236.214:8080
[02:28:02] ~ Trying => 103.122.255.18:8080
[02:28:02] ~ Trying => 101.231.104.82:80
[02:28:02] ~ Trying => 102.176.160.109:8080
[02:28:02] ~ Trying => 1.179.144.181:8080
[02:28:02] ~ Trying => 103.10.228.221:8080
[02:28:02] ~ Trying => 101.255.40.38:47638
[02:28:02] ~ Trying => 101.108.110.95:3128
[02:28:03] ~ Bad [!] 1.0.135.34:8080
[02:28:03] ~ Bad [!] 101.255.40.38:47638
[02:28:03] ~ Bad [!] 103.10.228.221:8080
[02:28:03] ~ Bad [!] 1.10.236.214:8080
[02:28:03] ~ Bad [!] 101.231.104.82:80
[02:28:05] ~ Trying => 103.215.200.125:8080
[02:28:05] ~ Trying => 101.108.102.231:8080
我认为它会更像这样
[02:28:02] ~ Trying => 127.0.0.1:8080
[02:28:03] ~ Bad [!] 127.0.0.1:80
[02:28:02] ~ Trying => 127.0.0.1:8080
[02:28:03] ~ Bad [!] 127.0.0.1:80
[02:28:02] ~ Trying => 127.0.0.1:47638
[02:28:03] ~ Bad [!] 127.0.0.1:80
[02:28:02] ~ Trying => 127.0.0.1:3128
最佳答案
我不知道为什么在使用多线程时期望同步执行。
为了让这段代码变得更好:
- 使用日志记录而不是打印 strftime
- 使用线程池代替带有 sleep 的 while 循环
- 使用 requests 而不是 urllib
- 使用 f 字符串格式而不是串联
- 使用 with 打开文件以确保它们已正确关闭
import logging
from concurrent.futures.thread import ThreadPoolExecutor
from typing import Tuple
import requests
def is_proxy_ok(proxy: str) -> bool:
try:
logging.info(f"Trying {proxy}")
proxies = {"https": proxy, "http": proxy}
response = requests.get("https://www.google.com", proxies=proxies, timeout=5)
response.raise_for_status()
logging.info(f"OK {proxy}")
return True
except requests.RequestException:
logging.info(f"Bad {proxy}")
return False
def check_proxy(proxy: str) -> Tuple[bool, str]:
return is_proxy_ok(proxy), proxy
if __name__ == "__main__":
logging.basicConfig(format="%(asctime)-15s - %(message)s", level=logging.INFO)
with open("/home/zion/Desktop/proxies.txt") as proxy_file:
proxies = [line.strip() for line in proxy_file]
print(f"Loading {len(proxies)} Proxies From Text File[!]")
working_proxies = []
with ThreadPoolExecutor(max_workers=25) as executor:
for is_working, proxy in executor.map(check_proxy, proxies):
if is_working:
working_proxies.append(proxy)
with open("working_proxies.txt", "w") as out_file:
print("\n".join(working_proxies), file=out_file)
关于python3 线程输出帮助。这是正确的输出吗?,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/59225244/