vps の主ディレクトリに python ファイルを作成します(例:proxy_test.py)。作成後、以下のコードをコピー&ペーストし、必要に応じてプロキシの URL やパスを自分で変更してください。
import requests
import concurrent.futures
import time
import os
import json
from urllib3.exceptions import InsecureRequestWarning
# urllib3からのInsecureRequestWarningのみを抑制
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
# --- 設定パラメータ ---
# **更新されたURL:より安定した公開プロキシリストのURLを使用**
PROXY_URL = 'https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/http.txt'
# プロキシの匿名性を確認するためのターゲット(リクエストデータをエコーバックする必要があります)
ANONYMITY_TARGET = 'http://httpbin.org/get'
# 最終的な接続確認のための厳格なターゲット(SSL/TLSをテスト)
FINAL_CONNECTIVITY_TARGET = 'https://www.google.com'
# **更新:** 単一プロキシテストの合計タイムアウト(秒)。5秒に戻します。
TIMEOUT = 5
# **新規/更新:** 最大許容合計レイテンシ(秒)。これより遅いものは拒否されます。
MAX_LATENCY = 4.5
# 最大同時スレッド数 - 安定性向上のために200に調整
MAX_WORKERS = 180
# ------------------
def fetch_proxies(url):
"""
指定されたURLからプロキシリストを取得します。
ブラウザをシミュレートするための詳細なヘッダーを含みます。
"""
print(f"{url} からプロキシリストを取得しようとしています...")
try:
# ブラウザのようなヘッダー
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36 Edg/129.0.0.0',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Referer': 'https://www.google.com/',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
}
# 一般的なプロキシの問題のためにSSL検証を無効にします
response = requests.get(url, headers=headers, timeout=10, verify=False)
response.raise_for_status()
proxies = [p.strip() for p in response.text.splitlines() if p.strip()]
print(f"{len(proxies)} 個のプロキシを正常に取得しました。")
return proxies
except requests.exceptions.RequestException as e:
print(f"❌ エラー:プロキシリストの取得に失敗しました。URLまたはネットワーク接続を確認してください。エラーメッセージ:{e}")
return []
def get_real_ip():
"""
比較のためにクライアントの実際のIPアドレスを取得します。
より堅牢性を向上させるために2つの方法を使用します。
"""
# --- 方法1を試す: httpbin.org ---
try:
print("httpbin.orgを使用して実際のIPを取得しようとしています...")
response = requests.get(ANONYMITY_TARGET, timeout=5)
response.raise_for_status()
real_ip = response.json().get('origin')
if real_ip:
# httpbin.orgは時々複数のIPアドレスを返すので、最初のものだけを取得します
return real_ip.split(',')[0].strip()
except Exception as e:
print(f"httpbin.orgの取得に失敗しました: {type(e).__name__}。代替案を試みています...")
# --- 方法2を試す: api.ipify.org (代替案) ---
try:
response = requests.get('https://api.ipify.org?format=json', timeout=5)
response.raise_for_status()
return response.json().get('ip')
except Exception as e:
print(f"api.ipify.orgの取得に失敗しました: {type(e).__name__}。")
return None
def check_proxy(proxy, real_ip):
"""
単一のプロキシの匿名性、速度(レイテンシ)、および実際の接続性をテストします。
成功した場合は (proxy, True, reason, latency) を返します。
"""
ip_port = proxy.split(':')
if len(ip_port) != 2:
return proxy, False, "フォーマットエラー", None
proxy_ip = ip_port[0]
proxies = {
'http': f'http://{proxy}',
'https': f'http://{proxy}',
}
# --- 1. 匿名性チェック (ANONYMITY_TARGETを使用) ---
try:
start_time = time.time()
# 初期接続のためにグローバルTIMEOUTを使用
response = requests.get(ANONYMITY_TARGET, proxies=proxies, timeout=TIMEOUT, verify=False)
# 基本的な接続チェック
if response.status_code != 200:
return proxy, False, "接続/ステータスコード異常", None
# 匿名性チェック
response_json = response.json()
reported_ip = response_json.get('origin')
# 匿名性ロジック:報告されたIPは実際のIPであってはならず、プロキシのIPと一致する必要があります
if not (reported_ip and reported_ip != real_ip and reported_ip.startswith(proxy_ip)):
return proxy, False, "透明/身元情報漏洩", None
except requests.exceptions.Timeout:
return proxy, False, f"タイムアウト (>{TIMEOUT}s)", None
except Exception as e:
return proxy, False, f"匿名性チェックに失敗しました: {type(e).__name__}", None
# --- 2. 最終的な実際の接続性チェック (FINAL_CONNECTIVITY_TARGETを使用) ---
try:
# Googleに対してテストします。最終テストのためにグローバルTIMEOUTを使用します。
requests.get(FINAL_CONNECTIVITY_TARGET, proxies=proxies, timeout=TIMEOUT, verify=False).raise_for_status()
# 最初のテストの開始からの合計レイテンシを計算します
total_latency = time.time() - start_time
# **速度チェック:** 最大レイテンシ閾値を下回るプロキシのみを受け入れます
if total_latency > MAX_LATENCY:
return proxy, False, f"速度が遅すぎます (合計レイテンシ: {total_latency:.2f}s)", None
# ここに到達すれば成功、ステータスコードは良好で、速度も許容範囲です
return proxy, True, f"高匿名/匿名/高速接続 (合計レイテンシ: {total_latency:.2f}s)", total_latency
except Exception as e:
# 匿名性を通過したが最終Googleテストに失敗した場合、ブロックされている/互換性がないと見なされます
return proxy, False, f"ターゲットサイトにブロックされました: {type(e).__name__}", None
def batch_check(proxies, real_ip):
"""マルチスレッドを使用してプロキシをバッチテストします。"""
successful_proxies = []
total_proxies = len(proxies)
# ThreadPoolExecutorを使用して同時実行を実現
with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
# すべてのタスクを提出し、各チェック関数に実際のIPを渡します
future_to_proxy = {executor.submit(check_proxy, proxy, real_ip): proxy for proxy in proxies}
# ソート用に (proxy, latency) を格納するリスト
working_proxies_with_latency = []
# 結果を1つずつ取得し、進捗を印刷します
for i, future in enumerate(concurrent.futures.as_completed(future_to_proxy)):
proxy = future_to_proxy[future]
try:
# 関数は4つの値を返します: (proxy, is_working, reason, latency)
proxy, is_working, reason, latency = future.result()
except Exception as exc:
is_working = False
latency = None
reason = f"スレッドエラー: {exc}"
# リアルタイムの進捗を印刷
progress = f"[{i+1}/{total_proxies}]"
print(f"\r{progress} 検出中... 現在の利用可能: {len(working_proxies_with_latency)}", end="", flush=True)
if is_working:
working_proxies_with_latency.append((proxy, latency))
# レイテンシでプロキシをソート(最速のものが最初)
working_proxies_with_latency.sort(key=lambda x: x[1])
successful_proxies = [p[0] for p in working_proxies_with_latency]
# 進捗出力をクリアし、改行
print("\n")
return successful_proxies
if __name__ == '__main__':
start_time = time.time()
# 0. 匿名性比較のためにクライアントの実際のIPを取得します
print("匿名性比較のためにあなたの実際のIPアドレスを取得しています...")
real_ip = get_real_ip()
if not real_ip:
print("❌ 警告:実際のIPを取得できませんでした。匿名性チェックをスキップします。プロキシの質が低下する可能性があります。")
real_ip = '0.0.0.0' # フォールバック値を使用
else:
print(f"✅ あなたの実際のIPアドレスは: {real_ip}")
# 1. プロキシリストを取得
proxy_list = fetch_proxies(PROXY_URL)
if not proxy_list:
print("プログラムを終了します。プロキシリストを取得できませんでした。")
else:
# 新しい速度制限を反映するように出力を更新
print(f"{len(proxy_list)} 個のプロキシをテストし始めます (匿名性を要求、速度 < {MAX_LATENCY:.1f}s, ターゲット: {FINAL_CONNECTIVITY_TARGET})、同時数:{MAX_WORKERS}...")
# 2. バッチチェック
successful_proxies = batch_check(proxy_list, real_ip)
end_time = time.time()
# 3. 結果の要約と保存
print("\n--- 結果の要約 ---")
print(f"総時間:{end_time - start_time:.2f} 秒")
print(f"総プロキシ数:{len(proxy_list)}")
print(f"利用可能(高匿名/匿名/高速)プロキシ数:{len(successful_proxies)}")
if len(proxy_list) > 0:
print(f"成功率:{len(successful_proxies)/len(proxy_list)*100:.2f}%")
output_file = 'working_proxies.txt'
with open(output_file, 'w') as f:
f.write('\n'.join(successful_proxies))
print(f"\n✅ 利用可能なプロキシがファイルに保存されました:{output_file} (レイテンシ順にソートされ、最速が前にあります)")
以下は socks5 バッチテストスクリプトです。
import requests
import concurrent.futures
import time
import os
import json
from urllib3.exceptions import InsecureRequestWarning
# 注意:SOCKS5プロキシを使用するには、最初に実行する必要があります:pip install requests[socks]
# urllib3からのInsecureRequestWarningのみを抑制
requests.packages.urllib3.disable_warnings(InsecureRequestWarning)
# --- 設定パラメータ ---
# **更新:SOCKS5プロキシリストのURLを使用**
PROXY_URL = 'https://raw.githubusercontent.com/TheSpeedX/PROXY-List/master/socks5.txt'
# プロキシの匿名性を確認するためのターゲット (リクエストデータを返す必要があります)
ANONYMITY_TARGET = 'http://httpbin.org/get'
# 最終的な接続確認のための厳格なターゲット (SSL/TLSをテスト)
FINAL_CONNECTIVITY_TARGET = 'https://www.google.com'
# 単一プロキシテストの接続タイムアウト(秒)。
TIMEOUT = 5
# 最大許容合計レイテンシ(秒)。これより遅いプロキシは拒否されます。
MAX_LATENCY = 4.0
# 最大同時スレッド数 - 安定性向上のために180に調整
MAX_WORKERS = 180
# ------------------
def fetch_proxies(url):
"""
指定されたURLからプロキシリストを取得します。
"""
print(f"{url} からプロキシリストを取得しようとしています...")
try:
# ブラウザをシミュレートするためのヘッダー
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36 Edg/129.0.0.0',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8',
'Referer': 'https://www.google.com/',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
}
# SSL検証を無効にします
response = requests.get(url, headers=headers, timeout=10, verify=False)
response.raise_for_status()
proxies = [p.strip() for p in response.text.splitlines() if p.strip()]
print(f"{len(proxies)} 個のプロキシを正常に取得しました。")
return proxies
except requests.exceptions.RequestException as e:
print(f"❌ エラー:プロキシリストの取得に失敗しました。URLまたはネットワーク接続を確認してください。エラーメッセージ:{e}")
return []
def get_real_ip():
"""
比較のためにクライアントの実際のIPアドレスを取得します。
二重バックアップメカニズムを使用します。
"""
# --- 方法1を試す: httpbin.org ---
try:
print("httpbin.orgを使用して実際のIPを取得しようとしています...")
response = requests.get(ANONYMITY_TARGET, timeout=5)
response.raise_for_status()
real_ip = response.json().get('origin')
if real_ip:
# httpbin.orgは時々複数のIPアドレスを返すので、最初のものだけを取得します
return real_ip.split(',')[0].strip()
except Exception as e:
print(f"httpbin.orgの取得に失敗しました: {type(e).__name__}。代替案を試みています...")
# --- 方法2を試す: api.ipify.org (代替案) ---
try:
response = requests.get('https://api.ipify.org?format=json', timeout=5)
response.raise_for_status()
return response.json().get('ip')
except Exception as e:
print(f"api.ipify.orgの取得に失敗しました: {type(e).__name__}。")
return None
def check_proxy(proxy, real_ip):
"""
単一のプロキシの匿名性、速度、および実際の接続性をテストします。
"""
ip_port = proxy.split(':')
if len(ip_port) != 2:
return proxy, False, "フォーマットエラー", None
proxy_ip = ip_port[0]
# **重要な変更:socks5://プロトコルを使用**
proxies = {
'http': f'socks5://{proxy}',
'https': f'socks5://{proxy}',
}
# --- 1. 匿名性チェック (ANONYMITY_TARGETを使用) ---
try:
start_time = time.time()
# 初期接続のためにグローバルTIMEOUTを使用
response = requests.get(ANONYMITY_TARGET, proxies=proxies, timeout=TIMEOUT, verify=False)
# 基本的な接続チェック
if response.status_code != 200:
return proxy, False, "接続/ステータスコード異常", None
# 匿名性チェック
response_json = response.json()
reported_ip = response_json.get('origin')
# 匿名性ロジック:報告されたIPは実際のIPであってはならず、プロキシのIPと一致する必要があります
if not (reported_ip and reported_ip != real_ip and reported_ip.startswith(proxy_ip)):
return proxy, False, "透明/身元情報漏洩", None
except requests.exceptions.Timeout:
return proxy, False, f"タイムアウト (>{TIMEOUT}s)", None
except Exception as e:
return proxy, False, f"匿名性チェックに失敗しました: {type(e).__name__}", None
# --- 2. 最終的な実際の接続性チェック (FINAL_CONNECTIVITY_TARGETを使用) ---
try:
# Googleに対してテストします。最終テストのためにグローバルTIMEOUTを使用します。
requests.get(FINAL_CONNECTIVITY_TARGET, proxies=proxies, timeout=TIMEOUT, verify=False).raise_for_status()
# 最初のテストの開始からの合計レイテンシを計算します
total_latency = time.time() - start_time
# **速度チェック:** 最大レイテンシ閾値を下回るプロキシのみを受け入れます
if total_latency > MAX_LATENCY:
return proxy, False, f"速度が遅すぎます (合計レイテンシ: {total_latency:.2f}s)", None
# すべてのチェックを通過した場合は成功
return proxy, True, f"高匿名/匿名/高速接続 (合計レイテンシ: {total_latency:.2f}s)", total_latency
except Exception as e:
# 匿名性チェックを通過したが、Googleテストに失敗した場合、ターゲットサイトにブロックされていると見なされます
return proxy, False, f"ターゲットサイトにブロックされました: {type(e).__name__}", None
def batch_check(proxies, real_ip):
"""マルチスレッドを使用してプロキシをバッチテストします。"""
successful_proxies = []
total_proxies = len(proxies)
# ThreadPoolExecutorを使用して同時実行を実現
with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
# すべてのタスクを提出し、実際のIPを各チェック関数に渡します
future_to_proxy = {executor.submit(check_proxy, proxy, real_ip): proxy for proxy in proxies}
# ソート用に (proxy, latency) を格納するリスト
working_proxies_with_latency = []
# 結果を1つずつ取得し、進捗を印刷します
for i, future in enumerate(concurrent.futures.as_completed(future_to_proxy)):
proxy = future_to_proxy[future]
try:
# 関数は4つの値を返します: (proxy, is_working, reason, latency)
proxy, is_working, reason, latency = future.result()
except Exception as exc:
is_working = False
latency = None
reason = f"スレッドエラー: {exc}"
# リアルタイムの進捗を印刷
progress = f"[{i+1}/{total_proxies}]"
print(f"\r{progress} 検出中... 現在の利用可能: {len(working_proxies_with_latency)}", end="", flush=True)
if is_working:
working_proxies_with_latency.append((proxy, latency))
# レイテンシでプロキシをソート(最速のものが最初)
working_proxies_with_latency.sort(key=lambda x: x[1])
successful_proxies = [p[0] for p in working_proxies_with_latency]
# 進捗出力をクリアし、改行
print("\n")
return successful_proxies
if __name__ == '__main__':
start_time = time.time()
# 0. 匿名性比較のためにクライアントの実際のIPを取得します
print("匿名性比較のためにあなたの実際のIPアドレスを取得しています...")
real_ip = get_real_ip()
if not real_ip:
print("❌ 警告:実際のIPを取得できませんでした。匿名性チェックをスキップします。プロキシの質が低下する可能性があります。")
real_ip = '0.0.0.0' # フォールバック値を使用
else:
print(f"✅ あなたの実際のIPアドレスは: {real_ip}")
# 1. プロキシリストを取得
proxy_list = fetch_proxies(PROXY_URL)
if not proxy_list:
print("プログラムを終了します。プロキシリストを取得できませんでした。")
else:
# 新しい速度制限を反映するように出力を更新
print(f"{len(proxy_list)} 個のプロキシをテストし始めます (匿名性を要求、速度 < {MAX_LATENCY:.1f}s, ターゲット: {FINAL_CONNECTIVITY_TARGET})、同時数:{MAX_WORKERS}...")
# 2. バッチチェック
successful_proxies = batch_check(proxy_list, real_ip)
end_time = time.time()
# 3. 結果の要約と保存
print("\n--- 結果の要約 ---")
print(f"総時間:{end_time - start_time:.2f} 秒")
print(f"総プロキシ数:{len(proxy_list)}")
print(f"利用可能(高匿名/匿名/高速)プロキシ数:{len(successful_proxies)}")
if len(proxy_list) > 0:
print(f"成功率:{len(successful_proxies)/len(proxy_list)*100:.2f}%")
output_file = 'working_socks5_proxies.txt'
with open(output_file, 'w') as f:
f.write('\n'.join(successful_proxies))
print(f"\n✅ 利用可能なプロキシがファイルに保存されました:{output_file} (レイテンシ順にソートされ、最速が前にあります)")
実行前にバージョンを確認し、必要なライブラリをインストールしてください。
Python バージョンを確認#
python3 --version
pip(Python パッケージ管理ツール)バージョンを確認#
pip3 --version
コマンドが存在しない場合は、次のコマンドを実行してください:
sudo apt update
sudo apt install python3 python3-pip -y
次に、仮想環境ツールパッケージをインストールします。
sudo apt update
sudo apt install python3-venv -y
主ディレクトリに仮想環境を作成します。
cd ~
python3 -m venv proxy_venv
環境をアクティブにします(アクティブにすると、root の前に (proxy_venv) が表示されます)。
source proxy_venv/bin/activate
環境内でライブラリをインストールします。
pip install requests
インストール後、環境内でスクリプトを実行します。
python proxy_test.py