Socks5Proxy

Python requestsプロキシ設定、SOCKS5とHTTPプロキシ接続方法

このガイドではPython requestsライブラリにSocks5ProxyのHTTPプロキシとSOCKS5プロキシを接続する方法、ユーザー名・パスワード認証、リストローテーションとリトライコードを段階的に説明します。

はじめる価格を見る

準備するものは？

Python 3.8以上とrequestsライブラリが必要です。SOCKS5を使用する場合はSOCKSサポートパッケージも一緒にインストールする必要があります。requests[socks]をインストールするとPySocksも一緒にインストールされ、socks5とsocks5hアドレスを使用できます。

プロキシ情報はダッシュボードの「マイプロキシ」画面で確認できます。この例ではIP 198.51.100.23、HTTPポート8000、SOCKS5ポート1080、ID spx_user、パスワード spx_passを使用します。実際の値に置き換えてください。

pip install "requests[socks]"

HTTPプロキシはどうやって接続しますか？

requestsはproxies辞書にアドレスを入れると、そのプロキシでリクエストを送信します。httpとhttpsの両方のキーに同じHTTPプロキシアドレスを入れることが重要です。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プロキシはどのように接続しますか？

SOCKS5はアドレスの前にsocks5hと記述することを推奨します。socks5hはドメイン名解決をプロキシ側で行うため、自サーバーのDNSリクエストが外部に漏れず、プロキシのある国基準で名前解決されます。socks5と記述すると、名前解決を自サーバーで行った後IPのみをプロキシに送信します。

ポートはSOCKS5ポートを使用してください。HTTPポートにSOCKS5アドレスを入力しても接続されないので、行ごとにある2つのポートを区別してください。

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

複数のプロキシを循環してリクエストするにはどうすればいいですか？

多くのページを収集する際は、リストを受け取ってリクエストごとに次のプロキシを使う方法が基本です。以下のコードはAPIでSOCKS5リストを受け取り順番に使い、失敗したリクエストは別のプロキシで最大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)

セッションとクッキーを維持するにはどうすればよいですか？

ログインやページ遷移のようにクッキーを維持する必要があるフローでは、requests.Sessionに1つのプロキシを固定して使用しましょう。セッションオブジェクトの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を登録すると、アドレスからIDとパスワードを省略できます。パスワードをコードや設定ファイルに置かなくて良いため、複数サーバーに同じコードを展開しやすくなります。リスト取得時もformatを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は1つのリクエストが終わるまで待つ方式なので、多くのページを速く取得するには複数のタスクを同時に実行する必要があります。最も簡単な方法はconcurrent.futuresのThreadPoolExecutorです。タスク数をプロキシ数より少なくすると、1つの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はリクエストごとにproxy引数にHTTPプロキシアドレスを設定し、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ポートを、socks5hアドレスにはSOCKS5ポートを使用したか確認してください。timeout値を必ず設定しておくと、応答のないリクエストが作業全体を止めてしまうことを防げます。

  • SOCKSエラー: pip install "requests[socks]"
  • 407: ユーザー名・パスワードまたは登録したIPを確認
  • タイムアウト: HTTPポートとSOCKS5ポートの区別
  • 全リクエストにtimeout設定
Python requestsプロキシ設定、SOCKS5・HTTP接続 | Socks5Proxy