"""环境管理器。""" from __future__ import annotations from pathlib import Path from typing import Any from aide.core import output from aide.core.config import ConfigManager from aide.env.registry import ModuleRegistry, register_builtin_modules # 运行时模块（--runtime 时使用） RUNTIME_MODULES = ["python", "uv"] # 默认启用的模块 DEFAULT_MODULES = ["python", "uv", "venv", "requirements"] class EnvManager: """环境管理器。""" def __init__(self, root: Path, cfg: ConfigManager): self.root = root self.cfg = cfg self.verbose = False # 确保模块已注册 register_builtin_modules() def list_modules(self) -> None: """列出所有可用模块（aide env list）。""" config = self.cfg.load_config() enabled = self._get_enabled_modules(config) print("可用模块:") print(f" {'模块':<14} {'描述':<20} {'能力':<16} {'需要配置'}") print(" " + "─" * 60) for info in ModuleRegistry.list_info(): caps = ", ".join(info.capabilities) req_cfg = "是" if info.requires_config else "否" if info.config_keys: req_cfg += f" [{', '.join(info.config_keys)}]" print(f" {info.name:<14} {info.description:<20} {caps:<16} {req_cfg}") print() if enabled: print(f"当前启用: {', '.join(enabled)}") else: output.warn("未配置启用模块列表") def ensure( self, runtime_only: bool = False, modules: list[str] | None = None, check_only: bool = False, verbose: bool = False, ) -> bool: """检测并修复环境。 Args: runtime_only: 仅检测运行时环境 modules: 指定要检测的模块 check_only: 仅检测不修复（--all 模式） verbose: 显示详细配置信息 Returns: 是否全部成功 """ self.verbose = verbose config = self.cfg.load_config() enabled_modules = self._get_enabled_modules(config) # verbose: 输出基础信息 if verbose: self._print_verbose_header(config, enabled_modules) # 确定要检测的模块列表 if runtime_only: target_modules = RUNTIME_MODULES elif modules: target_modules = modules elif check_only: # --all 模式 if not enabled_modules: output.warn("未配置启用模块列表，将检测所有支持的模块") target_modules = ModuleRegistry.names() else: target_modules = enabled_modules else: target_modules = enabled_modules if not target_modules: output.warn("没有要检测的模块") return True if verbose: print(f" 目标模块: {', '.join(target_modules)}") print() # 执行检测 all_success = True results: list[tuple[str, bool, str]] = [] for name in target_modules: is_enabled = name in enabled_modules success, msg = self._process_module( name=name, config=config, is_enabled=is_enabled, check_only=check_only, ) results.append((name, success, msg)) if not success and is_enabled: all_success = False break # 启用模块失败时停止 # 输出最终状态 if all_success and not check_only: # 构建摘要信息 summary_parts = [] for name, success, msg in results: if success and msg: summary_parts.append(f"{name}:{msg}") if summary_parts: output.ok(f"环境就绪 ({', '.join(summary_parts)})") return all_success def _print_verbose_header(self, config: dict[str, Any], enabled_modules: list[str]) -> None: """输出详细模式的头部信息。""" print("=" * 60) print("环境检测详细信息") print("=" * 60) print() print(f" 工作目录: {self.root}") print(f" 配置文件: {self.cfg.config_path}") print(f" 配置存在: {'是' if self.cfg.config_path.exists() else '否'}") print() print(f" 启用模块: {', '.join(enabled_modules) if enabled_modules else '(未配置)'}") print() def _print_verbose_module(self, name: str, module_config: dict[str, Any]) -> None: """输出模块的详细配置信息。""" print(f" [{name}] 配置:") if not module_config: print(" (无配置)") else: for key, value in module_config.items(): if key.startswith("_"): continue # 跳过内部字段 if key == "path": # 对于路径，显示绝对路径 abs_path = self.root / value print(f" {key}: {value}") print(f" {key} (绝对): {abs_path}") print(f" {key} (存在): {'是' if abs_path.exists() else '否'}") else: print(f" {key}: {value}") def _get_enabled_modules(self, config: dict[str, Any]) -> list[str]: """获取已启用的模块列表。""" env_config = config.get("env", {}) return env_config.get("modules", DEFAULT_MODULES) def _get_module_config(self, name: str, config: dict[str, Any]) -> dict[str, Any]: """获取模块配置。""" env_config = config.get("env", {}) # 尝试新格式 [env.模块名] module_config = env_config.get(name, {}) # 兼容旧格式：如果值是字符串而不是字典，转换为 {"path": value} if isinstance(module_config, str): module_config = {"path": module_config} # 兼容旧格式：如果没有配置但存在旧格式字段 if name == "venv" and not module_config: if "venv" in env_config and isinstance(env_config["venv"], str): module_config = {"path": env_config["venv"]} elif name == "requirements" and not module_config: if "requirements" in env_config and isinstance(env_config["requirements"], str): module_config = {"path": env_config["requirements"]} # 为 requirements 模块注入 venv 路径 if name == "requirements": venv_config = self._get_module_config("venv", config) if "path" in venv_config: module_config["_venv_path"] = venv_config["path"] # 从 runtime 配置获取 python 版本要求 if name == "python" and "min_version" not in module_config: runtime = config.get("runtime", {}) if "python_min" in runtime: module_config["min_version"] = runtime["python_min"] return module_config def _process_module( self, name: str, config: dict[str, Any], is_enabled: bool, check_only: bool, ) -> tuple[bool, str]: """处理单个模块的检测/修复。 Returns: (是否成功, 版本/路径信息) """ module = ModuleRegistry.get(name) if not module: if is_enabled: output.err(f"{name}: 未知模块") return False, "" else: output.warn(f"{name}: 未知模块") return True, "" module_config = self._get_module_config(name, config) # verbose: 输出模块配置 if self.verbose: self._print_verbose_module(name, module_config) # 检查类型B模块的配置 valid, err_msg = module.validate_config(module_config) if not valid: if is_enabled: output.err(f"{name}: 已启用但{err_msg}") return False, "" else: output.warn(f"{name}: {err_msg}，跳过检测") return True, "" # 执行检测 result = module.check(module_config, self.root) if result.success: version_info = result.version or "" extra = f" ({result.message})" if result.message else "" output.ok(f"{name}: {version_info}{extra}") return True, version_info # 检测失败 if check_only: # --all 模式：仅报告 output.warn(f"{name}: {result.message}") return True, "" if result.can_ensure and module.info.can_ensure: # 尝试修复 output.info(f"{name}: {result.message}，尝试修复...") ensure_result = module.ensure(module_config, self.root) if ensure_result.success: msg = ensure_result.message or "已修复" output.ok(f"{name}: {msg}") return True, ensure_result.version or "" else: if is_enabled: output.err(f"{name}: {ensure_result.message}") return False, "" else: output.warn(f"{name}: {ensure_result.message}") return True, "" else: # 不可修复 if is_enabled: extra = " (此模块不支持自动修复)" if not module.info.can_ensure else "" output.err(f"{name}: {result.message}{extra}") return False, "" else: output.warn(f"{name}: {result.message}") return True, ""