FastProxyChecker

.gitignore

@@ -0,0 +1,7 @@
+.idea
+__pycache__
+venv
+venv-lin
+checked_proxies.txt
+unchecked_proxies.txt
+fast_proxy_checker.spec

fast_proxy_checker.py

@@ -0,0 +1,93 @@
+import asyncio
+import sys
+import time
+from concurrent.futures import ThreadPoolExecutor
+from getpass import getpass
+from os.path import exists
+
+import click
+import requests
+
+
+def log_print(*objects, sep=' ', end='\n', file=sys.stdout, flush=False):
+    if objects:
+        print(time.strftime('[%d.%m.%Y - %H:%M:%S]'), *objects, sep=sep, end=end, file=file, flush=flush)
+
+
+def check_proxies(threads: int, proxies: list, ping_domain: str):
+    # Принимает прокси в формате http(s)/socks5:ip:port:login:pass
+    # Возвращает рабочие в том-же списке
+    requests_proxies = []
+    for idx, line in enumerate(proxies.copy()):
+        proxy_params = line.split(':')
+        if len(proxy_params) < 3:
+            log_print('Строка {}. Неправильный синтаксис.'.format(idx + 1))
+            proxies.remove(line)
+            continue
+        if not proxy_params[0] in ('http', 'https', 'socks5'):
+            log_print('Строка {}. Неправильный синтаксис.'.format(idx + 1))
+            proxies.remove(line)
+            continue
+        if len(proxy_params) >= 5:
+            requests_proxies.append('{}://{}:{}@{}:{}'.format(proxy_params[0], proxy_params[3], proxy_params[4], proxy_params[1], proxy_params[2]))
+        else:
+            requests_proxies.append('{}://{}:{}'.format(proxy_params[0], proxy_params[1], proxy_params[2]))
+    with ThreadPoolExecutor(max_workers=threads) as pool:
+        loop = asyncio.get_event_loop()
+        futures = [loop.run_in_executor(pool, _proxy_is_worked, proxy, ping_domain) for proxy in requests_proxies]
+        result = loop.run_until_complete(asyncio.gather(*futures))
+    for idx, proxy in enumerate(proxies.copy()):
+        if not result[idx]:
+            proxies.remove(proxy)
+
+
+def _proxy_is_worked(proxy: str, ping_domain: str) -> bool:
+    try:
+        response = requests.get(ping_domain, proxies={'http': proxy, 'https': proxy}, timeout=10)
+        if response.status_code not in (requests.codes.ok, requests.codes.no_content, requests.codes.created):
+            return False
+        return True
+    except Exception as e:
+        del e
+        return False
+
+
+@click.command()
+@click.option('--threads', '-t', default=1000, show_default=True, help='Количество потоков')
+@click.option('--repeats', '-r', default=1, show_default=True, help='Количество прогонов')
+@click.option('--input_file', '-i', default='unchecked_proxies.txt', show_default=True, help='Входной файл')
+@click.option('--output_file', '-o', default='checked_proxies.txt', show_default=True, help='Выходной файл')
+@click.option('--ping_domain', '-p', default='https://www.google.com/gen_204', show_default=True, help='Домен для проверки')
+def run_proxies_checker(threads: int, repeats: int, input_file: str, output_file: str, ping_domain: str):
+    """Многопоточный чекер прокси"""
+    if not exists(input_file):
+        log_print('Файл "{}" не найден!'.format(input_file))
+        return
+    if repeats < 0:
+        log_print('Количество прогонов не может быть отрицательным!')
+        return
+    with open(input_file) as fp:
+        proxies_list = []
+        for line in fp:
+            valuable_line = line.rstrip()
+            if valuable_line:
+                proxies_list.append(valuable_line)
+    proxies_list = list(set(proxies_list))  # Удаляем повторяющиеся строки
+    log_print('Обнаружено {} строк. Запуск проверки прокси.'.format(len(proxies_list)))
+    for repeat in range(repeats):
+        log_print('Старт прогона №{}'.format(repeat + 1))
+        check_proxies(threads, proxies_list, ping_domain)
+        log_print('Прогон №{} окончен! Найдено {} рабочих прокси!'.format(repeat + 1, len(proxies_list)))
+        time.sleep(5)
+    with open(output_file, 'w') as fp:
+        for proxy in proxies_list:
+            fp.write('{}\n'.format(proxy))
+
+
+def main():
+    run_proxies_checker()
+    getpass('Нажмите Enter, чтобы завершить программу.')
+
+
+if __name__ == '__main__':
+    main()

requirements.txt

@@ -0,0 +1,12 @@
+altgraph==0.16.1
+certifi==2018.11.29
+chardet==3.0.4
+Click==7.0
+future==0.17.1
+idna==2.8
+macholib==1.11
+pefile==2018.8.8
+PyInstaller==3.4
+pywin32-ctypes==0.2.0
+requests==2.21.0
+urllib3==1.24.1