import os gui_path = 'src/virtualbox_client/gui.py' with open(gui_path, 'r', encoding='utf-8') as f: content = f.read() # ===== 1. 添加 ScriptWorker 类（放在 DownloadWorker 之后） ===== old = """class WorkspaceWorker(QObject):""" new = """class ScriptWorker(QObject): """上传本地脚本到虚拟机、执行并下载日志""" log_signal = Signal(str, str) finished_signal = Signal(bool) error_signal = Signal(str) def __init__(self, config: dict, local_path: str): super().__init__() self.config = config self.local_path = local_path self.cancel_event = threading.Event() def cancel(self): self.cancel_event.set() def run(self): ssh_only = self.config.get("ssh_only", False) _com_initialized = False if not ssh_only: try: import ctypes ctypes.windll.ole32.CoInitializeEx(None, 0) _com_initialized = True except Exception: pass try: if ssh_only: from virtualbox_client.utils.ssh_manager import SSHManager vm = SSHManager( self.config["ssh_host"], self.config["ssh_port"], self.config["ssh_user"], self.config["ssh_pass"], ) else: from virtualbox_client.utils.vbox_ssh_manager import VBoxSSHManager vm = VBoxSSHManager( self.config["vm_name"], self.config["ssh_host"], self.config["ssh_port"], self.config["ssh_user"], self.config["ssh_pass"], ) vm.set_cancel_event(self.cancel_event) if self.cancel_event.is_set(): return import builtins original_print = print def custom_print(*args, **kwargs): msg = " ".join(str(a) for a in args) self.log_signal.emit(msg, "info") original_print(*args, **kwargs) builtins.print = custom_print try: script_name = os.path.basename(self.local_path) remote_path = f"/tmp/{script_name}" log_path = f"/tmp/{script_name}.log" local_log_path = f"{script_name}.log" # 1. 上传脚本 self.log_signal.emit(f"📤 [1/3] 上传脚本: {self.local_path} -> {remote_path}", "info") if not vm.upload_file(self.local_path, remote_path): self.log_signal.emit("❌ 上传失败", "error") return # 2. 执行脚本 self.log_signal.emit(f"⚡ [2/3] 执行脚本: {remote_path}", "info") cmd = ( f"chmod +x {remote_path} && " f"echo '{self.config['ssh_pass']}' | sudo -S {remote_path} 2>&1 | " f"tee {log_path}" ) out, err, code = vm.exec_command(cmd, get_pty=False) self.log_signal.emit(f"✔ 脚本执行完成，退出码: {code}", "info") # 3. 下载日志 self.log_signal.emit(f"📥 [3/3] 下载日志: {log_path} -> {local_log_path}", "info") if vm.download_file(log_path, local_log_path): self.log_signal.emit(f"✔ 日志已保存到本地: {local_log_path}", "success") else: self.log_signal.emit("⚠ tee 日志下载失败，尝试保存终端输出", "warning") if out: with open(local_log_path, "w") as f: f.write(out) self.log_signal.emit(f"✔ 终端输出已保存到: {local_log_path}", "success") # 4. 清理远程临时文件 vm.exec_command(f'rm -f "{remote_path}" "{log_path}"') self.log_signal.emit("🧹 远程临时文件已清理", "info") self.log_signal.emit("🎉 脚本执行全部完成！", "success") self.finished_signal.emit(True) finally: builtins.print = original_print except ImportError as e: self.log_signal.emit(f"❌ 缺少依赖: {e}", "error") self.error_signal.emit(str(e)) except Exception as e: self.log_signal.emit(f"❌ 脚本执行错误: {str(e)}", "error") self.error_signal.emit(str(e)) finally: if _com_initialized: try: import ctypes ctypes.windll.ole32.CoUninitialize() except Exception: pass class WorkspaceWorker(QObject):""" content = content.replace(old, new) # ===== 2. 在 TaskPanel.setup_ui 的文件传输 tab 后面添加脚本执行 tab ===== old_file_tab_end = """ file_layout.addStretch() self.tab_widget.addTab(file_tab, "📁 文件传输")""" new_file_tab_end = """ file_layout.addStretch() self.tab_widget.addTab(file_tab, "📁 文件传输") # ====== 脚本执行 Tab ====== script_tab = QWidget() script_layout = QVBoxLayout(script_tab) script_layout.setContentsMargins(8, 8, 8, 8) script_group = QGroupBox("⚡ 上传并执行脚本") script_group_layout = QVBoxLayout(script_group) script_path_layout = QHBoxLayout() self.script_path_edit = QLineEdit(placeholderText="选择本地 .sh 脚本文件...") script_path_layout.addWidget(self.script_path_edit) self.script_browse_btn = QPushButton("📂 浏览") self.script_browse_btn.setFixedWidth(80) self.script_browse_btn.clicked.connect(self._browse_script) script_path_layout.addWidget(self.script_browse_btn) script_group_layout.addWidget(QLabel("本地脚本路径:")) script_group_layout.addLayout(script_path_layout) self.run_script_btn = QPushButton("⚡ 上传并执行脚本") self.run_script_btn.setStyleSheet(\""" QPushButton { background-color: #8e44ad; color: white; padding: 10px 30px; font-size: 14px; font-weight: bold; border: none; border-radius: 6px; } QPushButton:hover { background-color: #9b59b6; } QPushButton:disabled { background-color: #555; color: #999; } \""") self.run_script_btn.clicked.connect(self._run_script) script_group_layout.addWidget(self.run_script_btn) script_layout.addWidget(script_group) # 执行说明 desc_group = QGroupBox("📖 说明") desc_layout = QVBoxLayout(desc_group) desc_label = QLabel( "1. 选择一个本地 .sh 脚本文件\\n" "2. 点击按钮自动完成：\\n" " 📤 上传脚本到虚拟机 /tmp/ 目录\\n" " ⚡ 以 sudo 权限执行脚本\\n" " 📥 下载执行日志到本地\\n" " 🧹 清理远程临时文件" ) desc_label.setStyleSheet("color: #aaa; font-size: 12px; line-height: 1.6;") desc_layout.addWidget(desc_label) script_layout.addWidget(desc_group) script_layout.addStretch() self.tab_widget.addTab(script_tab, "⚡ 脚本执行")""" content = content.replace(old_file_tab_end, new_file_tab_end) # ===== 3. 在 TaskPanel 的 _set_ui_running 中禁用脚本按钮 ===== old_running = """ def _set_ui_running(self, running: bool): self.run_btn.setEnabled(not running) self.cancel_btn.setEnabled(running) self.task_input.setEnabled(not running) if self._config_panel: self._config_panel.set_enabled(not running)""" new_running = """ def _set_ui_running(self, running: bool): self.run_btn.setEnabled(not running) self.cancel_btn.setEnabled(running) self.task_input.setEnabled(not running) self.run_script_btn.setEnabled(not running) self.script_browse_btn.setEnabled(not running) self.script_path_edit.setEnabled(not running) if self._config_panel: self._config_panel.set_enabled(not running)""" content = content.replace(old_running, new_running) # ===== 4. 在 TaskPanel 中添加 _browse_script 和 _run_script 方法 ===== # 放在 _on_download_error 之后、class MainWindow 之前 old_mainwindow = """class MainWindow(QMainWindow):""" new_methods = """ _script_thread = None _script_worker = None def _cleanup_script(self): if self._script_thread: try: self._script_thread.finished.disconnect() except (TypeError, RuntimeError): pass if self._script_thread.isRunning(): self._script_thread.quit() if not self._script_thread.wait(5000): self._script_thread.terminate() self._script_thread.wait(3000) self._script_thread = None if self._script_worker: try: self._script_worker.log_signal.disconnect() self._script_worker.finished_signal.disconnect() self._script_worker.error_signal.disconnect() except (TypeError, RuntimeError): pass self._script_worker = None def _browse_script(self): file_path, _ = QFileDialog.getOpenFileName( self, "选择要执行的脚本", "", "Shell 脚本 (*.sh);;所有文件 (*)" ) if file_path: self.script_path_edit.setText(file_path) def _run_script(self): if not self._config_panel or not self._config_panel.validate(): return local_path = self.script_path_edit.text().strip() if not local_path: QMessageBox.warning(self, "提示", "请选择要执行的脚本文件") return if not os.path.isfile(local_path): QMessageBox.warning(self, "提示", f"本地脚本不存在: {local_path}") return self._cleanup_script() self._set_ui_running(True) self.tab_widget.setCurrentIndex(0) # 切换到任务日志 tab self.log_panel.clear_log() config = self._config_panel.get_config() self._script_worker = ScriptWorker(config, local_path) self._script_thread = QThread(self) self._script_worker.moveToThread(self._script_thread) self._script_worker.log_signal.connect(self.log_panel.append_log) self._script_worker.finished_signal.connect(self._on_script_finished) self._script_worker.error_signal.connect(self._on_script_error) self._script_thread.started.connect(self._script_worker.run) self._script_thread.finished.connect(self._cleanup_script) self._script_thread.start() def _on_script_finished(self, success: bool): self._set_ui_running(False) if success: self.log_panel.append_log("✅ 脚本执行已完成", "success") def _on_script_error(self, error_msg: str): self._set_ui_running(False) self.log_panel.append_log(f"❌ 脚本执行失败: {error_msg}", "error") """ + old_mainwindow content = content.replace(old_mainwindow, new_methods) with open(gui_path, 'w', encoding='utf-8') as f: f.write(content) print('✅ gui.py 已集成脚本执行功能')