Konfiguracja proxy w Python requests, podłączanie SOCKS5 i HTTP

Co należy przygotować?

Wymagany jest Python 3.8 lub nowszy oraz biblioteka requests. Aby użyć SOCKS5, należy zainstalować pakiet z obsługą SOCKS. Zainstalowanie requests[socks] instaluje również PySocks, co umożliwia korzystanie z adresów socks5 i socks5h.

Informacje o proxy można sprawdzić w sekcji Moje proxy w panelu. W tym przykładzie używamy IP 198.51.100.23, port HTTP 8000, port SOCKS5 1080, nazwa użytkownika spx_user, hasło spx_pass. Zastąp te wartości własnymi.

pip install "requests[socks]"

Jak połączyć się z proxy HTTP?

W requests, jeśli dodasz adres proxy do słownika proxies, żądania będą wysyłane przez ten proxy. Ważne jest, aby dodać ten sam adres proxy zarówno pod kluczem http, jak i https. Brak klucza https spowoduje, że żądania HTTPS będą wysyłane bezpośrednio, omijając proxy.

Żądania do witryn HTTPS przechodzą przez proxy za pomocą metody CONNECT, przesyłając dane w zaszyfrowanej formie od początku do końca. Poniższy kod wysyła żądanie do adresu sprawdzającego IP i sprawdza, czy w odpowiedzi widoczne jest IP proxy.

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

Jak połączyć się z proxy SOCKS5?

Zalecamy używanie socks5h przed adresem. socks5h przekazuje rozpoznawanie nazw domen do proxy, dzięki czemu zapytania DNS z Twojego serwera nie są widoczne na zewnątrz, a nazwy są rozpoznawane na podstawie kraju, w którym znajduje się proxy. Jeśli użyjesz socks5, rozpoznawanie nazw odbywa się na Twoim serwerze, a tylko IP jest przekazywane do proxy.

Należy używać portu SOCKS5. Podanie adresu SOCKS5 na porcie HTTP uniemożliwi połączenie, więc zwróć uwagę na oba porty w każdej linii.

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())

Jak rotować wiele proxy przy wysyłaniu żądań?

Przy zbieraniu wielu stron standardem jest pobranie listy proxy i używanie kolejnego dla każdego żądania. Poniższy kod pobiera listę SOCKS5 przez API, używa jej kolejno i ponawia nieudane żądania (max 3 razy) z innym proxy. Dodano też krótkie przerwy, aby nie wysyłać zbyt wielu żądań zbyt szybko pod tę samą domenę.

Lepiej pozwolić nieudanemu proxy odpocząć przez chwilę niż od razu je odrzucić. Tymczasowe błędy często są spowodowane stanem docelowej strony.

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)

Jak utrzymać sesję i ciasteczka?

W przepływach wymagających ciągłości ciasteczek, takich jak logowanie lub nawigacja po stronach, używaj jednego proxy na jedną sesję requests.Session. Wprowadź adres do właściwości proxies obiektu sesji, a wszystkie żądania z tej sesji będą wychodzić z tego samego IP. Po zakończeniu przepływu zamknij sesję i utwórz nową z następnym proxy.

Jeśli ustawisz nagłówki User-Agent i Accept-Language dla każdej sesji, strona docelowa będzie widzieć spójne dane.

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()

Jak używać uwierzytelniania IP do połączenia?

Zarejestruj IPv4 serwera wykonującego ten kod w panelu, aby używać adresów bez loginu. Nie musisz przechowywać haseł w kodzie ani konfiguracji, co ułatwia wdrożenie na wielu serwerach. Wybierz format ip:port przy pobieraniu listy.

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)

Jak wysyłać wiele żądań jednocześnie i szybko?

requests czeka na zakończenie każdego żądania, więc aby szybko pobrać wiele stron, należy uruchomić wiele zadań jednocześnie. Najprostszym sposobem jest użycie ThreadPoolExecutor z concurrent.futures. Utrzymuj liczbę zadań poniżej liczby proxy, aby uniknąć przeciążenia pojedynczego IP.

Poniższy kod dzieli listę adresów na 20 wątków, a każdy wątek wybiera kolejny proxy z listy. itertools.cycle nie ma problemów z pobieraniem kolejnych wartości nawet w wielu wątkach, ale jeśli potrzebujesz dokładnej kolejności, użyj blokady.

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)

Czy mogę używać bibliotek asynchronicznych, takich jak httpx lub aiohttp?

Tak. W httpx wystarczy przekazać adres proxy jako argument proxy, a dla SOCKS5 należy zainstalować httpx[socks]. AsyncClient również obsługuje te same argumenty.

aiohttp wymaga podania adresu HTTP proxy w parametrze proxy dla każdego żądania oraz proxy_auth z aiohttp.BasicAuth. aiohttp nie obsługuje natywnie SOCKS, więc do SOCKS5 potrzebny jest connector z pakietu aiohttp-socks.

Podejście asynchroniczne pozwala przetwarzać tysiące żądań przy małych zasobach, ale bez limitu równoczesnych żądań możesz przeciążyć zarówno docelową stronę, jak i proxy. Użyj asyncio.Semaphore, aby ograniczyć liczbę równoczesnych żądań.

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)

Jak rozwiązać częste błędy?

Błąd Missing dependencies for SOCKS support oznacza, że requests[socks] nie jest zainstalowany. Zainstaluj go ponownie w tym samym środowisku wirtualnym.

407 Proxy Authentication Required oznacza, że nazwa użytkownika lub hasło są nieprawidłowe lub używasz autoryzacji IP, ale adres IP serwera nie jest zarejestrowany. Jeśli hasło zawiera znaki specjalne, musisz je zakodować za pomocą urllib.parse.quote.

Przekroczenie czasu połączenia najczęściej wynika z nieprawidłowego portu. Upewnij się, że używasz portu HTTP dla adresów HTTP i portu SOCKS5 dla adresów socks5h. Wartość timeout zapobiega zatrzymaniu całego zadania przez brak odpowiedzi.

• Błąd SOCKS: pip install "requests[socks]"
• 407: Sprawdź nazwę użytkownika, hasło lub zarejestrowane IP
• Przekroczenie czasu: rozróżnienie portów HTTP i SOCKS5
• Dodaj timeout do wszystkich żądań