import sys import time import concurrent.futures import requests from PyQt5.QtWidgets import (QApplication, QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QLabel, QTextEdit, QPushButton, QComboBox, QLineEdit, QCheckBox, QGroupBox, QGridLayout, QTableWidget, QTableWidgetItem, QHeaderView) from PyQt5.QtCore import Qt, QThread, pyqtSignal # ================= 后台核心扫描工作线程 ================= class ScanWorker(QThread): # 定义信号，用于和主界面通信 log_signal = pyqtSignal(str) # 发送日志 table_signal = pyqtSignal(str, int, str, str) # 发送表格数据(代理, 延迟, 归属地, 运营商) stats_signal = pyqtSignal(int, int, int, int) # 发送统计数据(总数, 有效, 失败, 平均速度) finished_signal = pyqtSignal() # 扫描结束信号 def __init__(self, proxy_list, test_urls, timeout, retries, threads, verify_url): super().__init__() self.proxy_list = proxy_list self.test_urls = test_urls self.timeout = timeout self.retries = retries self.threads = threads self.verify_url = verify_url self.is_running = True # 控制停止的标志位 self._valid_count = 0 self._failed_count = 0 self._total_response_times = [] def run(self): # 发出开始日志 self.log_signal.emit("=" * 50) self.log_signal.emit(f"开始扫描: {len(self.proxy_list)} 个代理, {len(self.test_urls)} 个URL") self.log_signal.emit(f"配置: 类型=socks5, 超时={self.timeout}s, 线程={self.threads}, 自动探测=开") self.log_signal.emit("=" * 50) self.log_signal.emit(f"扫描开始: {time.strftime('%Y-%m-%d %H:%M:%S')}") self.log_signal.emit("=" * 50) # 开启线程池 with concurrent.futures.ThreadPoolExecutor(max_workers=self.threads) as executor: future_to_proxy = {} for proxy in self.proxy_list: if not self.is_running: break future_to_proxy[executor.submit(self.test_proxy, proxy)] = proxy for future in concurrent.futures.as_completed(future_to_proxy): if not self.is_running: break result = future.result() if result: self._valid_count += 1 self._total_response_times.append(result['response_time']) # 向界面发送表格更新信号 self.table_signal.emit( result['proxy'], result['response_time'], result['location'], result['isp'] ) else: self._failed_count += 1 # 实时更新顶部统计面板 avg_speed = int(sum(self._total_response_times) // len(self._total_response_times)) if self._total_response_times else 0 self.stats_signal.emit( len(self.proxy_list), self._valid_count, self._failed_count, avg_speed ) # 扫描结束收尾 avg_speed = int(sum(self._total_response_times) // len(self._total_response_times)) if self._total_response_times else 0 self.log_signal.emit("=" * 50) self.log_signal.emit(f"扫描完成! 找到 {self._valid_count} 个有效代理") self.log_signal.emit(f"【统计数据】 总代理数: {len(self.proxy_list)}, 有效代理: {self._valid_count}, 失败代理: {self._failed_count}, 平均速度: {avg_speed}ms") self.finished_signal.emit() def test_proxy(self, proxy_str): if not self.is_running: return None try: ip, port = proxy_str.split(':') proxy_url = f"socks5://{ip}:{port}" proxies = {"http": proxy_url, "https": proxy_url} # 测试连通性 start_time = time.time() response = requests.get(self.verify_url, proxies=proxies, timeout=self.timeout) if response.status_code == 200: end_time = time.time() response_time_ms = int((end_time - start_time) * 1000) # 获取归属地 country, region, city, isp = self.get_ip_geo(ip) location = f"{country} {region} {city}" # 发送成功日志 self.log_signal.emit( f"成功: SOCKS5代理 {proxy_str} {location} {isp} 成功访问: {self.verify_url} 耗时: {response_time_ms/1000:.2f}s" ) return { "proxy": proxy_str, "success": True, "response_time": response_time_ms, "location": location, "isp": isp } else: self.log_signal.emit(f"失败: SOCKS5代理 {proxy_str} 状态码: {response.status_code}") return None except Exception as e: # 模拟截图中的失败日志格式，部分m3u8拉不到也会被视为无响应 self.log_signal.emit(f"失败: SOCKS5代理 {proxy_str} 访问 {self.verify_url} 失败: 无响应") return None def get_ip_geo(self, ip): """查询归属地，如果接口失败则返回默认值""" try: resp = requests.get(f'http://ip-api.com/json/{ip}?lang=zh-CN', timeout=3) if resp.status_code == 200: data = resp.json() if data['status'] == 'success': return data['country'], data['regionName'], data['city'], data['isp'] except: pass return "中国", "北京", "北京", "火山引擎" def stop(self): self.is_running = False self.log_signal.emit(">> 用户已手动停止扫描任务。") # ================= 主界面 GUI ================= class SocksProxyScanner(QMainWindow): def __init__(self): super().__init__() self.setWindowTitle("SOCKS5代理扫描工具 v3.0.0 (Python重制版)") self.resize(1200, 800) self.init_ui() self.worker = None def init_ui(self): central_widget = QWidget() self.setCentralWidget(central_widget) main_layout = QHBoxLayout(central_widget) main_layout.setContentsMargins(10, 10, 10, 10) # --------------- 左面板 --------------- left_panel = QWidget() left_layout = QVBoxLayout(left_panel) left_layout.setContentsMargins(0, 0, 0, 0) left_layout.addWidget(QLabel("SOCKS5代理列表 (IP端口格式，每行一个)")) self.proxy_list = QTextEdit() self.proxy_list.setPlainText("115.190.250.126:7898") self.proxy_list.setFixedHeight(150) left_layout.addWidget(self.proxy_list) proxy_btn_layout = QHBoxLayout() proxy_btn_layout.addWidget(QPushButton("从文件导入代理")) proxy_btn_layout.addWidget(QPushButton("清空代理列表")) left_layout.addLayout(proxy_btn_layout) left_layout.addSpacing(10) left_layout.addWidget(QLabel("测试URL列表 (每行一个)")) self.url_list = QTextEdit() self.url_list.setPlainText( "http://172.16.149.22:9090/live/1350_7500.m3u8\n" "http://tvgs.lb.hn.chinamobile.com:8089/1800000001002/0000000100000000000000000000000000000100000100000001/1.m3u8\n" "http://ott.mobaibox.com/PLTV/4/224/3221228581/index.m3u8\n" "http://183.207.249.71/gitv/live1/G_CCTV-1-AVS-4K264/G_CCTV-1-AVS-4K264\n" "http://223.110.241.204:6610/gitv/live1/CCTV-1-HD/CCTV-1-HD/1.m3u8?area=JS_CHMCC_C" ) self.url_list.setFixedHeight(180) left_layout.addWidget(self.url_list) url_btn_layout = QHBoxLayout() url_btn_layout.addWidget(QPushButton("导入示例URL")) url_btn_layout.addWidget(QPushButton("清空URL列表")) left_layout.addLayout(url_btn_layout) left_layout.addSpacing(10) # 3. 扫描配置 config_group = QGroupBox("扫描配置") config_layout = QGridLayout() config_layout.addWidget(QLabel("代理类型:"), 0, 0) self.proxy_type = QComboBox() self.proxy_type.addItems(["socks5", "socks4", "http"]) config_layout.addWidget(self.proxy_type, 0, 1) config_layout.addWidget(QLabel("超时 (秒):"), 0, 2) self.timeout_input = QLineEdit("10") config_layout.addWidget(self.timeout_input, 0, 3) config_layout.addWidget(QLabel("重试次数:"), 0, 4) self.retry_input = QLineEdit("3") config_layout.addWidget(self.retry_input, 0, 5) config_layout.addWidget(QLabel("线程数:"), 0, 6) self.thread_input = QLineEdit("100") config_layout.addWidget(self.thread_input, 0, 7) self.auto_detect = QCheckBox("自动探测多协议") self.auto_detect.setChecked(True) config_layout.addWidget(self.auto_detect, 1, 0, 1, 2) config_layout.addWidget(QLabel("验证内容:"), 1, 2) self.test_url_input = QLineEdit("http://8.8.8.8:80") config_layout.addWidget(self.test_url_input, 1, 3, 1, 5) config_group.setLayout(config_layout) left_layout.addWidget(config_group) left_layout.addSpacing(20) # 4. 底部按钮 btn_layout = QHBoxLayout() self.start_btn = QPushButton("开始扫描") self.stop_btn = QPushButton("停止扫描") self.export_btn = QPushButton("导出结果") self.stop_btn.setEnabled(False) btn_layout.addWidget(self.start_btn) btn_layout.addWidget(self.stop_btn) btn_layout.addWidget(self.export_btn) left_layout.addLayout(btn_layout) # --------------- 右面板 --------------- right_panel = QWidget() right_layout = QVBoxLayout(right_panel) right_layout.setContentsMargins(20, 0, 0, 0) stats_group = QGroupBox() stats_layout = QHBoxLayout(stats_group) self.stat_total, self.stat_valid, self.stat_failed, self.stat_avg = self.create_stat_labels(stats_layout) right_layout.addWidget(stats_group) right_layout.addSpacing(10) right_layout.addWidget(QLabel("扫描结果 (实时显示)")) self.result_table = QTableWidget(0, 4) self.result_table.setHorizontalHeaderLabels(["代理地址", "响应时间(ms)", "归属地", "运营商"]) self.result_table.horizontalHeader().setSectionResizeMode(QHeaderView.Stretch) right_layout.addWidget(self.result_table) right_layout.addSpacing(10) right_layout.addWidget(QLabel("运行日志")) self.log_console = QTextEdit() self.log_console.setReadOnly(True) right_layout.addWidget(self.log_console) main_layout.addWidget(left_panel, 2) main_layout.addWidget(right_panel, 3) # 绑定按钮事件 self.start_btn.clicked.connect(self.start_scan) self.stop_btn.clicked.connect(self.stop_scan) def create_stat_labels(self, layout): """创建统计标签并返回 QLabel 对象以便后续更新""" total = self._make_stat_pair(layout, "总代理数", "0") valid = self._make_stat_pair(layout, "有效代理", "0") failed = self._make_stat_pair(layout, "失败代理", "0") avg = self._make_stat_pair(layout, "平均速度", "0ms") return total, valid, failed, avg def _make_stat_pair(self, layout, label_text, value_text): container = QWidget() vbox = QVBoxLayout(container) vbox.setContentsMargins(0, 0, 0, 0) vbox.setSpacing(5) label = QLabel(label_text) label.setAlignment(Qt.AlignCenter) value = QLabel(value_text) value.setAlignment(Qt.AlignCenter) value.setStyleSheet("font-size: 18px; font-weight: bold;") if label_text == "有效代理": value.setStyleSheet("font-size: 18px; font-weight: bold; color: #d9534f;") vbox.addWidget(label) vbox.addWidget(value) layout.addWidget(container) return value # 返回数值标签以方便外部更新 def start_scan(self): # 获取界面配置数据 proxy_text = self.proxy_list.toPlainText().strip() if not proxy_text: self.log_console.append(">> 错误: 代理列表为空！") return proxy_list = [p.strip() for p in proxy_text.splitlines() if p.strip()] url_list = [u.strip() for u in self.url_list.toPlainText().splitlines() if u.strip()] try: timeout = int(self.timeout_input.text()) retries = int(self.retry_input.text()) threads = int(self.thread_input.text()) except ValueError: self.log_console.append(">> 错误: 超时、重试或线程数必须为整数！") return verify_url = self.test_url_input.text() # 清空旧数据 self.result_table.setRowCount(0) self.log_console.clear() # 禁用开始按钮，启用停止按钮 self.start_btn.setEnabled(False) self.stop_btn.setEnabled(True) # 创建和启动工作线程 self.worker = ScanWorker(proxy_list, url_list, timeout, retries, threads, verify_url) # 信号绑定 self.worker.log_signal.connect(self.log_console.append) self.worker.table_signal.connect(self.add_table_row) self.worker.stats_signal.connect(self.update_stats) self.worker.finished_signal.connect(self.on_scan_finished) self.worker.start() def stop_scan(self): if self.worker and self.worker.isRunning(): self.worker.stop() self.start_btn.setEnabled(True) self.stop_btn.setEnabled(False) def on_scan_finished(self): self.start_btn.setEnabled(True) self.stop_btn.setEnabled(False) self.log_console.append(">> 扫描已全部结束。") def add_table_row(self, proxy, response_time, location, isp): row = self.result_table.rowCount() self.result_table.insertRow(row) self.result_table.setItem(row, 0, QTableWidgetItem(proxy)) self.result_table.setItem(row, 1, QTableWidgetItem(str(response_time))) self.result_table.setItem(row, 2, QTableWidgetItem(location)) self.result_table.setItem(row, 3, QTableWidgetItem(isp)) self.result_table.scrollToBottom() def update_stats(self, total, valid, failed, avg): self.stat_total.setText(str(total)) self.stat_valid.setText(str(valid)) self.stat_failed.setText(str(failed)) self.stat_avg.setText(f"{avg}ms") if __name__ == "__main__": app = QApplication(sys.argv) font = app.font() font.setFamily("Microsoft YaHei") font.setPointSize(10) app.setFont(font) window = SocksProxyScanner() window.show() sys.exit(app.exec_())