banner
DarianBlog

DarianBlog

达里安博客-专注于技术分享与交流的博客。博客致力于创造一个简单而直接的技术学习平台。
email
telegram

httpとsocks5プロキシのバッチテストスクリプト

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
読み込み中...
文章は、創作者によって署名され、ブロックチェーンに安全に保存されています。