from __future__ import annotations import json import os import subprocess from dataclasses import dataclass from pathlib import Path from typing import Callable, Iterable CommandRunner = Callable[[list[str]], subprocess.CompletedProcess[str]] DEFAULT_CHECKS: dict[str, list[str]] = { "JDK": ["java", "-version"], "Maven": ["mvn", "-version"], "Gradle": ["gradle", "-version"], "Node.js": ["node", "--version"], "npm": ["npm", "--version"], "Vue CLI": ["vue", "--version"], "MySQL": ["mysql", "--version"], "Git": ["git", "--version"], } EXECUTABLE_NAMES: dict[str, list[str]] = { "JDK": ["java.exe", "java"], "Maven": ["mvn.cmd", "mvn.bat", "mvn"], "Gradle": ["gradle.bat", "gradle"], "Node.js": ["node.exe", "node"], "npm": ["npm.cmd", "npm"], "Vue CLI": ["vue.cmd", "vue"], "MySQL": ["mysql.exe", "mysql"], "Git": ["git.exe", "git"], } CONFIG_DIR = Path(".deploy_tool_config") TOOL_PATHS_FILE = CONFIG_DIR / "tool-paths.json" @dataclass(frozen=True) class EnvironmentCheck: name: str command: list[str] installed: bool version: str message: str source: str = "PATH" def check_environment( runner: CommandRunner | None = None, checks: dict[str, list[str]] | None = None, manual_paths: dict[str, str | Path] | None = None, search_roots: Iterable[Path] | None = None, ) -> list[EnvironmentCheck]: command_runner = runner or _run_command selected_checks = checks or DEFAULT_CHECKS configured_paths = {name: Path(value) for name, value in (manual_paths or {}).items() if str(value).strip()} roots = list(search_roots) if search_roots is not None else default_search_roots() results: list[EnvironmentCheck] = [] for name, command in selected_checks.items(): path_result = _try_command(name, command, "PATH", command_runner) if path_result.installed: results.append(path_result) continue manual_path = configured_paths.get(name) if manual_path and manual_path.exists(): manual_command = [str(manual_path), *command[1:]] manual_result = _try_command(name, manual_command, "手动指定", command_runner) if manual_result.installed: results.append(manual_result) continue discovered = discover_executable(name, roots) if discovered: discovered_command = [str(discovered), *command[1:]] discovered_result = _try_command(name, discovered_command, "自动发现", command_runner) if discovered_result.installed: results.append(discovered_result) continue results.append(path_result) return results def discover_executable(name: str, search_roots: Iterable[Path] | None = None) -> Path | None: executable_names = {item.lower() for item in EXECUTABLE_NAMES.get(name, [])} if not executable_names: return None for root in search_roots or default_search_roots(): if not root.exists() or not root.is_dir(): continue for candidate in _direct_candidates(name, root): if candidate.exists() and candidate.name.lower() in executable_names: return candidate return None def default_search_roots() -> list[Path]: roots = [ Path(os.environ.get("ProgramFiles", r"C:\Program Files")), Path(os.environ.get("ProgramFiles(x86)", r"C:\Program Files (x86)")), Path(os.environ.get("LOCALAPPDATA", "")), Path(r"C:\ProgramData"), Path(r"C:\xampp"), Path(r"C:\phpstudy_pro"), Path(r"D:\Program Files"), Path(r"D:\phpstudy_pro"), ] return [root for root in roots if str(root)] def load_tool_paths(config_file: Path = TOOL_PATHS_FILE) -> dict[str, Path]: if not config_file.exists(): return {} try: data = json.loads(config_file.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError): return {} if not isinstance(data, dict): return {} return {str(name): Path(value) for name, value in data.items() if isinstance(value, str) and value.strip()} def save_tool_paths(paths: dict[str, str | Path], config_file: Path = TOOL_PATHS_FILE) -> None: config_file.parent.mkdir(parents=True, exist_ok=True) data = {name: str(path) for name, path in paths.items() if str(path).strip()} config_file.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8") def _try_command(name: str, command: list[str], source: str, runner: CommandRunner) -> EnvironmentCheck: try: completed = runner(command) except FileNotFoundError: return EnvironmentCheck(name, command, False, "", f"{name} 未安装或不在 PATH 中", source) except subprocess.TimeoutExpired: return EnvironmentCheck(name, command, False, "", f"{name} 检测超时", source) except OSError as exc: return EnvironmentCheck(name, command, False, "", f"{name} 检测失败: {exc}", source) output = _combine_output(completed) installed = completed.returncode == 0 message = "已安装" if installed else f"命令返回错误码 {completed.returncode}" if installed and source != "PATH": message = f"已安装，来源: {source}" return EnvironmentCheck(name, command, installed, output, message, source) def _run_command(command: list[str]) -> subprocess.CompletedProcess[str]: return subprocess.run(command, capture_output=True, text=True, timeout=6, shell=False) def _combine_output(completed: subprocess.CompletedProcess[str]) -> str: output = "\n".join(part.strip() for part in (completed.stdout, completed.stderr) if part and part.strip()) return output.splitlines()[0] if output else "" def _direct_candidates(name: str, root: Path) -> list[Path]: if name == "MySQL": return [ root / "MySQL" / "MySQL Server 8.4" / "bin" / "mysql.exe", root / "MySQL" / "MySQL Server 8.0" / "bin" / "mysql.exe", root / "mysql" / "bin" / "mysql.exe", root / "Extensions" / "MySQL" / "bin" / "mysql.exe", ] if name == "JDK": return _glob(root, "Java/*/bin/java.exe") + _glob(root, "Eclipse Adoptium/*/bin/java.exe") if name == "Node.js": return [root / "nodejs" / "node.exe", root / "Programs" / "nodejs" / "node.exe"] if name == "npm": return [root / "nodejs" / "npm.cmd", root / "Programs" / "nodejs" / "npm.cmd"] if name == "Maven": return _glob(root, "apache-maven*/bin/mvn.cmd") if name == "Gradle": return _glob(root, "gradle*/bin/gradle.bat") if name == "Git": return [root / "Git" / "bin" / "git.exe", root / "Git" / "cmd" / "git.exe"] return [] def _glob(root: Path, pattern: str) -> list[Path]: try: return list(root.glob(pattern)) except OSError: return []