from __future__ import annotations import queue import threading import tkinter as tk from pathlib import Path from tkinter import filedialog, messagebox, ttk from .deployer import DeploymentConfig, build_deployment_plan, run_steps from .environment import EnvironmentCheck, check_environment, load_tool_paths, save_tool_paths from .installer import build_install_command, recommend_jdk_version, run_install_command from .path_manager import apply_user_environment, build_environment_update from .scanner import ScanResult, scan_project TEXT = { "title": "\u0053\u0070\u0072\u0069\u006e\u0067\u0020\u0042\u006f\u006f\u0074\u0020\u002b\u0020\u0056\u0075\u0065\u0020\u81ea\u52a8\u5316\u90e8\u7f72\u5de5\u5177", "workbench": "\u90e8\u7f72\u5de5\u4f5c\u53f0", "select_dir": "\u9009\u62e9\u76ee\u5f55", "scan": "\u626b\u63cf\u9879\u76ee", "check_env": "\u68c0\u6d4b\u73af\u5883", "deploy": "\u6267\u884c\u90e8\u7f72", "results": "\u9879\u76ee\u4e0e\u73af\u5883\u7ed3\u679c", "config": "\u90e8\u7f72\u914d\u7f6e", "tool_paths": "\u5de5\u5177\u8def\u5f84\uff08\u53ef\u9009\uff09", "logs": "\u8fd0\u884c\u65e5\u5fd7", } SIDEBAR_ITEMS = ("项目识别", "环境检测", "工具路径", "数据库", "部署任务", "运行日志") def database_name_from_sql_files(sql_files: list[Path]) -> str: return sql_files[0].stem if sql_files else "" class DeployWorkbench(tk.Tk): def __init__(self) -> None: super().__init__() self.title(TEXT["title"]) self.geometry("1260x820") self.minsize(1080, 700) self.project_dir = tk.StringVar() self.backend_port = tk.StringVar(value="8080") self.user_frontend_port = tk.StringVar(value="8082") self.admin_frontend_port = tk.StringVar(value="8081") self.mysql_user = tk.StringVar(value="root") self.mysql_password = tk.StringVar() self.mysql_database = tk.StringVar() self.output_dir = tk.StringVar() self.jdk_version = tk.StringVar(value="8") self.jdk_reason = tk.StringVar(value="默认推荐 JDK 8，扫描项目后会自动调整。") self.tool_paths = load_tool_paths() self.tool_path_vars: dict[str, tk.StringVar] = { name: tk.StringVar(value=str(self.tool_paths.get(name, ""))) for name in ("MySQL", "JDK", "Maven", "Gradle", "Node.js", "npm", "Git") } self.scan_result: ScanResult | None = None self.environment_results: list[EnvironmentCheck] = [] self.ui_queue: queue.Queue[tuple[str, object]] = queue.Queue() self.sidebar_buttons: dict[str, tk.Button] = {} self.tool_path_entries: dict[str, ttk.Entry] = {} self.deployment_running = False self.deploy_buttons: list[ttk.Button] = [] self._configure_style() self._build_layout() self.after(120, self._drain_queue) def _configure_style(self) -> None: self.configure(bg="#f5f7fb") self.style = ttk.Style(self) self.style.theme_use("clam") self.style.configure("TFrame", background="#f5f7fb") self.style.configure("Sidebar.TFrame", background="#172033") self.style.configure("Panel.TFrame", background="#ffffff", relief="flat") self.style.configure("TLabel", background="#f5f7fb", foreground="#243043", font=("Microsoft YaHei UI", 10)) self.style.configure("Sidebar.TLabel", background="#172033", foreground="#dce6f7", font=("Microsoft YaHei UI", 10)) self.style.configure("Title.TLabel", background="#f5f7fb", foreground="#172033", font=("Microsoft YaHei UI", 18, "bold")) self.style.configure("CardTitle.TLabel", background="#ffffff", foreground="#172033", font=("Microsoft YaHei UI", 11, "bold")) self.style.configure("CardValue.TLabel", background="#ffffff", foreground="#2563eb", font=("Microsoft YaHei UI", 12, "bold")) self.style.configure("Muted.TLabel", background="#ffffff", foreground="#667085", font=("Microsoft YaHei UI", 9)) self.style.configure("Primary.TButton", font=("Microsoft YaHei UI", 10, "bold"), padding=(14, 8)) self.style.configure("TButton", font=("Microsoft YaHei UI", 10), padding=(10, 6)) self.style.configure("TEntry", padding=6) def _build_layout(self) -> None: shell = ttk.Frame(self) shell.pack(fill=tk.BOTH, expand=True) sidebar = ttk.Frame(shell, style="Sidebar.TFrame", width=190) sidebar.pack(side=tk.LEFT, fill=tk.Y) sidebar.pack_propagate(False) ttk.Label(sidebar, text=TEXT["workbench"], style="Sidebar.TLabel", font=("Microsoft YaHei UI", 16, "bold")).pack( anchor=tk.W, padx=22, pady=(26, 22) ) for item in SIDEBAR_ITEMS: button = tk.Button( sidebar, text=item, command=lambda name=item: self.activate_sidebar_item(name), anchor=tk.W, relief=tk.FLAT, bd=0, cursor="hand2", bg="#172033", activebackground="#22314d", fg="#dce6f7", activeforeground="#ffffff", font=("Microsoft YaHei UI", 10), padx=20, pady=8, ) button.pack(fill=tk.X, padx=6, pady=1) self.sidebar_buttons[item] = button content = ttk.Frame(shell) content.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=24, pady=22) self._build_header(content) self._build_dashboard(content) self._build_middle(content) self._build_logs(content) def _build_header(self, parent: ttk.Frame) -> None: ttk.Label(parent, text="Spring Boot + Vue 自动化部署", style="Title.TLabel").pack(anchor=tk.W) action_bar = ttk.Frame(parent) action_bar.pack(fill=tk.X, pady=(14, 16)) self.project_dir_entry = ttk.Entry(action_bar, textvariable=self.project_dir) self.project_dir_entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 10)) ttk.Button(action_bar, text=TEXT["select_dir"], command=self.choose_folder).pack(side=tk.LEFT, padx=4) ttk.Button(action_bar, text=TEXT["scan"], style="Primary.TButton", command=self.scan_folder).pack(side=tk.LEFT, padx=4) ttk.Button(action_bar, text=TEXT["check_env"], command=self.check_environment_async).pack(side=tk.LEFT, padx=4) ttk.Button(action_bar, text="\u5b89\u88c5\u73af\u5883", style="Primary.TButton", command=self.show_environment_installer).pack(side=tk.LEFT, padx=4) self.deploy_button = ttk.Button( action_bar, text=TEXT["deploy"], style="Primary.TButton", command=self.run_deploy_plan_async ) self.deploy_button.pack(side=tk.LEFT, padx=4) self.deploy_buttons.append(self.deploy_button) def _build_dashboard(self, parent: ttk.Frame) -> None: cards = ttk.Frame(parent) cards.pack(fill=tk.X, pady=(0, 16)) self.card_labels: dict[str, ttk.Label] = {} for key, title in (("backend", "后端"), ("frontend", "前端"), ("sql", "SQL 文件"), ("env", "环境")): card = ttk.Frame(cards, style="Panel.TFrame", padding=16) card.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 12)) ttk.Label(card, text=title, style="CardTitle.TLabel").pack(anchor=tk.W) value = ttk.Label(card, text="未扫描", style="CardValue.TLabel") value.pack(anchor=tk.W, pady=(9, 3)) ttk.Label(card, text="等待操作", style="Muted.TLabel").pack(anchor=tk.W) self.card_labels[key] = value def _build_middle(self, parent: ttk.Frame) -> None: middle = ttk.Frame(parent) middle.pack(fill=tk.BOTH, expand=True) left = ttk.Frame(middle, style="Panel.TFrame", padding=16) left.pack(side=tk.LEFT, fill=tk.BOTH, expand=True, padx=(0, 14)) ttk.Label(left, text=TEXT["results"], style="CardTitle.TLabel").pack(anchor=tk.W) self.result_text = tk.Text(left, height=16, wrap=tk.WORD, relief=tk.FLAT, bg="#ffffff", fg="#243043", font=("Consolas", 10)) self.result_text.pack(fill=tk.BOTH, expand=True, pady=(10, 0)) right = ttk.Frame(middle, style="Panel.TFrame", padding=16, width=420) right.pack(side=tk.LEFT, fill=tk.Y) right.pack_propagate(False) ttk.Label(right, text=TEXT["config"], style="CardTitle.TLabel").pack(anchor=tk.W) self.mysql_user_widget = self._add_field(right, "MySQL 用户", self.mysql_user) self.mysql_password_widget = self._add_field(right, "MySQL 密码", self.mysql_password, show="*") self.database_widget = self._add_field(right, "数据库名", self.mysql_database) ttk.Label( right, text="服务端口使用默认值：8080 / 8082 / 8081", style="Muted.TLabel", ).pack(anchor=tk.W, pady=(12, 0)) def _build_jdk_install_section(self, parent: ttk.Frame) -> None: ttk.Label(parent, text="JDK 推荐与安装", style="CardTitle.TLabel").pack(anchor=tk.W) row = ttk.Frame(parent, style="Panel.TFrame") row.pack(fill=tk.X, pady=(8, 4)) ttk.Label(row, text="版本", style="Muted.TLabel").pack(side=tk.LEFT, padx=(0, 8)) ttk.Combobox(row, textvariable=self.jdk_version, values=("8", "11", "17", "21"), state="readonly", width=8).pack(side=tk.LEFT) ttk.Button(row, text="安装 JDK", command=self.install_jdk_async).pack(side=tk.LEFT, padx=(10, 0)) ttk.Button(row, text="查看命令", command=self.preview_jdk_install_command).pack(side=tk.LEFT, padx=(6, 0)) ttk.Button(row, text="加入 PATH", command=lambda: self.add_tool_to_path("JDK")).pack(side=tk.LEFT, padx=(6, 0)) ttk.Label(parent, textvariable=self.jdk_reason, style="Muted.TLabel", wraplength=360).pack(anchor=tk.W, pady=(2, 0)) def _build_logs(self, parent: ttk.Frame) -> None: log_panel = ttk.Frame(parent, style="Panel.TFrame", padding=14) log_panel.pack(fill=tk.BOTH, pady=(16, 0)) ttk.Label(log_panel, text=TEXT["logs"], style="CardTitle.TLabel").pack(anchor=tk.W) self.log_text = tk.Text(log_panel, height=9, wrap=tk.WORD, relief=tk.FLAT, bg="#111827", fg="#d1fadf", insertbackground="#d1fadf", font=("Consolas", 10)) self.log_text.pack(fill=tk.BOTH, expand=True, pady=(10, 0)) def _add_field(self, parent: ttk.Frame, label: str, variable: tk.StringVar, show: str | None = None) -> ttk.Entry: ttk.Label(parent, text=label, style="Muted.TLabel").pack(anchor=tk.W, pady=(10, 4)) entry = ttk.Entry(parent, textvariable=variable, show=show) entry.pack(fill=tk.X) return entry def _add_tool_path_row(self, parent: ttk.Frame, name: str) -> None: ttk.Label(parent, text=name, style="Muted.TLabel").pack(anchor=tk.W, pady=(8, 3)) row = ttk.Frame(parent, style="Panel.TFrame") row.pack(fill=tk.X) entry = ttk.Entry(row, textvariable=self.tool_path_vars[name]) entry.pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 6)) self.tool_path_entries[name] = entry ttk.Button(row, text="选择", command=lambda tool=name: self.choose_tool_path(tool)).pack(side=tk.LEFT) if name not in {"JDK", "npm"}: ttk.Button(row, text="安装", command=lambda tool=name: self.install_tool_async(tool)).pack(side=tk.LEFT, padx=(6, 0)) ttk.Button(row, text="PATH", command=lambda tool=name: self.add_tool_to_path(tool)).pack(side=tk.LEFT, padx=(6, 0)) def show_environment_installer(self) -> None: dialog = tk.Toplevel(self) dialog.title("\u5b89\u88c5\u73af\u5883") dialog.geometry("680x420") dialog.minsize(620, 360) dialog.configure(bg="#f5f7fb") dialog.transient(self) panel = ttk.Frame(dialog, style="Panel.TFrame", padding=18) panel.pack(fill=tk.BOTH, expand=True, padx=16, pady=16) ttk.Label(panel, text="\u5b89\u88c5\u73af\u5883", style="CardTitle.TLabel").pack(anchor=tk.W) ttk.Label( panel, text="\u5b89\u88c5\u524d\u4f1a\u5f39\u7a97\u786e\u8ba4 winget \u547d\u4ee4\uff0c\u5b89\u88c5\u5b8c\u6210\u540e\u4f1a\u81ea\u52a8\u91cd\u65b0\u68c0\u6d4b\u73af\u5883\u3002", style="Muted.TLabel", ).pack(anchor=tk.W, pady=(4, 12)) jdk_row = ttk.Frame(panel, style="Panel.TFrame") jdk_row.pack(fill=tk.X, pady=5) ttk.Label(jdk_row, text="JDK", width=12, style="CardTitle.TLabel").pack(side=tk.LEFT) ttk.Combobox(jdk_row, textvariable=self.jdk_version, values=("8", "11", "17", "21"), state="readonly", width=8).pack(side=tk.LEFT) ttk.Button(jdk_row, text="\u5b89\u88c5", command=self.install_jdk_async).pack(side=tk.LEFT, padx=(8, 0)) ttk.Button(jdk_row, text="\u52a0\u5165 PATH", command=lambda: self.add_tool_to_path("JDK")).pack(side=tk.LEFT, padx=(8, 0)) ttk.Label(jdk_row, textvariable=self.jdk_reason, style="Muted.TLabel").pack(side=tk.LEFT, fill=tk.X, expand=True, padx=(10, 0)) for name in ("Maven", "Gradle", "Node.js", "Git", "MySQL"): row = ttk.Frame(panel, style="Panel.TFrame") row.pack(fill=tk.X, pady=5) ttk.Label(row, text=name, width=12, style="CardTitle.TLabel").pack(side=tk.LEFT) ttk.Button(row, text="\u5b89\u88c5", command=lambda tool=name: self.install_tool_async(tool)).pack(side=tk.LEFT) ttk.Button(row, text="\u9009\u62e9\u8def\u5f84", command=lambda tool=name: self.choose_tool_path(tool)).pack(side=tk.LEFT, padx=(8, 0)) ttk.Button(row, text="\u52a0\u5165 PATH", command=lambda tool=name: self.add_tool_to_path(tool)).pack(side=tk.LEFT, padx=(8, 0)) ttk.Separator(panel).pack(fill=tk.X, pady=14) ttk.Button(panel, text="\u68c0\u6d4b\u73af\u5883", command=self.check_environment_async).pack(side=tk.LEFT) ttk.Button(panel, text="\u5173\u95ed", command=dialog.destroy).pack(side=tk.RIGHT) def activate_sidebar_item(self, item: str) -> None: for name, button in self.__dict__.get("sidebar_buttons", {}).items(): button.configure(bg="#22314d" if name == item else "#172033", fg="#ffffff" if name == item else "#dce6f7") if item == "项目识别": self._focus_widget(self.__dict__.get("project_dir_entry")) elif item == "环境检测": self.check_environment_async() elif item == "工具路径": entries = self.__dict__.get("tool_path_entries", {}) first_entry = next(iter(entries.values()), None) self._focus_widget(first_entry) elif item == "数据库": self._focus_widget(self.__dict__.get("database_widget")) elif item == "部署任务": self._focus_widget(self.__dict__.get("deploy_button")) elif item == "运行日志": log_text = self.__dict__.get("log_text") self._focus_widget(log_text) if log_text is not None: log_text.see(tk.END) def _focus_widget(self, widget: object | None) -> None: if widget is not None: widget.focus_set() def choose_folder(self) -> None: folder = filedialog.askdirectory(title="选择前后端项目根目录") if folder: self.project_dir.set(folder) if not self.output_dir.get(): self.output_dir.set(str(Path(folder) / "deploy")) def choose_output_dir(self) -> None: folder = filedialog.askdirectory(title="选择部署输出目录") if folder: self.output_dir.set(folder) def choose_tool_path(self, name: str) -> None: file_path = filedialog.askopenfilename(title=f"选择 {name} 可执行文件", filetypes=(("可执行文件", "*.exe *.cmd *.bat"), ("所有文件", "*.*"))) if file_path: self.tool_path_vars[name].set(file_path) self.save_tool_path_settings(silent=True) def save_tool_path_settings(self, silent: bool = False) -> None: paths = {name: value.get().strip() for name, value in self.tool_path_vars.items() if value.get().strip()} save_tool_paths(paths) self.tool_paths = load_tool_paths() if not silent: self._append_log("工具路径已保存。") def scan_folder(self) -> None: if not self.project_dir.get().strip(): messagebox.showwarning("需要目录", "请先选择或输入项目文件夹。") return try: self.scan_result = scan_project(self.project_dir.get().strip()) except Exception as exc: messagebox.showerror("扫描失败", str(exc)) return database_name = database_name_from_sql_files(self.scan_result.sql_files) if database_name: self.mysql_database.set(database_name) self._append_log(f"已根据 SQL 文件自动填写数据库名: {database_name}") self._apply_jdk_recommendation() self._render_scan_result() self._append_log("项目扫描完成。") def _apply_jdk_recommendation(self) -> None: recommendation = recommend_jdk_version(self.scan_result) self.jdk_version.set(recommendation.version) self.jdk_reason.set(recommendation.reason) self._append_log(f"JDK 推荐: {recommendation.reason}") def check_environment_async(self) -> None: self.save_tool_path_settings(silent=True) manual_paths = load_tool_paths() self._append_log("开始检测本机环境，会同时检查 PATH、手动路径和常见安装目录...") threading.Thread(target=lambda: self.ui_queue.put(("env", check_environment(manual_paths=manual_paths))), daemon=True).start() def preview_jdk_install_command(self) -> None: command = build_install_command("JDK", version=self.jdk_version.get()) self._append_log(f"JDK {self.jdk_version.get()} 安装命令: {' '.join(command)}") def install_jdk_async(self) -> None: command = build_install_command("JDK", version=self.jdk_version.get()) if not messagebox.askyesno( "确认安装 JDK", "即将通过 winget 安装 JDK。\n\n" f"版本: JDK {self.jdk_version.get()}\n" f"命令: {' '.join(command)}\n\n" "安装可能需要管理员权限，并可能修改 PATH。是否继续？", ): return self._append_log(f"开始安装 JDK {self.jdk_version.get()}...") threading.Thread(target=lambda: self.ui_queue.put(("install_done", ("JDK", run_install_command(command, self._thread_log)))), daemon=True).start() def install_tool_async(self, tool_name: str) -> None: try: command = build_install_command(tool_name) except ValueError as exc: messagebox.showwarning("暂不支持自动安装", str(exc)) return if not messagebox.askyesno( f"确认安装 {tool_name}", f"即将通过 winget 安装 {tool_name}。\n\n" f"命令: {' '.join(command)}\n\n" "安装可能需要管理员权限，并可能修改 PATH。是否继续？", ): return self._append_log(f"开始安装 {tool_name}...") threading.Thread( target=lambda: self.ui_queue.put(("install_done", (tool_name, run_install_command(command, self._thread_log)))), daemon=True, ).start() def add_tool_to_path(self, tool_name: str) -> None: result = self._find_environment_result(tool_name) executable = Path(result.command[0]) if result and Path(result.command[0]).exists() else None configured = self.tool_path_vars.get(tool_name) if executable is None and configured and configured.get().strip() and Path(configured.get().strip()).exists(): executable = Path(configured.get().strip()) if executable is None: messagebox.showwarning("未找到可执行文件", f"请先检测环境，或手动选择 {tool_name} 的可执行文件。") return update = build_environment_update(tool_name, executable) env_text = "\n".join(f"{name}={value}" for name, value in update.env_vars.items()) or "无额外环境变量" if not messagebox.askyesno( "确认加入 PATH", f"将写入当前用户环境变量，不修改系统级 PATH。\n\n" f"工具: {tool_name}\n" f"PATH 目录: {update.path_entries[0]}\n" f"{env_text}\n\n" "修改后新打开的终端会生效。是否继续？", ): return try: apply_user_environment(update) except Exception as exc: messagebox.showerror("写入失败", str(exc)) return self._append_log(f"{tool_name} 已加入当前用户 PATH。") self.check_environment_async() def add_all_available_tools_to_path(self) -> None: available = [ item.name for item in self.environment_results if item.installed and Path(item.command[0]).exists() and item.source in {"手动指定", "自动发现"} ] if not available: messagebox.showinfo("没有可加入项", "请先检测环境，或手动指定工具路径。") return if not messagebox.askyesno("确认加入 PATH", f"将这些工具加入当前用户 PATH：\n{', '.join(available)}\n\n是否继续？"): return for name in available: result = self._find_environment_result(name) if result: apply_user_environment(build_environment_update(name, Path(result.command[0]))) self._append_log(f"{name} 已加入当前用户 PATH。") self.check_environment_async() def _find_environment_result(self, tool_name: str) -> EnvironmentCheck | None: for item in self.environment_results: if item.name == tool_name: return item return None def _set_deployment_running(self, running: bool) -> None: self.deployment_running = running state = tk.DISABLED if running else tk.NORMAL for button in self.deploy_buttons: button.configure(state=state) def _run_deployment_worker(self, plan) -> None: exit_code = 1 try: exit_code = run_steps(plan, self._thread_log) except Exception as exc: self.ui_queue.put(("log", f"部署线程异常: {exc}")) finally: self.ui_queue.put(("deploy_done", exit_code)) def run_deploy_plan_async(self) -> None: if self.deployment_running: self._append_log("部署任务正在执行，请等待当前任务结束。") return plan = self._current_plan() if not plan: return if not messagebox.askyesno("确认执行", "将开始自动执行本地部署：构建前后端、复制前端产物、导入 SQL，并启动后端、用户端和管理端服务。请确认继续。"): return self._set_deployment_running(True) worker = threading.Thread(target=self._run_deployment_worker, args=(plan,), daemon=True) try: worker.start() except Exception as exc: self._set_deployment_running(False) self._append_log(f"无法启动部署线程: {exc}") def _current_plan(self): if self.scan_result is None: messagebox.showwarning("需要扫描", "请先扫描项目。") return [] config = DeploymentConfig( mysql_user=self.mysql_user.get().strip() or "root", mysql_password=self.mysql_password.get(), mysql_database=self.mysql_database.get().strip(), backend_port="8080", user_frontend_port="8082", admin_frontend_port="8081", output_dir=None, tool_paths=load_tool_paths(), ) return build_deployment_plan(self.scan_result, config) def _render_scan_result(self) -> None: assert self.scan_result is not None scan = self.scan_result self.card_labels["backend"].configure(text="已识别" if scan.backend else "未找到") self.card_labels["frontend"].configure(text=f"{len(scan.detected_frontends)} 个" if scan.detected_frontends else "未找到") self.card_labels["sql"].configure(text=f"{len(scan.sql_files)} 个") lines = [f"根目录: {scan.root}", ""] lines.append(f"后端: {scan.backend.path if scan.backend else '未识别'}") if scan.backend: lines.append(f" 依据: {scan.backend.detail}") if scan.backend.metadata: lines.append(" 版本线索:") for key, value in scan.backend.metadata.items(): lines.append(f" {key}: {value}") lines.append(f"前端: {len(scan.detected_frontends)} 个") for frontend in scan.detected_frontends: display_name = frontend.metadata.get("display_name", frontend.path.name) port = frontend.metadata.get("port", "未配置") lines.append(f" {display_name}: {frontend.path}") lines.append(f" 端口: {port}，依据: {frontend.detail}") lines.append(f"JDK 推荐: JDK {self.jdk_version.get()} - {self.jdk_reason.get()}") lines.append("SQL:") lines.extend(f" - {path}" for path in scan.sql_files[:20]) if len(scan.sql_files) > 20: lines.append(f" ... 还有 {len(scan.sql_files) - 20} 个") self._set_result_text("\n".join(lines)) def _render_environment(self, results: list[EnvironmentCheck]) -> None: self.environment_results = results installed = sum(1 for item in results if item.installed) self.card_labels["env"].configure(text=f"{installed}/{len(results)} 可用") lines = ["环境检测结果:", ""] for item in results: status = "OK" if item.installed else "MISS" lines.append(f"[{status}] {item.name} ({item.source}): {item.message}") lines.append(f" 命令: {' '.join(item.command)}") if item.version: lines.append(f" 版本: {item.version}") self._set_result_text("\n".join(lines)) self._append_log("环境检测完成。") def _set_result_text(self, value: str) -> None: self.result_text.configure(state=tk.NORMAL) self.result_text.delete("1.0", tk.END) self.result_text.insert(tk.END, value) def _append_log(self, value: str) -> None: self.log_text.configure(state=tk.NORMAL) self.log_text.insert(tk.END, value + "\n") self.log_text.see(tk.END) def _thread_log(self, value: str) -> None: self.ui_queue.put(("log", value)) def _drain_queue(self) -> None: while True: try: event, payload = self.ui_queue.get_nowait() except queue.Empty: break if event == "env": self._render_environment(payload) # type: ignore[arg-type] elif event == "log": self._append_log(str(payload)) elif event == "deploy_done": self._set_deployment_running(False) self._append_log(f"部署任务结束，退出码: {payload}") elif event == "install_done": tool_name, exit_code = payload # type: ignore[misc] self._append_log(f"{tool_name} 安装命令结束，退出码: {exit_code}") self.check_environment_async() self.after(1800, lambda name=tool_name: self._prompt_add_tool_after_install(name)) self.after(120, self._drain_queue) def _prompt_add_tool_after_install(self, tool_name: str) -> None: result = self._find_environment_result(tool_name) if result and result.installed and Path(result.command[0]).exists(): extra = " 并设置 JAVA_HOME" if tool_name == "JDK" else "" if messagebox.askyesno(f"{tool_name} 已安装", f"检测到 {tool_name} 可执行文件，是否加入当前用户 PATH{extra}？"): self.add_tool_to_path(tool_name) def run_app() -> None: app = DeployWorkbench() app.mainloop()