"""主窗口 - PyQt6 GUI界面""" import sys from PyQt6.QtWidgets import ( QMainWindow, QWidget, QVBoxLayout, QHBoxLayout, QPushButton, QLabel, QTableWidget, QTableWidgetItem, QTextEdit, QGroupBox, QFormLayout, QLineEdit, QSpinBox, QComboBox, QCheckBox, QMessageBox, QHeaderView, QProgressBar, QTabWidget, QDialog, QDialogButtonBox, QFileDialog ) from PyQt6.QtCore import Qt, QTimer, pyqtSignal, QThread from PyQt6.QtGui import QFont, QColor from loguru import logger import yaml import os from core import ProxyManager from core.models import ProxyInfo, ProxyStatus class WorkerThread(QThread): """工作线程 - 处理耗时操作""" progress_signal = pyqtSignal(str) finished_signal = pyqtSignal(bool, str) def __init__(self, task_func, *args, **kwargs): super().__init__() self.task_func = task_func self.args = args self.kwargs = kwargs def run(self): try: result = self.task_func(*self.args, **self.kwargs) self.finished_signal.emit(True, str(result)) except Exception as e: self.finished_signal.emit(False, str(e)) class ConfigDialog(QDialog): """配置对话框""" def __init__(self, config: dict, parent=None): super().__init__(parent) self.config = config.copy() self.setWindowTitle("配置设置") self.setFixedSize(600, 500) self.init_ui() def init_ui(self): layout = QVBoxLayout(self) # 代理源设置 source_group = QGroupBox("代理源设置") source_layout = QFormLayout() self.url_edit = QLineEdit(self.config.get('proxy_sources', {}).get('auto_fetch', {}).get('url', '')) source_layout.addRow("自动获取URL:", self.url_edit) self.pages_spin = QSpinBox() self.pages_spin.setRange(1, 20) self.pages_spin.setValue(self.config.get('proxy_sources', {}).get('auto_fetch', {}).get('pages', 3)) source_layout.addRow("抓取页数:", self.pages_spin) self.refresh_interval_spin = QSpinBox() self.refresh_interval_spin.setRange(1, 60) self.refresh_interval_spin.setValue(self.config.get('proxy_sources', {}).get('auto_fetch', {}).get('refresh_interval', 10)) source_layout.addRow("刷新间隔(分钟):", self.refresh_interval_spin) self.local_file_checkbox = QCheckBox("启用本地文件") self.local_file_checkbox.setChecked(self.config.get('proxy_sources', {}).get('local_file', {}).get('enabled', True)) source_layout.addRow("", self.local_file_checkbox) source_group.setLayout(source_layout) layout.addWidget(source_group) # 轮转策略 rotation_group = QGroupBox("轮转策略") rotation_layout = QFormLayout() self.mode_combo = QComboBox() self.mode_combo.addItems(["manual", "auto"]) self.mode_combo.setCurrentText(self.config.get('rotation', {}).get('mode', 'manual')) rotation_layout.addRow("切换模式:", self.mode_combo) self.switch_interval_spin = QSpinBox() self.switch_interval_spin.setRange(30, 3600) self.switch_interval_spin.setValue(self.config.get('rotation', {}).get('auto_switch_interval', 300)) rotation_layout.addRow("自动切换间隔(秒):", self.switch_interval_spin) self.latency_threshold_spin = QSpinBox() self.latency_threshold_spin.setRange(100, 2000) self.latency_threshold_spin.setValue(self.config.get('rotation', {}).get('latency_threshold', 500)) rotation_layout.addRow("延迟阈值(ms):", self.latency_threshold_spin) rotation_group.setLayout(rotation_layout) layout.addWidget(rotation_group) # 输出设置 output_group = QGroupBox("输出设置") output_layout = QFormLayout() self.host_edit = QLineEdit(self.config.get('output', {}).get('host', '127.0.0.1')) output_layout.addRow("监听地址:", self.host_edit) self.port_spin = QSpinBox() self.port_spin.setRange(1024, 65535) self.port_spin.setValue(self.config.get('output', {}).get('port', 8745)) output_layout.addRow("监听端口:", self.port_spin) output_group.setLayout(output_layout) layout.addWidget(output_group) # 按钮 button_box = QDialogButtonBox( QDialogButtonBox.StandardButton.Ok | QDialogButtonBox.StandardButton.Cancel ) button_box.accepted.connect(self.accept) button_box.rejected.connect(self.reject) layout.addWidget(button_box) def get_config(self) -> dict: """获取配置""" return { 'proxy_sources': { 'auto_fetch': { 'enabled': True, 'url': self.url_edit.text(), 'pages': self.pages_spin.value(), 'refresh_interval': self.refresh_interval_spin.value(), 'output_file': 'proxy.json' }, 'local_file': { 'enabled': self.local_file_checkbox.isChecked(), 'path': 'local.json' } }, 'rotation': { 'mode': self.mode_combo.currentText(), 'auto_switch_interval': self.switch_interval_spin.value(), 'latency_threshold': self.latency_threshold_spin.value() }, 'output': { 'host': self.host_edit.text(), 'port': self.port_spin.value() } } class MainWindow(QMainWindow): """主窗口""" def __init__(self, config_path='config.yaml'): super().__init__() self.config_path = config_path self.config = self.load_config() self.proxy_manager = None self.worker_thread = None self.setWindowTitle("WhereAmI - 多协议轮转代理工具") self.resize(1200, 800) self.init_ui() self.init_proxy_manager() def load_config(self) -> dict: """加载配置文件""" if os.path.exists(self.config_path): with open(self.config_path, 'r', encoding='utf-8') as f: return yaml.safe_load(f) return {} def save_config(self): """保存配置文件""" with open(self.config_path, 'w', encoding='utf-8') as f: yaml.dump(self.config, f, allow_unicode=True, default_flow_style=False) def init_ui(self): """初始化UI""" central_widget = QWidget() self.setCentralWidget(central_widget) main_layout = QVBoxLayout(central_widget) # 顶部状态栏 status_layout = QHBoxLayout() self.status_label = QLabel("状态: 未启动") self.status_label.setStyleSheet("font-weight: bold; color: red;") status_layout.addWidget(self.status_label) self.active_proxy_label = QLabel("当前代理: 无") status_layout.addWidget(self.active_proxy_label) self.stats_label = QLabel("统计: 0/0") status_layout.addWidget(self.stats_label) status_layout.addStretch() main_layout.addLayout(status_layout) # 控制按钮 control_layout = QHBoxLayout() self.start_button = QPushButton("▶ 开始服务") self.start_button.clicked.connect(self.on_start_service) control_layout.addWidget(self.start_button) self.stop_button = QPushButton("⏹ 停止服务") self.stop_button.clicked.connect(self.on_stop_service) self.stop_button.setEnabled(False) control_layout.addWidget(self.stop_button) self.switch_button = QPushButton("🔄 切换下一个") self.switch_button.clicked.connect(self.on_switch_proxy) self.switch_button.setEnabled(False) control_layout.addWidget(self.switch_button) self.fetch_button = QPushButton("🌐 获取免费代理") self.fetch_button.clicked.connect(self.on_fetch_proxies) control_layout.addWidget(self.fetch_button) self.use_local_button = QPushButton("📁 使用本地代理") self.use_local_button.clicked.connect(self.on_use_local_proxies) control_layout.addWidget(self.use_local_button) self.fetch_other_button = QPushButton("🔧 其他代理（预留）") self.fetch_other_button.clicked.connect(self.on_fetch_other_proxies) self.fetch_other_button.setEnabled(False) # 暂时禁用，预留功能 control_layout.addWidget(self.fetch_other_button) self.config_button = QPushButton("⚙️ 配置") self.config_button.clicked.connect(self.on_open_config) control_layout.addWidget(self.config_button) main_layout.addLayout(control_layout) # 进度条 self.progress_bar = QProgressBar() self.progress_bar.setVisible(False) main_layout.addWidget(self.progress_bar) # 标签页 tab_widget = QTabWidget() # 代理列表标签页 proxy_tab = QWidget() proxy_layout = QVBoxLayout(proxy_tab) # 过滤选项 filter_layout = QHBoxLayout() self.show_unavailable_checkbox = QCheckBox("显示不可用代理") self.show_unavailable_checkbox.setChecked(False) self.show_unavailable_checkbox.stateChanged.connect(self.on_filter_changed) filter_layout.addWidget(self.show_unavailable_checkbox) filter_layout.addStretch() proxy_layout.addLayout(filter_layout) self.proxy_table = QTableWidget() self.proxy_table.setColumnCount(10) self.proxy_table.setHorizontalHeaderLabels([ "IP地址", "端口", "协议", "国家", "状态", "延迟(ms)", "匿名级别", "速度", "运行时间", "最后更新" ]) self.proxy_table.horizontalHeader().setSectionResizeMode(QHeaderView.ResizeMode.Stretch) self.proxy_table.setAlternatingRowColors(True) self.proxy_table.setSelectionBehavior(QTableWidget.SelectionBehavior.SelectRows) self.proxy_table.setSelectionMode(QTableWidget.SelectionMode.SingleSelection) self.proxy_table.cellClicked.connect(self.on_proxy_selected) proxy_layout.addWidget(self.proxy_table) tab_widget.addTab(proxy_tab, "代理列表") # 连接日志标签页 connection_log_tab = QWidget() connection_log_layout = QVBoxLayout(connection_log_tab) self.connection_log_text = QTextEdit() self.connection_log_text.setReadOnly(True) self.connection_log_text.setFont(QFont("Consolas", 9)) connection_log_layout.addWidget(self.connection_log_text) clear_connection_log_button = QPushButton("清空连接日志") clear_connection_log_button.clicked.connect(self.connection_log_text.clear) connection_log_layout.addWidget(clear_connection_log_button) tab_widget.addTab(connection_log_tab, "连接日志") # 日志标签页 log_tab = QWidget() log_layout = QVBoxLayout(log_tab) self.log_text = QTextEdit() self.log_text.setReadOnly(True) self.log_text.setFont(QFont("Consolas", 9)) log_layout.addWidget(self.log_text) clear_log_button = QPushButton("清空日志") clear_log_button.clicked.connect(self.log_text.clear) log_layout.addWidget(clear_log_button) tab_widget.addTab(log_tab, "日志") main_layout.addWidget(tab_widget) def init_proxy_manager(self): """初始化代理管理器""" # 定义连接日志回调 def log_connection(message: str): from datetime import datetime timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") self.connection_log_text.append(f"[{timestamp}] {message}") self.proxy_manager = ProxyManager(self.config, connection_log_callback=log_connection) self.selected_proxy = None # 用户选择的代理 self.log("代理管理器已初始化") def on_start_service(self): """启动服务 - 仅启动端口监听，需要先选择代理""" # 检查是否选择了代理 if not self.selected_proxy: QMessageBox.warning(self, "警告", "请先选择一个代理节点！\n\n点击代理列表中的某一行来选择代理。") return self.start_button.setEnabled(False) self.progress_bar.setVisible(True) self.progress_bar.setRange(0, 0) self.log(f"正在启动服务，使用代理: {self.selected_proxy.get_address()}") def start_task(): # 只使用选中的代理 self.proxy_manager.set_proxies([self.selected_proxy]) return self.proxy_manager.start_service() self.worker_thread = WorkerThread(start_task) self.worker_thread.finished_signal.connect(self.on_service_started) self.worker_thread.start() def on_service_started(self, success: bool, message: str): """服务启动完成""" self.progress_bar.setVisible(False) self.start_button.setEnabled(True) if success: self.status_label.setText("状态: 运行中") self.status_label.setStyleSheet("font-weight: bold; color: green;") self.stop_button.setEnabled(True) self.switch_button.setEnabled(True) self.log("服务启动成功") self.update_proxy_table() self.update_stats() # 定时更新状态 self.update_timer = QTimer() self.update_timer.timeout.connect(self.update_status) self.update_timer.start(5000) else: self.log(f"服务启动失败: {message}") QMessageBox.warning(self, "错误", f"启动服务失败:\n{message}") def on_stop_service(self): """停止服务""" self.log("正在停止服务...") self.proxy_manager.stop_service() self.status_label.setText("状态: 已停止") self.status_label.setStyleSheet("font-weight: bold; color: red;") self.start_button.setEnabled(True) self.stop_button.setEnabled(False) self.switch_button.setEnabled(False) if hasattr(self, 'update_timer'): self.update_timer.stop() self.log("服务已停止") def on_switch_proxy(self): """切换代理""" self.log("正在切换代理...") def switch_task(): return self.proxy_manager.switch_to_next_proxy() self.worker_thread = WorkerThread(switch_task) self.worker_thread.finished_signal.connect(self.on_proxy_switched) self.worker_thread.start() def on_proxy_switched(self, success: bool, message: str): """代理切换完成""" if success: self.log("代理切换成功") self.update_status() self.update_proxy_table() else: self.log(f"代理切换失败: {message}") QMessageBox.warning(self, "警告", f"切换代理失败:\n{message}") def on_fetch_proxies(self): """获取代理 - 保留旧方法以兼容""" self.log("正在从网页获取代理...") self.progress_bar.setVisible(True) self.progress_bar.setRange(0, 0) def fetch_task(): count = self.proxy_manager.load_proxies() self.proxy_manager.check_all_proxies() return count self.worker_thread = WorkerThread(fetch_task) self.worker_thread.finished_signal.connect(self.on_fetch_completed) self.worker_thread.start() def on_use_local_proxies(self): """使用本地代理""" self.log("正在加载本地代理...") self.progress_bar.setVisible(True) self.progress_bar.setRange(0, 0) def load_local_task(): # 只加载本地文件 config_backup = self.proxy_manager.config.copy() self.proxy_manager.config['proxy_sources']['auto_fetch']['enabled'] = False self.proxy_manager.config['proxy_sources']['local_file']['enabled'] = True count = self.proxy_manager.load_proxies() self.proxy_manager.check_all_proxies() # 恢复配置 self.proxy_manager.config = config_backup return count self.worker_thread = WorkerThread(load_local_task) self.worker_thread.finished_signal.connect(self.on_fetch_completed) self.worker_thread.start() def on_fetch_other_proxies(self): """获取其他代理（预留功能 - 后续用于启动 scratch.py 测试）""" self.log("⚠️ 此功能暂未实现，预留用于后续测试") QMessageBox.information(self, "提示", "此功能暂未实现。\n\n" "计划用途：启动 scratch.py 进行代理测试\n" "后续将根据需求完善此功能。") def on_fetch_completed(self, success: bool, message: str): """获取代理完成""" self.progress_bar.setVisible(False) if success: self.log(f"成功获取 {message} 个代理") self.update_proxy_table() self.update_stats() else: self.log(f"获取代理失败: {message}") QMessageBox.warning(self, "错误", f"获取代理失败:\n{message}") def on_open_config(self): """打开配置对话框""" dialog = ConfigDialog(self.config, self) if dialog.exec() == QDialog.DialogCode.Accepted: self.config = dialog.get_config() self.save_config() self.log("配置已保存") def on_proxy_selected(self, row: int, column: int): """用户选择代理""" proxies = self.get_filtered_proxies() if row < len(proxies): self.selected_proxy = proxies[row] self.active_proxy_label.setText( f"选中代理: {self.selected_proxy.ip_address}:{self.selected_proxy.port} " f"({self.selected_proxy.protocol.value}) - {self.selected_proxy.latency_ms:.0f}ms" ) self.log(f"已选择代理: {self.selected_proxy.get_address()}") def on_filter_changed(self, state): """过滤条件改变""" self.update_proxy_table() def get_filtered_proxies(self): """获取过滤后的代理列表""" if not self.proxy_manager: return [] proxies = self.proxy_manager.proxies # 如果不显示不可用代理，则过滤掉 if not self.show_unavailable_checkbox.isChecked(): from core.models import ProxyStatus proxies = [p for p in proxies if p.status != ProxyStatus.UNAVAILABLE] return proxies def update_status(self): """更新状态显示""" active_proxy = self.proxy_manager.get_active_proxy() if active_proxy: self.active_proxy_label.setText( f"当前代理: {active_proxy.ip_address}:{active_proxy.port} " f"({active_proxy.protocol.value}) - {active_proxy.latency_ms:.0f}ms" ) else: self.active_proxy_label.setText("当前代理: 无") self.update_stats() def update_stats(self): """更新统计信息""" stats = self.proxy_manager.get_statistics() self.stats_label.setText( f"统计: 总计{stats['total']} | " f"可用{stats['available']} | " f"优秀{stats['excellent']} | " f"不可用{stats['unavailable']}" ) def update_proxy_table(self): """更新代理列表表格""" proxies = self.get_filtered_proxies() self.proxy_table.setRowCount(len(proxies)) for row, proxy in enumerate(proxies): # IP地址 self.proxy_table.setItem(row, 0, QTableWidgetItem(proxy.ip_address)) # 端口 self.proxy_table.setItem(row, 1, QTableWidgetItem(str(proxy.port))) # 协议 self.proxy_table.setItem(row, 2, QTableWidgetItem(proxy.protocol.value)) # 国家 self.proxy_table.setItem(row, 3, QTableWidgetItem(proxy.country or "-")) # 状态 status_item = QTableWidgetItem(self.get_status_text(proxy.status)) status_item.setForeground(self.get_status_color(proxy.status)) self.proxy_table.setItem(row, 4, status_item) # 延迟 latency_text = f"{proxy.latency_ms:.0f}" if proxy.latency_ms < 9999 else "-" self.proxy_table.setItem(row, 5, QTableWidgetItem(latency_text)) # 匿名级别 self.proxy_table.setItem(row, 6, QTableWidgetItem(proxy.anonymity or "-")) # 速度 self.proxy_table.setItem(row, 7, QTableWidgetItem(proxy.speed or "-")) # 运行时间 self.proxy_table.setItem(row, 8, QTableWidgetItem(proxy.uptime_percentage or "-")) # 最后更新 self.proxy_table.setItem(row, 9, QTableWidgetItem(proxy.last_updated or "-")) def get_status_text(self, status: ProxyStatus) -> str: """获取状态文本""" status_map = { ProxyStatus.UNKNOWN: "未知", ProxyStatus.CHECKING: "检测中", ProxyStatus.AVAILABLE: "可用", ProxyStatus.EXCELLENT: "优秀", ProxyStatus.UNAVAILABLE: "不可用" } return status_map.get(status, "未知") def get_status_color(self, status: ProxyStatus) -> QColor: """获取状态颜色""" color_map = { ProxyStatus.UNKNOWN: QColor("gray"), ProxyStatus.CHECKING: QColor("orange"), ProxyStatus.AVAILABLE: QColor("green"), ProxyStatus.EXCELLENT: QColor("darkgreen"), ProxyStatus.UNAVAILABLE: QColor("red") } return color_map.get(status, QColor("gray")) def log(self, message: str): """添加日志""" from datetime import datetime timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") self.log_text.append(f"[{timestamp}] {message}") def closeEvent(self, event): """关闭事件""" if self.proxy_manager: self.proxy_manager.stop_service() event.accept()