vxray节点 用Python通过xray_core检测

import requests
import json
import os
import subprocess
import time
import platform
import zipfile
import socket
import stat
from urllib.parse import urlparse, parse_qs

# --- 配置区域 ---
TEST_URL = "http://www.google.com/generate_204"
XRAY_PATH = "xray_core"
OUTPUT_FILE = "available_nodes.txt"

# --- 1. 智能识别系统环境 (已适配 Android) ---
uname_info = platform.uname()
system_name = platform.system().lower()
machine_name = platform.machine().lower()

print(f"📱 检测系统信息: {system_name} | 架构: {machine_name}")

IS_ANDROID = "android" in str(uname_info).lower() or "aarch64" in machine_name
IS_WINDOWS = "windows" in system_name

# 确定文件名和下载链接
if IS_WINDOWS:
    XRAY_BIN_NAME = "xray.exe"
    DOWNLOAD_ASSET = "Xray-windows-64.zip"
elif IS_ANDROID or "aarch64" in machine_name:
    # 安卓手机通常是 arm64 架构
    XRAY_BIN_NAME = "xray"
    DOWNLOAD_ASSET = "Xray-linux-arm64-v8a.zip" 
else:
    # 默认 Linux x64 (PC/服务器)
    XRAY_BIN_NAME = "xray"
    DOWNLOAD_ASSET = "Xray-linux-64.zip"

CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
XRAY_BIN = os.path.join(CURRENT_DIR, XRAY_PATH, XRAY_BIN_NAME)
OUTPUT_PATH = os.path.join(CURRENT_DIR, OUTPUT_FILE)

def get_free_port():
    """获取本地空闲端口"""
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            # 安卓 Termux 环境建议绑定 127.0.0.1
            s.bind(('127.0.0.1', 0))
            return s.getsockname()[1]
    except:
        import random
        return random.randint(20000, 50000)

def install_xray():
    """下载并安装 Xray"""
    if not os.path.exists(XRAY_PATH):
        os.makedirs(XRAY_PATH)
    
    if os.path.exists(XRAY_BIN):
        if not IS_WINDOWS:
            st = os.stat(XRAY_BIN)
            os.chmod(XRAY_BIN, st.st_mode | stat.S_IEXEC)
        return True

    print(f"⬇️ 正在下载适配手机的核心: {DOWNLOAD_ASSET} ...")
    
    # 增加加速源
    urls = [
        f"https://mirror.ghproxy.com/https://github.com/XTLS/Xray-core/releases/download/v1.8.4/{DOWNLOAD_ASSET}",
        f"https://github.com/XTLS/Xray-core/releases/download/v1.8.4/{DOWNLOAD_ASSET}"
    ]

    download_success = False
    zip_path = "xray_temp.zip"

    for url in urls:
        try:
            print(f"尝试下载: {url}")
            headers = {'User-Agent': 'Mozilla/5.0'}
            r = requests.get(url, headers=headers, stream=True, timeout=15)
            if r.status_code == 200:
                with open(zip_path, 'wb') as f:
                    for chunk in r.iter_content(chunk_size=8192):
                        f.write(chunk)
                download_success = True
                break
        except Exception as e:
            print(f"下载失败: {e}")
            continue
            
    if not download_success:
        print("❌ 下载失败，请检查网络。")
        return False

    try:
        print("📦 解压中...")
        with zipfile.ZipFile(zip_path, 'r') as zip_ref:
            zip_ref.extractall(XRAY_PATH)
        os.remove(zip_path)
        
        if os.path.exists(XRAY_BIN):
            if not IS_WINDOWS:
                print("🔒 赋予执行权限...")
                st = os.stat(XRAY_BIN)
                os.chmod(XRAY_BIN, st.st_mode | stat.S_IEXEC)
            return True
        else:
            print("❌ 解压后未找到文件")
            return False
    except Exception as e:
        print(f"❌ 解压错误: {e}")
        return False

def vless_to_config(node_str, local_port):
    # (保持原有的转换逻辑不变)
    try:
        parsed = urlparse(node_str)
        if not node_str.startswith("vless://"): return None
        if '@' in parsed.netloc:
            user_id, host_port = parsed.netloc.split('@', 1)
        else: return None

        if ':' in host_port:
            address = host_port.rpartition(':')[0]
            port = int(host_port.rpartition(':')[2])
        else:
            address, port = host_port, 443

        qs = parse_qs(parsed.query)
        security = qs.get('security', ['none'])[0]
        type_network = qs.get('type', ['tcp'])[0]
        sni = qs.get('sni', [''])[0]
        host = qs.get('host', [''])[0]
        path = qs.get('path', ['/'])[0]
        fp = qs.get('fp', ['chrome'])[0]
        pbk = qs.get('pbk', [''])[0]
        sid = qs.get('sid', [''])[0]
        service_name = qs.get('serviceName', [''])[0]
        server_name = sni if sni else (host if host else address)

        outbound = {
            "protocol": "vless",
            "settings": {
                "vnext": [{"address": address, "port": port, "users": [{"id": user_id, "encryption": "none"}]}]
            },
            "streamSettings": {
                "network": type_network,
                "security": security
            }
        }
        if security == 'tls':
            outbound['streamSettings']['tlsSettings'] = {"serverName": server_name, "fingerprint": fp}
        elif security == 'reality':
            outbound['streamSettings']['realitySettings'] = {"serverName": server_name, "publicKey": pbk, "shortId": sid, "fingerprint": fp}
        
        if type_network == 'ws':
            outbound['streamSettings']['wsSettings'] = {"path": path, "headers": {"Host": host if host else server_name}}
        elif type_network == 'grpc':
            outbound['streamSettings']['grpcSettings'] = {"serviceName": service_name}

        config = {
            "log": {"loglevel": "none"},
            "inbounds": [{"port": local_port, "protocol": "socks", "settings": {"auth": "noauth", "udp": True}}],
            "outbounds": [outbound]
        }
        return config
    except: return None

def check_node_real(node_str):
    current_port = get_free_port()
    config = vless_to_config(node_str, current_port)
    
    if not config: return False, 0

    config_path = os.path.join(XRAY_PATH, f"config_{current_port}.json")
    with open(config_path, 'w') as f: json.dump(config, f)

    process = None
    try:
        # 安卓端启动无需 startupinfo
        process = subprocess.Popen([XRAY_BIN, "run", "-c", config_path], 
                                 stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        
        time.sleep(1.5) # 手机性能稍弱，给 1.5秒

        proxies = {
            'http': f'socks5h://127.0.0.1:{current_port}',
            'https': f'socks5h://127.0.0.1:{current_port}'
        }
        
        start = time.time()
        resp = requests.get(TEST_URL, proxies=proxies, timeout=5)
        latency = time.time() - start
        
        if resp.status_code in [200, 204]:
            return True, latency
        else:
            return False, 0

    except Exception:
        return False, 0
    finally:
        if process:
            try:
                process.terminate()
                process.wait(timeout=1)
            except:
                try: process.kill()
                except: pass
        if os.path.exists(config_path):
            try: os.remove(config_path)
            except: pass

def main(url_list):
    if not install_xray(): return

    print("\n📡 正在获取节点...")
    all_nodes = []
    for url in url_list:
        try:
            if url.startswith("http"): content = requests.get(url, timeout=10).text
            else: 
                if os.path.exists(url):
                    with open(url, 'r', encoding='utf-8') as f: content = f.read()
                else: continue
            
            for line in content.split('\n'):
                if line.strip().startswith("vless://"): all_nodes.append(line.strip())
        except: pass

    unique_nodes = list(set(all_nodes))
    print(f"📝 共 {len(unique_nodes)} 个节点，结果将保存在: {OUTPUT_FILE}\n")

    with open(OUTPUT_PATH, 'w', encoding='utf-8') as f: f.write("")

    for i, node in enumerate(unique_nodes):
        alias = node.split('#')[-1] if '#' in node else "无名"
        print(f"[{i+1}] {alias[:10]:<10} ... ", end="", flush=True)
        
        is_ok, latency = check_node_real(node)
        
        if is_ok:
            print(f"✅ {int(latency*1000)}ms")
            with open(OUTPUT_PATH, 'a', encoding='utf-8') as f:
                f.write(node + '\n')
        else:
            print(f"❌")

if __name__ == "__main__":
    urls = ['https://xxxxxxx.txt']
    main(urls)