Налаштування проксі у Python requests: підключення SOCKS5 та HTTP-проксі

Цей посібник пояснює, як підключити HTTP- та SOCKS5-проксі Socks5Proxy до бібліотеки Python requests, використовуючи аутентифікацію за логіном та паролем, а також як ротувати список IP і налаштувати повторні спроби.

Початок Переглянути ціни

Що потрібно підготувати?

Потрібні Python 3.8+ та бібліотека requests. Для використання SOCKS5 встановіть додатковий пакет 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 та виводить 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, DNS-розв'язка відбувається на вашому сервері, і лише 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, використовує проксі по порядку та повторює невдалі запити через інший проксі до трьох разів. Також додано паузу, щоб уникнути надто швидких запитів до одного домену.

Краще дати невдалому проксі відпочити, ніж відразу відмовлятися від нього. Часто помилки виникають через тимчасові проблеми на цільовому сайті.

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, а авторизацію через aiohttp.BasicAuth. aiohttp не підтримує 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 до всіх запитів