Ты когда пустой скрин отправляешь, какой реакции ждёшь ?
он наркоман
потому что на нем получится достичь оптимальной скорости.
Так а в чем тут питон выигрывает у того же go например?
Я конечно не пишу на питоне но насколько знаю питон это не про скорость работы а про скорость написания кода )
можно на "go" ты прав
Вы вернулись. Супер.
import asyncio
from sys import argv
import aiohttp
proxy_type = "http"
test_url = "http://www.google.com"
timeout_sec = 4
class bcolors:
HEADER = '\033[95m'
OKBLUE = '\033[94m'
OKGREEN = '\033[92m'
WARNING = '\033[93m'
FAIL = '\033[91m'
ENDC = '\033[0m'
BOLD = '\033[1m'
UNDERLINE = '\033[4m'
# read the list of proxy IPs in proxyList from the first Argument given
filename = argv[1]
proxylistfile = open(filename)
proxyList = proxylistfile.read().splitlines()
async def is_bad_proxy(ipport):
try:
session = aiohttp.ClientSession()
resp = await session.get(test_url, proxy=ipport, timeout=timeout_sec)
if not resp.headers["Via"]:
raise "Error"
print(bcolors.OKBLUE + "Working:", ipport + bcolors.ENDC)
session.close()
except:
session.close()
print(bcolors.FAIL + "Not Working:", ipport + bcolors.ENDC)
tasks = []
loop = asyncio.get_event_loop()
for item in proxyList:
tasks.append(asyncio.ensure_future(is_bad_proxy("http://" + item)))
print(bcolors.HEADER + "Starting... \n" + bcolors.ENDC)
loop.run_until_complete(asyncio.wait(tasks))
print(bcolors.HEADER + "\n...Finished" + bcolors.ENDC)
loop.close()Интересный код. Вот еще одна реализация, но быстрее:
import aiohttp
import asyncio
import ipaddress
from itertools import islice
from random import randint, sample
class Colors:
HEADER = '\033[95m'
OKBLUE = '\033[94m'
FAIL = '\033[91m'
ENDC = '\033[0m'
async def check_proxy(session, ip, port, protocol, timeout_sec, print_lock):
url = f"{protocol}://{ip}:{port}"
start_time = asyncio.get_event_loop().time()
try:
async with session.get("http://google.com", proxy=url, timeout=timeout_sec) as response:
response.raise_for_status()
elapsed_time = asyncio.get_event_loop().time() - start_time
async with print_lock:
print(f"{Colors.OKBLUE}Прокси {url} работает. Время ответа: {elapsed_time:.2f} секунд{Colors.ENDC}", flush=True)
except asyncio.TimeoutError:
elapsed_time = asyncio.get_event_loop().time() - start_time
async with print_lock:
print(f"{Colors.FAIL}Прокси {url} превысило время ожидания. Время ответа: {elapsed_time:.2f} секунд (Тайм-аут){Colors.ENDC}", flush=True)
except aiohttp.ClientError as e:
elapsed_time = asyncio.get_event_loop().time() - start_time
async with print_lock:
print(f"{Colors.FAIL}Прокси {url} не работает. Ошибка: {e}. Время ответа: {elapsed_time:.2f} секунд{Colors.ENDC}", flush=True)
async def generate_proxy_list(num_proxies, filename):
ip_network = ipaddress.IPv4Network("192.168.0.0/16")
all_ips = [ip for ip in islice(ip_network.hosts(), 1, None)]
selected_ips = sample(all_ips, num_proxies)
with open(filename, 'w') as file:
for ip in selected_ips:
port = randint(80, 9000) # Исправлено: случайные порты в пределах 80-9000
file.write(f"{ip}:{port}\n")
async def process_proxy_list(session, proxy_list, timeout_sec, print_lock):
await asyncio.gather(*(check_proxy(session, *proxy, timeout_sec, print_lock) for proxy in proxy_list))
async def main():
filename = "proxy_list.txt"
num_proxies = 2000
timeout_sec = 4
# Закомментируйте, чтобы не генерировать список прокси-адресов
await generate_proxy_list(num_proxies, filename)
with open(filename, encoding='utf-8') as proxylistfile:
proxyList = [line.strip().split(':') for line in proxylistfile]
proxies = [(ip, int(port), 'http') for ip, port in proxyList]
timeout = aiohttp.ClientTimeout(total=timeout_sec)
async with aiohttp.ClientSession(timeout=timeout) as session:
print_lock = asyncio.Lock()
await process_proxy_list(session, proxies, timeout_sec, print_lock)
if __name__ == "__main__":
asyncio.run(main())не факт. чекер на GO выигрывает.
Я не по теме конечно но объясните мне , если не сложно .
посмотри видео и все поймешь. не плохо чувак объясняет
На питоне не пишу но за асинхронность я понимаю . Мне интересно зачем создавать 1к+ потоков (именно thread)
