Настройка прокси в Python requests: подключение SOCKS5 и HTTP

Что нужно подготовить?

Требуется Python 3.8+ и библиотека requests. Для SOCKS5 нужно установить пакет с поддержкой SOCKS. Установка requests[socks] включает PySocks, что позволяет использовать адреса socks5 и socks5h.

Данные прокси доступны в разделе 'Мои прокси' в консоли. Пример: IP 198.51.100.23, HTTP-порт 8000, SOCKS5-порт 1080, логин spx_user, пароль spx_pass. Подставьте свои значения.

pip install "requests[socks]"

Как подключиться к HTTP-прокси?

requests отправляет запросы через указанный в словаре proxies адрес. Важно указать один и тот же HTTP-адрес прокси для ключей http и https. Без https HTTPS-запросы будут отправляться напрямую, минуя прокси.

Запросы к HTTPS проходят через CONNECT-туннель в зашифрованном виде. Код ниже проверяет IP через прокси.

import requests

proxy = "http://spx_user:spx_pass@198.51.100.23:8000"
proxies = {"http": proxy, "https": proxy}

r = requests.get("https://api.ipify.org?format=json", proxies=proxies, timeout=20)
print(r.status_code, r.json())  # the "ip" field should show the proxy IP

Как подключиться к прокси SOCKS5?

Рекомендуем использовать префикс socks5h в адресе. socks5h делегирует DNS-запросы прокси-серверу, скрывая ваши DNS-запросы и разрешая домены по стране прокси. Если использовать socks5, домены будут разрешаться на вашем сервере, и только IP передается через прокси.

Используйте порт SOCKS5. Если вставить адрес SOCKS5 в порт HTTP, соединение не установится, поэтому различайте порты в каждой строке.

import requests

proxy = "socks5h://spx_user:spx_pass@198.51.100.23:1080"
proxies = {"http": proxy, "https": proxy}

r = requests.get("https://api.ipify.org?format=json", proxies=proxies, timeout=20)
print(r.json())

Как циклически переключать прокси для запросов?

При сборе многих страниц базовый подход - загрузить список прокси и использовать следующий для каждого запроса. Этот код получает SOCKS5-прокси через API, использует их по очереди и повторяет неудачные запросы до 3 раз с другим прокси. Также добавлены паузы, чтобы избежать слишком частых запросов к одному домену.

Лучше не удалять прокси сразу после ошибки, а дать ему отдохнуть и попробовать снова. Часто ошибки временные и связаны с состоянием целевого сайта.

import itertools
import time
import requests

API = "https://socks5proxy.io/api/v1"
KEY = "YOUR_API_KEY"

lines = requests.get(
    API + "/proxies/download",
    params={"format": "user:pass@ip:port", "protocol": "socks5"},
    headers={"Authorization": "Bearer " + KEY},
    timeout=30,
).text.split()
pool = itertools.cycle(lines)

def fetch(url, tries=3):
    for _ in range(tries):
        p = "socks5h://" + next(pool)
        try:
            r = requests.get(url, proxies={"http": p, "https": p}, timeout=20)
            if r.status_code == 200:
                return r.text
        except requests.RequestException:
            pass
        time.sleep(1)
    return None

for page in range(1, 6):
    html = fetch("https://example.com/list?page=" + str(page))
    print(page, "ok" if html else "failed")
    time.sleep(0.5)

Как сохранить сессию и cookies?

Для последовательностей с куки, таких как вход или навигация по страницам, зафиксируйте один прокси на один requests.Session. Установите адрес в атрибут proxies сессии, и все запросы в ней будут идти с одного IP. По завершении закройте сессию и создайте новую с другим прокси.

Если также задать User-Agent и Accept-Language заголовки для каждой сессии, внешний вид на целевом сайте будет постоянным.

import requests

s = requests.Session()
s.proxies = {
    "http": "http://spx_user:spx_pass@198.51.100.23:8000",
    "https": "http://spx_user:spx_pass@198.51.100.23:8000",
}
s.headers.update({"User-Agent": "Mozilla/5.0", "Accept-Language": "en-US,en;q=0.9"})

s.get("https://example.com/", timeout=20)          # cookies are stored in the session
r = s.get("https://example.com/account", timeout=20)  # same IP, same cookies
print(r.status_code)
s.close()

Как использовать авторизацию по IP?

Если зарегистрировать IPv4 сервера, где выполняется этот код, в панели управления, то можно использовать адрес без логина и пароля. Это упрощает развертывание кода на нескольких серверах, так как пароль не нужно хранить в коде или конфигурационных файлах. Список также можно получить в формате ip:port.

import requests

proxy = "http://198.51.100.23:8000"  # no user or password, server IP is authorized in the dashboard
r = requests.get("https://api.ipify.org", proxies={"http": proxy, "https": proxy}, timeout=20)
print(r.text)

Как быстро отправлять несколько запросов одновременно?

requests работает синхронно, поэтому для быстрой загрузки множества страниц используйте параллельные задачи, например, ThreadPoolExecutor из concurrent.futures. Держите число задач меньше числа прокси, чтобы избежать перегрузки одного IP.

Код ниже распределяет список адресов между 20 воркерами, и каждый воркер выбирает следующий прокси из списка. itertools.cycle безопасен для использования в нескольких потоках, но если нужен точный порядок, используйте блокировки.

import itertools
from concurrent.futures import ThreadPoolExecutor
import requests

lines = open("proxies.txt").read().split()  # user:pass@ip:port per line
pool = itertools.cycle(lines)

def get(url):
    p = "http://" + next(pool)
    try:
        return url, requests.get(url, proxies={"http": p, "https": p}, timeout=20).status_code
    except requests.RequestException as e:
        return url, type(e).__name__

urls = ["https://example.com/item/" + str(i) for i in range(200)]
with ThreadPoolExecutor(max_workers=20) as ex:
    for url, result in ex.map(get, urls):
        print(url, result)

Можно ли использовать асинхронные библиотеки, такие как httpx или aiohttp?

Да. В httpx можно указать адрес прокси в параметре proxy при создании клиента, а для SOCKS5 нужно установить httpx[socks], чтобы использовать socks5 адрес. Асинхронный клиент AsyncClient также принимает те же параметры.

В aiohttp укажите адрес HTTP прокси в параметре proxy, а для аутентификации используйте proxy_auth с aiohttp.BasicAuth. aiohttp не поддерживает SOCKS из коробки - для SOCKS5 используйте коннектор из пакета aiohttp-socks.

Асинхронный метод позволяет обрабатывать тысячи запросов с малыми ресурсами, но без ограничения одновременных запросов возможна перегрузка сайта и прокси. Используйте asyncio.Semaphore для ограничения количества параллельных запросов.

import httpx

with httpx.Client(proxy="http://spx_user:spx_pass@198.51.100.23:8000", timeout=20) as client:
    print(client.get("https://api.ipify.org").text)

Как исправить частые ошибки?

Ошибка 'Missing dependencies for SOCKS support' означает, что не установлен requests[socks]. Установите его в той же виртуальной среде.

407 Proxy Authentication Required означает, что имя пользователя или пароль неверны, или IP сервера не добавлен в список разрешенных IP. Если в пароле есть специальные символы, используйте urllib.parse.quote для их экранирования.

Чаще всего таймаут соединения возникает из-за неверного порта. Убедитесь, что используете HTTP-порт для HTTP-адресов и SOCKS5-порт для socks5h. Указание timeout предотвращает зависание всего задания из-за отсутствия ответа.

• Ошибка SOCKS: pip install "requests[socks]"
• 407: проверьте имя пользователя/пароль или зарегистрированные IP
• Таймаут: различие между HTTP и SOCKS5 портами
• Добавлять timeout ко всем запросам