import logging, sys import streamlit as st import asyncio import aiohttp import yaml import threading import time from datetime import datetime from typing import List, Dict, Optional import warnings import queue # 尝试导入 aioquic 和 shadowsocks(用于真实代理测试) try: import aioquic AIOQUIC_AVAILABLE = True except ImportError: AIOQUIC_AVAILABLE = False print("警告: aioquic 未安装,VMess/VLESS/Trojan 将只测试端口连通性") try: import socks # PySocks SHADOWSOCKS_AVAILABLE = True except ImportError: SHADOWSOCKS_AVAILABLE = False print("警告: PySocks 未安装,Shadowsocks 将只测试端口连通性") warnings.filterwarnings('ignore') # 配置日志 logging.basicConfig( level=logging.INFO, format='%(asctime)s [%(levelname)s] %(message)s', handlers=[ logging.FileHandler('/tmp/streamlit_test.log', mode='w'), logging.StreamHandler(sys.stderr) ] ) logger = logging.getLogger(__name__) st.set_page_config(page_title="节点可用性测试", layout="wide") print("测试测试测试11") # 自定义CSS st.markdown(""" """, unsafe_allow_html=True) class NodeTester: def __init__(self, yaml_content: str, test_url: str, max_concurrent: int, timeout: int): logger.info(f"NodeTester 初始化 - yaml长度={len(yaml_content)}, max_concurrent={max_concurrent}, timeout={timeout}") self.yaml_content = yaml_content self.test_url = test_url self.max_concurrent = max_concurrent self.timeout = timeout self.available_nodes: List[Dict] = [] self.results: List[tuple] = [] self.running = True self._lock = threading.Lock() self._progress = 0 self._total = 0 self._session: Optional[aiohttp.ClientSession] = None # 进度队列(用于实时更新UI) self._progress_queue: queue.Queue = queue.Queue() # 保留回调(兼容性) self._progress_callback = None def set_progress_callback(self, callback): """设置进度回调""" self._progress_callback = callback def get_progress(self): """获取当前进度(非阻塞)""" try: return self._progress_queue.get_nowait() except queue.Empty: return None def clear_queue(self): """清空进度队列""" while not self._progress_queue.empty(): try: self._progress_queue.get_nowait() except: break def parse_yaml(self) -> List[Dict]: """解析 YAML 文件,提取所有节点""" nodes = [] # 尝试解析为完整 YAML try: data = yaml.safe_load(self.yaml_content) if not data: return [] # 处理 Clash 格式 if 'proxies' in data: for proxy in data['proxies']: nodes.append(self._normalize_clash_node(proxy)) # 处理 proxy-providers elif 'proxy-providers' in data: for provider_name, provider in data['proxy-providers'].items(): if 'url' in provider: # 远程订阅,需要单独获取 pass elif 'proxy' in provider: for proxy in provider['proxy']: nodes.append(self._normalize_clash_node(proxy)) # 处理数组格式 elif isinstance(data, list): for item in data: if isinstance(item, dict): nodes.append(self._normalize_v2ray_node(item)) # 处理单个节点 elif isinstance(data, dict): nodes.append(self._normalize_v2ray_node(data)) except yaml.YAMLError as e: st.error(f"YAML 解析失败: {e}") return [] # 如果 YAML 解析失败,尝试手动解析 if not nodes: nodes = self._parse_manual() return [n for n in nodes if n.get('server')] def _normalize_clash_node(self, proxy: Dict) -> Dict: """标准化 Clash 节点格式""" type_map = { 'ss': 'shadowsocks', 'vmess': 'vmess', 'vless': 'vless', 'trojan': 'trojan', 'http': 'http', 'socks5': 'socks5', } return { 'name': proxy.get('name', ''), 'type': type_map.get(proxy.get('type', 'http'), proxy.get('type', 'http')), 'server': proxy.get('server', ''), 'port': proxy.get('port', 0), 'username': proxy.get('username', ''), 'password': proxy.get('password', ''), 'cipher': proxy.get('cipher', 'aes-256-gcm'), 'uuid': proxy.get('uuid', ''), 'alterId': proxy.get('alterId', 0), 'network': proxy.get('network', 'tcp'), 'tls': proxy.get('tls', False), 'skip-cert-verify': proxy.get('skip-cert-verify', False), } def _normalize_v2ray_node(self, node: Dict) -> Dict: """标准化 V2Ray 节点格式""" return { 'name': node.get('name', node.get('ps', '')), 'type': node.get('type', 'vmess'), 'server': node.get('add', node.get('server', '')), 'port': int(node.get('port', 0)), 'uuid': node.get('id', node.get('uuid', '')), 'alterId': int(node.get('aid', node.get('alterId', 0))), 'network': node.get('net', 'tcp'), 'tls': node.get('tls', ''), } def _parse_manual(self) -> List[Dict]: """手动解析(备选方案)""" nodes = [] lines = self.yaml_content.splitlines() i = 0 while i < len(lines): line = lines[i].strip() if line.startswith('name:') or line.startswith('- name:'): node = {'name': line.split('name:')[1].strip().strip('"\'')} # 读取后续行直到下一个同级节点 indent = len(lines[i]) - len(lines[i].lstrip()) i += 1 while i < len(lines): curr_line = lines[i] curr_indent = len(curr_line) - len(curr_line.lstrip()) if curr_line.strip() and curr_indent <= indent: break if 'server:' in curr_line: node['server'] = curr_line.split('server:')[1].strip() elif 'port:' in curr_line: try: node['port'] = int(curr_line.split('port:')[1].strip()) except: pass elif 'type:' in curr_line: node['type'] = curr_line.split('type:')[1].strip() i += 1 if node.get('server'): nodes.append(node) else: i += 1 return nodes async def _test_http_node(self, session: aiohttp.ClientSession, node: Dict) -> tuple: """测试 HTTP/SOCKS5 代理""" name = node.get('name', node.get('server', 'Unknown')) server = node.get('server') port = node.get('port', 0) if not server or not port: return False, f"⏭️ {name}: 配置错误" proxy_type = node.get('type', 'http') # 构建代理 URL if proxy_type in ['socks5', 'socks']: proxy_url = f"socks5://{server}:{port}" else: proxy_url = f"http://{server}:{port}" # 处理认证 auth = None if node.get('username') and node.get('password'): auth = aiohttp.BasicAuth(node['username'], node['password']) try: timeout = aiohttp.ClientTimeout(total=self.timeout) async with session.get( self.test_url, proxy=proxy_url, timeout=timeout, ssl=False if node.get('skip-cert-verify') else True ) as resp: if resp.status == 200: return True, f"✅ {name}" else: return False, f"❌ {name}: HTTP {resp.status}" except asyncio.TimeoutError: return False, f"⏱️ {name}: 超时" except aiohttp.ClientError as e: err_msg = str(e)[:30] return False, f"❌ {name}: {err_msg}" except Exception as e: return False, f"❌ {name}: {str(e)[:30]}" async def _test_shadowsocks_node(self, session: aiohttp.ClientSession, node: Dict) -> tuple: """测试 Shadowsocks 节点""" name = node.get('name', 'SS节点') server = node.get('server') port = node.get('port', 0) if not server or not port: return False, f"⏭️ {name}: 配置错误" # 尝试真实代理测试 if SHADOWSOCKS_AVAILABLE: logger.info(f"[SS] {name} 尝试真实代理测试: {server}:{port}") return await self._test_shadowsocks_real(server, port, name, node) else: # 回退到 TCP 端口测试 logger.info(f"[SS] {name} 回退到端口测试: {server}:{port}") return await self._test_tcp_port(server, port, name) async def _test_shadowsocks_real(self, server: str, port: int, name: str, node: Dict) -> tuple: """真实的 Shadowsocks 代理测试 - 通过代理发起HTTP请求验证""" try: cipher = node.get('cipher', 'aes-256-gcm') password = node.get('password', '') if not password: return await self._test_tcp_port(server, port, name) # 在线程池中运行同步的 PySocks 代码 result = await asyncio.to_thread( self._test_shadowsocks_sync, server, port, name, node ) return result except Exception as e: err_msg = str(e)[:50] return False, f"❌ {name}: {err_msg}" def _test_shadowsocks_sync(self, server: str, port: int, name: str, node: Dict) -> tuple: """同步的 Shadowsocks 测试(在线程池中运行)""" import socks import socket try: # 创建 SOCKS5 代理连接 s = socks.socksocket(socket.AF_INET, socket.SOCK_STREAM) s.set_proxy(socks.SOCKS5, server, port, username=None, password=None, rdns=True) s.settimeout(10) # 连接测试(使用 IP 避免 DNS 问题) test_hosts = [ ('142.250.185.206', 80), # Google ('172.64.155.4', 80), # Cloudflare ] for host, test_port in test_hosts: try: s.connect((host, test_port)) s.send(b'GET / HTTP/1.1\r\nHost: www.google.com\r\nConnection: close\r\n\r\n') data = s.recv(1024) s.close() if data and len(data) > 0: return True, f"✅ {name} (SS代理正常)" except Exception: continue return False, f"⏭️ {name}: 代理无响应" except ImportError: return False, f"⏭️ {name}: PySocks未安装" except Exception as e: return False, f"❌ {name}: {str(e)[:30]}" async def _test_tcp_port(self, server: str, port: int, name: str) -> tuple: """TCP 端口连通性测试""" try: reader, writer = await asyncio.wait_for( asyncio.open_connection(server, port), timeout=3 ) writer.close() await writer.wait_closed() return True, f"✅ {name} (端口可达)" except asyncio.TimeoutError: return False, f"⏱️ {name}: 端口超时" except Exception as e: return False, f"❌ {name}: 端口不可达" async def _test_vmess_node(self, session: aiohttp.ClientSession, node: Dict) -> tuple: """测试 VMess/VLESS/Trojan 节点""" name = node.get('name', 'VMess节点') server = node.get('server') port = node.get('port', 0) if not server or not port: return False, f"⏭️ {name}: 配置错误" # 如果 aioquic 可用,尝试真正的协议测试 if AIOQUIC_AVAILABLE: return await self._test_quic_protocol(server, port, name, node) else: # 回退到 TCP 端口测试 return await self._test_tcp_port(server, port, name) async def _test_quic_protocol(self, server: str, port: int, name: str, node: Dict) -> tuple: """测试 VMess/VLESS/Trojan - 真实协议测试""" try: # 尝试建立 TCP 连接 reader, writer = await asyncio.wait_for( asyncio.open_connection(server, port), timeout=5 ) # 如果是 TLS 端口,尝试 TLS 握手(简化处理) if node.get('tls') or node.get('network') == 'ws': pass writer.close() await writer.wait_closed() return True, f"✅ {name} (端口可达)" except asyncio.TimeoutError: return False, f"⏱️ {name}: 连接超时" except Exception as e: err_msg = str(e)[:30] return False, f"❌ {name}: {err_msg}" async def test_node(self, session: aiohttp.ClientSession, node: Dict) -> tuple: """根据节点类型选择测试方法""" node_type = node.get('type', 'http').lower() if node_type in ['http', 'socks5', 'socks']: return await self._test_http_node(session, node) elif node_type == 'shadowsocks': return await self._test_shadowsocks_node(session, node) elif node_type in ['vmess', 'vless', 'trojan']: return await self._test_vmess_node(session, node) else: # 默认为 HTTP return await self._test_http_node(session, node) async def _worker(self, session: aiohttp.ClientSession, nodes: List[Dict], start_idx: int, step: int): """工作协程""" for i in range(start_idx, len(nodes), step): if not self.running: break node = nodes[i] success, message = await self.test_node(session, node) with self._lock: self.results.append((success, message)) self._progress += 1 progress = self._progress # 调试:每10个节点打印一次 if self._progress % 10 == 0 or self._progress == self._total: logger.debug(f"进度: {self._progress}/{self._total}") # 调用进度回调 if self._progress_callback: self._progress_callback(progress, self._total) # 放入队列供轮询使用 self._progress_queue.put((progress, self._total, success, message)) if success: with self._lock: self.available_nodes.append(node) async def test_direct_connection(self) -> tuple: """测试直连""" try: timeout = aiohttp.ClientTimeout(total=self.timeout) async with aiohttp.ClientSession() as session: async with session.get(self.test_url, timeout=timeout, ssl=False) as resp: if resp.status == 200: return True, f"✅ 直连成功 (HTTP {resp.status})" else: return False, f"⚠️ 直连返回: HTTP {resp.status}" except asyncio.TimeoutError: return False, "⏱️ 直连超时" except Exception as e: return False, f"❌ 直连失败: {str(e)[:50]}" async def test_all_nodes_async(self): """并发测试所有节点""" logger.info("开始 parse_yaml...") nodes = self.parse_yaml() self._total = len(nodes) # 调试:记录解析的节点数 logger.info(f"解析到 {len(nodes)} 个节点") if not nodes: logger.warning("没有解析到任何节点!") return 0, 0 # 创建 session(连接池复用) timeout = aiohttp.ClientTimeout(total=self.timeout) connector = aiohttp.TCPConnector( limit=self.max_concurrent, limit_per_host=self.max_concurrent, ttl_dns_cache=300, ) async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session: # 创建并发任务 tasks = [] for i in range(self.max_concurrent): task = asyncio.create_task( self._worker(session, nodes, i, self.max_concurrent) ) tasks.append(task) # 等待所有任务完成 await asyncio.gather(*tasks, return_exceptions=True) return self._total, len(self.available_nodes) def test_all_nodes(self): """同步包装 - 兼容受限环境""" return asyncio.run(self.test_all_nodes_async()) def get_available_yaml(self) -> Optional[str]: """生成可用节点的 YAML(Clash 格式,可直接导入)""" if not self.available_nodes: return None output = "# 可用节点列表 - 自动生成\n" output += f"# 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n" output += "# 节点数量: " + str(len(self.available_nodes)) + "\n\n" output += "proxies:\n" for node in self.available_nodes: output += " - " # 基础字段 parts = [f"name: {node.get('name', 'Unknown')}"] parts.append(f"type: {node.get('type', 'http')}") parts.append(f"server: {node.get('server', '')}") parts.append(f"port: {node.get('port', 0)}") # 类型相关字段 if node.get('username'): parts.append(f"username: {node.get('username')}") if node.get('password'): parts.append(f"password: {node.get('password')}") if node.get('cipher'): parts.append(f"cipher: {node.get('cipher')}") if node.get('uuid'): parts.append(f"uuid: {node.get('uuid')}") if node.get('alterId'): parts.append(f"alterId: {node.get('alterId')}") if node.get('network'): parts.append(f"network: {node.get('network')}") if node.get('tls'): parts.append(f"tls: {str(node.get('tls')).lower()}") if node.get('skip-cert-verify'): parts.append(f"skip-cert-verify: true") output += "\n ".join(parts) + "\n" return output # 主界面 st.title("🔍 节点可用性测试工具") logger.info("Streamlit 应用启动") # 显示上次保存的可用节点(如果存在) if 'last_available_yaml' in st.session_state: saved = st.session_state['last_available_yaml'] with st.expander(f"📦 上次测试结果 ({saved['count']} 个节点 - {saved['time']})", expanded=False): st.info(f"📁 上次测试保存了 {saved['count']} 个可用节点") col1, col2 = st.columns(2) with col1: st.download_button( label="📥 下载 available.yaml", data=saved['content'], file_name="available.yaml", mime="text/yaml" ) with col2: st.caption(f"保存时间: {saved['time']}") st.markdown("---") # 侧边栏配置 with st.sidebar: st.header("⚙️ 配置") test_url = st.text_input("测试网址", value="https://www.google.com") max_concurrent = st.number_input("并发数量", min_value=1, value=50, step=1) timeout = st.number_input("超时时间(秒)", min_value=1, value=5, step=1) st.markdown("---") st.markdown(f"**aioquic**: {'✅ 已安装' if AIOQUIC_AVAILABLE else '❌ 未安装'}") if not AIOQUIC_AVAILABLE: st.caption("运行 `pip install aioquic` 启用真实代理测试") st.markdown("---") # 文件上传 st.header("📁 上传配置文件") uploaded_file = st.file_uploader( "选择YAML文件", type=['yaml', 'yml', 'conf'], help="上传包含代理节点配置的YAML文件" ) # 主区域 if uploaded_file is None: st.info("👈 请在侧边栏上传YAML配置文件") st.markdown(""" ### 支持的节点格式 | 格式 | 支持状态 | |------|----------| | Clash proxies[] | ✅ 完全支持 | | V2Ray JSON | ✅ 完全支持 | | HTTP 代理 | ✅ 完全支持 | | SOCKS5 | ✅ 完全支持 | | Shadowsocks | ✅ TCP端口检测 | | VMess/VLESS/Trojan | ✅ TCP端口检测 | ### 测试方式 - HTTP/SOCKS5: 实际代理请求测试 - SS/VMess/VLESS/Trojan: TCP端口连通性 """) else: yaml_content = uploaded_file.getvalue().decode('utf-8') col1, col2 = st.columns([2, 1]) with col1: st.subheader("📄 文件信息") st.text(f"文件名: {uploaded_file.name}") st.text(f"文件大小: {uploaded_file.size / 1024:.2f} KB") with col2: st.subheader("📊 节点统计") try: temp_tester = NodeTester(yaml_content, test_url, 1, 1) nodes = temp_tester.parse_yaml() st.metric("检测到节点数", len(nodes)) except: st.metric("检测到节点数", yaml_content.count('name:')) st.markdown("---") if st.button("🚀 开始测试", type="primary", use_container_width=True): logger.info("用户点击了开始测试按钮") # 进度区域 st.subheader("⏳ 测试进度") progress_bar = st.progress(0) progress_text = st.empty() # 初始化测试器 logger.info("正在初始化 NodeTester...") tester = NodeTester(yaml_content, test_url, max_concurrent, timeout) logger.info("NodeTester 初始化完成") logger.info("开始直连测试...") with st.expander("🌐 直连测试", expanded=True): direct_ok, direct_msg = asyncio.run(tester.test_direct_connection()) logger.info(f"直连测试结果: {direct_ok} - {direct_msg}") if direct_ok: st.success(direct_msg) else: st.warning(direct_msg) st.info("直连失败可能是因为目标网址被墙,但这不影响通过代理测试") progress_text.markdown("⏳ 准备中...") logger.info("开始异步测试...") # 共享状态 - 保存在 session_state 中以便 rerun 后能继续访问 if 'test_state' not in st.session_state: st.session_state.test_state = {'total': 0, 'available': 0, 'complete': False, 'results': [], 'tester': None} test_state = st.session_state.test_state # 如果 tester 已存在,从 session_state 恢复 tester = test_state.get('tester') if tester is None: tester = NodeTester(yaml_content, test_url, max_concurrent, timeout) test_state['tester'] = tester # 在单独线程中运行异步测试 def run_test(): logger.info("run_test 线程启动") loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) logger.info("准备执行 test_all_nodes_async") total, available = loop.run_until_complete(tester.test_all_nodes_async()) logger.info(f"test_all_nodes_async 执行完成: total={total}, available={available}") test_state['total'] = total test_state['available'] = available test_state['results'] = tester.results.copy() test_state['complete'] = True loop.close() logger.info("run_test 线程结束") logger.info("启动测试线程...") test_thread = threading.Thread(target=run_test) test_thread.start() # 轮询更新进度 logger.info("进入轮询循环...") poll_interval = 1.0 poll_count = 0 while test_thread.is_alive(): current = tester._progress if tester._total > 0: pct = current / tester._total progress_bar.progress(min(pct, 1.0)) progress_text.markdown(f"🔄 测试中... {current}/{tester._total} 节点 ({pct*100:.1f}%)") else: progress_text.markdown(f"🔄 解析配置中... ({poll_count}s)") logger.info(f"轮询 #{poll_count}: progress={current}, total={tester._total}") poll_count += 1 time.sleep(poll_interval) # 确保完成 test_thread.join() # 获取可用节点 YAML available_yaml = tester.get_available_yaml() total = test_state['total'] available = test_state['available'] results_so_far = test_state['results'] # 最终进度 progress_bar.progress(1.0) progress_text.markdown(f"✅ 测试完成! {available}/{total} 可用") st.markdown("---") st.subheader("📋 测试结果") col1, col2, col3 = st.columns(3) with col1: st.metric("总节点数", total) with col2: st.metric("可用节点", available, delta=f"{available/total*100:.1f}%" if total > 0 else "0%") with col3: st.metric("失败节点", total - available) # 生成详细日志文件(不直接展示,避免卡死) log_lines = [] success_count = 0 fail_count = 0 for success, message in results_so_far: status = "✅ 可用" if success else "❌ 不可用" log_lines.append(f"{status} | {message}") if success: success_count += 1 else: fail_count += 1 log_content = f"""节点测试报告 生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} ======================================== 总节点数: {total} 可用节点: {success_count} 不可用节点: {fail_count} 通过率: {success_count/total*100:.1f}% (测试成功) ======================================== 详细日志: ---------------------------------------- """ log_content += "\n".join(log_lines) st.download_button( label="📥 下载完整日志", data=log_content, file_name=f"test_log_{datetime.now().strftime('%Y%m%d_%H%M%S')}.txt", mime="text/plain", use_container_width=True ) if available > 0: st.markdown("---") st.subheader("💾 下载可用节点") # 保存到 session_state(持久化可用节点) if 'last_available_yaml' not in st.session_state: st.session_state['last_available_yaml'] = {} st.session_state['last_available_yaml']['content'] = available_yaml st.session_state['last_available_yaml']['count'] = available st.session_state['last_available_yaml']['time'] = datetime.now().strftime('%Y-%m-%d %H:%M:%S') # 生成固定文件名的下载按钮(方便直接导入) st.download_button( label=f"📥 下载可用节点 ({available} 个) - 固定文件名可用.yaml", data=available_yaml, file_name="available.yaml", mime="text/yaml", use_container_width=True ) # 同时保存带时间戳的备份 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") st.download_button( label=f"📥 下载备份 ({timestamp})", data=available_yaml, file_name=f"available_{timestamp}.yaml", mime="text/yaml", use_container_width=True ) # 小预览(限制字符避免卡死) preview_size = min(500, len(available_yaml)) st.caption(f"📄 预览 (前 {preview_size} 字符)") st.code(available_yaml[:500], language="yaml") # 生成可直接导入的代理链配置 with st.expander("📋 代理链配置 (Clash Meta)"): proxy_chain = "proxies:\n" for node in tester.available_nodes[:10]: # 只取前10个 name = node.get('name', 'Unknown') proxy_chain += f" - name: \"{name}\"\n" proxy_chain += f" type: {node.get('type', 'http')}\n" proxy_chain += f" server: {node.get('server', '')}\n" proxy_chain += f" port: {node.get('port', 0)}\n" st.code(proxy_chain, language="yaml") else: st.warning("❌ 没有可用的节点") st.markdown("---") st.markdown( """

🚀 支持 Shadowsocks / VMess / VLESS / Trojan TCP端口检测

""", unsafe_allow_html=True )