Runion
This is a sample guest message. Register a free account today to become a member! Once signed in, you'll be able to participate on this site by adding your own topics and posts, as well as connect with other members through your own private inbox!
Ищу python разработчика для написания прокси-чекера
- Thread starter asdasd
- Start date
johnsmcsilver
Light Weight
- Депозит
- $0
Так а в чем тут питон выигрывает у того же go например?
В любом случае код упирается в I/O и в ядра. Потоков наплодить можно хоть сразу 10кк - да только больше времени потратится на переключении контекста.
Тут запускать в N-1 потоков (где N количество ядер), ограничивать таймаутами да и все наверное, чтение\запись в файл копейки во времени по сравнению со всем остальным, остальной код и того меньше.
Куда уже быстрее ¯\_(ツ)_/¯
А дальше если ускорять, то докупкой доп серверов
Исходный код
Автор - я. Писал от скуки несколько недель назад. Доработай напильником и будет тебе чекер.
Асинхронный, но не очень быстрый (питон это действительно не про скорость). И суть здесь не в ресурсах системы, а в таймауте ответа. Можно дать 3 секунды на соединение, а можно 10. И каждый невалидный проксик заблочит тебе поток на указанное количество секунд.
Последнее редактирование: 18.01.2024
Автор - я. Писал от скуки несколько недель назад. Доработай напильником и будет тебе чекер.
Асинхронный, но не очень быстрый (питон это действительно не про скорость). И суть здесь не в ресурсах системы, а в таймауте ответа. Можно дать 3 секунды на соединение, а можно 10. И каждый невалидный проксик заблочит тебе поток на указанное количество секунд.
Последнее редактирование: 18.01.2024
Интересный код. Вот еще одна реализация, но быстрее:
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())
Вы вернулись. Супер.
Вот еще как пример асинхронного чекера с применением модуля asyncio и aiohttp.
Если потребуется что-то добавить, отпишите.
Python:
Скопировать в буфер обмена
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()
Последнее редактирование: 18.01.2024
Python:
Скопировать в буфер обмена
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())
P.S. Продублировал для удобства.
Скопировать в буфер обмена
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())
P.S. Продублировал для удобства.
Если речь реально идет о лямах проксей и нужна макс скорость, то я бы решал задачу так:
- сначала raw-сокетами посылаем SYN-пакеты на все пары ip
ort с определенным рейтом, и в параллельном потоке\процессе ловим прилетающие SYN+ACK (примерно так работает масскан). На выходе получаем первичный список iport которые онлайн и гарантированно слушают нужный порт.
- далее по этому списку проходимся уже обычными сокетами с использованием низкоуровневого сокетного мультиплексирования (функи select/poll/epoll), т.е. устанавливаем полноценное tcp-соединение, посылаем чек-запрос, вычитываем респонс и тд.
- сначала raw-сокетами посылаем SYN-пакеты на все пары ip
ort с определенным рейтом, и в параллельном потоке\процессе ловим прилетающие SYN+ACK (примерно так работает масскан). На выходе получаем первичный список iport которые онлайн и гарантированно слушают нужный порт.- далее по этому списку проходимся уже обычными сокетами с использованием низкоуровневого сокетного мультиплексирования (функи select/poll/epoll), т.е. устанавливаем полноценное tcp-соединение, посылаем чек-запрос, вычитываем респонс и тд.
А я бы сделал серверную часть, которая получает и хранит пары проксей. И клиентскую часть, которая запрашивает у сервера кого надо чекнуть, а затем вернуть результаты чека. При этом сервер должен отдавать пачками штук по 100 прокси.
Клиентов можно масштабировать согласно требованиям.
Клиентов можно масштабировать согласно требованиям.
ГошаМарков
Midle Weight
- Депозит
- $0
посмотри видео и все поймешь. не плохо чувак объясняет