275 lines
9.5 KiB
Python
275 lines
9.5 KiB
Python
"""环境管理器。"""
|
||
|
||
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, ""
|