import requests
import pandas as pd
import time
from concurrent.futures import ThreadPoolExecutor, as_completed

def check_port(ip: str, port: int) -> str:
    """
    向给定IP和端口发送GET请求，返回特定响应或超时指示。

    参数:
    ip: 表示IP地址的字符串。
    port: 表示端口号的整数。

    返回:
    表示结果的字符串（'https_error' 或 'timeout'）。
    """
    url = f"https://{ip}:{port}/cdn-cgi/trace"
    try:
        # 禁用重定向，并设置超时为 1.5 秒
        response = requests.get(url, timeout=1.5, allow_redirects=False)
        if ("400 The plain HTTP request was sent to HTTPS port" in response.text and "cloudflare" in response.text) or "visit_scheme=http" in response.text:
            return 'https_error'
        return None
    except requests.exceptions.Timeout:
        return 'timeout'
    except requests.exceptions.RequestException:
        return None

def scan_ports(file_path: str) -> None:
    """
    从CSV文件中逐批读取IP和端口进行扫描，并将特定结果记录到文件中。

    参数:
    file_path: CSV文件的路径。
    """
    df = pd.read_csv(file_path, encoding='ISO-8859-1')
    with open('06-提取反代了CF的ip及端口.txt', 'w') as file:
        for start in range(0, len(df), 100):
            batch = df[start:start + 100]
            with ThreadPoolExecutor() as executor:
                futures = {
                    executor.submit(check_port, row[0], row[1]): (row[0], row[1])
                    for row in batch.itertuples(index=False)
                }
                for future in as_completed(futures):
                    ip_port = futures[future]
                    result = future.result()
                    if result == 'https_error':
                        file.write(f"{ip_port[0]}:{ip_port[1]}\n")
            time.sleep(0.01)  # 间隔 10 毫秒，防止 CPU 被干爆

# 使用示例
file_path = './05-可视化扫描结果.csv'  # 替换为CSV文件的实际路径
scan_ports(file_path)

import subprocess
import threading
from queue import Queue
import os
import tempfile

def run_command(ip_port: str, output_file: str) -> None:
    ip, port = ip_port.split(':')
    
    # 【！！！如果哪天下面这个测速链接不能用了，请换成自己的测速链接！！！】
    command = f"CloudflareST.exe -ip {ip} -tp {port} -url https://cloudflare.cdn.openbsd.org/pub/OpenBSD/7.3/src.tar.gz -o "" -tl 5000 -dn 20 -p 20"
    
    with open(output_file, "w") as file:
        process = subprocess.Popen(command, shell=True, stdout=file, stderr=subprocess.STDOUT,
                                   creationflags=subprocess.CREATE_NO_WINDOW)

        try:
            process.wait(timeout=20)  # 并行测速 50 个确定反代了 CloudFlare 的【ip:port】时，总的超时时间，单位（秒）
        except subprocess.TimeoutExpired:
            pass

        process.send_signal(subprocess.signal.CTRL_C_EVENT)

def worker(queue: Queue, output_dir: str) -> None:
    while not queue.empty():
        ip_port = queue.get()
        temp_file = os.path.join(output_dir, f"output_{ip_port.replace(':', '_')}.txt")
        run_command(ip_port, temp_file)
        queue.task_done()

def main() -> None:
    input_file = '06-提取反代了CF的ip及端口.txt'
    output_file = '08-针对性测速结果完整文件.txt'
    max_threads = 50  # 并行测速的数量

    queue = Queue()
    threads = []
    output_dir = tempfile.mkdtemp()
    print(f"临时测速结果文件夹路径: {output_dir}")

    with open(input_file, 'r') as file:
        for line in file:
            queue.put(line.strip())

    for _ in range(min(max_threads, queue.qsize())):
        thread = threading.Thread(target=worker, args=(queue, output_dir))
        thread.start()
        threads.append(thread)

    for thread in threads:
        thread.join()

    # 顺序合并临时文件
    temp_files = sorted(os.listdir(output_dir))
    with open(output_file, 'w', encoding='utf-8') as outfile:
        for filename in temp_files:
            with open(os.path.join(output_dir, filename), 'r', encoding='utf-8') as infile:
                outfile.write(infile.read())
                outfile.write('\n')

if __name__ == "__main__":
    main()