Files
AutoDeploy/deploy_tool/environment.py
2026-07-14 16:31:23 +08:00

190 lines
6.8 KiB
Python

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 []