feat: 实现扩展模块和设置命令

This commit is contained in:
2025-12-14 07:35:29 +08:00
parent f016b51709
commit a48fbb5fae
19 changed files with 1415 additions and 255 deletions

View File

@@ -16,6 +16,40 @@ RUNTIME_MODULES = ["python", "uv"]
DEFAULT_MODULES = ["python", "uv", "venv", "requirements"]
def parse_module_name(name: str) -> tuple[str, str | None]:
"""解析模块名称,支持实例化命名。
Args:
name: 模块名称,如 "node_deps""node_deps:react"
Returns:
(模块类型, 实例名) - 实例名可能为 None
"""
if ":" in name:
parts = name.split(":", 1)
return parts[0], parts[1]
return name, None
def validate_modules(module_names: list[str]) -> tuple[bool, list[str]]:
"""验证模块名称是否有效。
Args:
module_names: 要验证的模块名称列表
Returns:
(是否全部有效, 无效的模块类型列表)
"""
register_builtin_modules()
available = set(ModuleRegistry.names())
invalid = []
for name in module_names:
module_type, _ = parse_module_name(name)
if module_type not in available:
invalid.append(module_type)
return len(invalid) == 0, invalid
class EnvManager:
"""环境管理器。"""
@@ -163,12 +197,20 @@ class EnvManager:
return env_config.get("modules", DEFAULT_MODULES)
def _get_module_config(self, name: str, config: dict[str, Any]) -> dict[str, Any]:
"""获取模块配置。"""
"""获取模块配置。
支持实例化命名,如 node_deps:react 会查找 [env."node_deps:react"]
"""
env_config = config.get("env", {})
# 尝试新格式 [env.模块名]
# 支持实例化命名:先尝试完整名称(如 node_deps:react
module_config = env_config.get(name, {})
# 如果没找到且是实例化命名,尝试不带引号的格式
if not module_config and ":" in name:
# TOML 中可能存储为 env."node_deps:react" 或嵌套格式
pass # 已经在上面尝试过了
# 兼容旧格式:如果值是字符串而不是字典,转换为 {"path": value}
if isinstance(module_config, str):
module_config = {"path": module_config}
@@ -204,10 +246,15 @@ class EnvManager:
) -> tuple[bool, str]:
"""处理单个模块的检测/修复。
支持实例化命名,如 node_deps:react
Returns:
(是否成功, 版本/路径信息)
"""
module = ModuleRegistry.get(name)
# 解析模块名称,支持实例化命名
module_type, instance_name = parse_module_name(name)
module = ModuleRegistry.get(module_type)
if not module:
if is_enabled:
output.err(f"{name}: 未知模块")
@@ -216,6 +263,7 @@ class EnvManager:
output.warn(f"{name}: 未知模块")
return True, ""
# 获取配置时使用完整名称(包含实例名)
module_config = self._get_module_config(name, config)
# verbose: 输出模块配置
@@ -272,3 +320,54 @@ class EnvManager:
else:
output.warn(f"{name}: {result.message}")
return True, ""
def set_modules(self, module_names: list[str]) -> bool:
"""设置启用的模块列表(带验证)。
Args:
module_names: 要启用的模块名称列表
Returns:
是否设置成功
"""
# 验证模块名称
valid, invalid = validate_modules(module_names)
if not valid:
available = ModuleRegistry.names()
output.err(f"未知模块: {', '.join(invalid)}")
output.info(f"可用模块: {', '.join(available)}")
return False
# 设置配置
self.cfg.set_value("env.modules", module_names)
return True
def set_module_config(self, module_name: str, key: str, value: Any) -> bool:
"""设置模块配置(带验证)。
支持实例化命名,如 node_deps:react.path
Args:
module_name: 模块名称(可包含实例名,如 node_deps:react
key: 配置键
value: 配置值
Returns:
是否设置成功
"""
# 解析模块名称,支持实例化命名
module_type, _ = parse_module_name(module_name)
# 验证模块类型是否存在
module = ModuleRegistry.get(module_type)
if not module:
available = ModuleRegistry.names()
output.err(f"未知模块: {module_type}")
output.info(f"可用模块: {', '.join(available)}")
return False
# 设置配置,使用完整模块名(包含实例名)
# 注意TOML 中带冒号的键需要引号,但 set_value 会自动处理
config_key = f"env.{module_name}.{key}"
self.cfg.set_value(config_key, value)
return True

146
aide-program/aide/env/modules/android.py vendored Normal file
View File

@@ -0,0 +1,146 @@
"""Android 开发环境检测模块。"""
from __future__ import annotations
import os
import subprocess
from pathlib import Path
from typing import Any
from aide.env.modules.base import BaseModule, CheckResult, ModuleInfo
class AndroidModule(BaseModule):
"""Android 开发环境检测模块类型A无需配置
检测 Android SDK 和相关工具:
- ANDROID_HOME / ANDROID_SDK_ROOT 环境变量
- Android SDK 目录结构
- 关键工具adb, aapt, sdkmanager
"""
@property
def info(self) -> ModuleInfo:
return ModuleInfo(
name="android",
description="Android SDK",
capabilities=["check"],
requires_config=False,
)
def check(self, config: dict[str, Any], root: Path) -> CheckResult:
"""检测 Android 开发环境。"""
# 检测 ANDROID_HOME 或 ANDROID_SDK_ROOT
sdk_root = self._get_sdk_root()
if not sdk_root:
return CheckResult(
success=False,
message="ANDROID_HOME 或 ANDROID_SDK_ROOT 未设置",
can_ensure=False,
)
sdk_path = Path(sdk_root)
if not sdk_path.exists():
return CheckResult(
success=False,
message=f"Android SDK 目录不存在: {sdk_root}",
can_ensure=False,
)
# 检测关键目录
platform_tools = sdk_path / "platform-tools"
build_tools = sdk_path / "build-tools"
platforms = sdk_path / "platforms"
missing = []
if not platform_tools.exists():
missing.append("platform-tools")
if not build_tools.exists():
missing.append("build-tools")
if not platforms.exists():
missing.append("platforms")
if missing:
return CheckResult(
success=False,
message=f"缺少 SDK 组件: {', '.join(missing)}",
can_ensure=False,
)
# 获取版本信息
build_tools_versions = self._get_build_tools_versions(build_tools)
platform_versions = self._get_platform_versions(platforms)
# 检测 adb
adb_version = self._get_adb_version(platform_tools)
# 构建版本信息
version_info = []
if adb_version:
version_info.append(f"adb {adb_version}")
if build_tools_versions:
version_info.append(f"build-tools {build_tools_versions[0]}")
if platform_versions:
version_info.append(f"API {platform_versions[0]}")
return CheckResult(
success=True,
version=sdk_root,
message=", ".join(version_info) if version_info else None,
)
def _get_sdk_root(self) -> str | None:
"""获取 Android SDK 根目录。"""
return os.environ.get("ANDROID_HOME") or os.environ.get("ANDROID_SDK_ROOT")
def _get_adb_version(self, platform_tools: Path) -> str | None:
"""获取 adb 版本。"""
adb_path = platform_tools / "adb"
if not adb_path.exists():
return None
try:
result = subprocess.run(
[str(adb_path), "version"],
capture_output=True,
text=True,
timeout=10,
)
if result.returncode == 0:
# Android Debug Bridge version 1.0.41
lines = result.stdout.strip().split("\n")
if lines:
parts = lines[0].split()
if len(parts) >= 5:
return parts[4]
return None
except (subprocess.TimeoutExpired, Exception):
return None
def _get_build_tools_versions(self, build_tools: Path) -> list[str]:
"""获取已安装的 build-tools 版本列表(降序)。"""
if not build_tools.exists():
return []
versions = []
for item in build_tools.iterdir():
if item.is_dir() and item.name[0].isdigit():
versions.append(item.name)
return sorted(versions, reverse=True)
def _get_platform_versions(self, platforms: Path) -> list[str]:
"""获取已安装的 platform 版本列表(降序)。"""
if not platforms.exists():
return []
versions = []
for item in platforms.iterdir():
if item.is_dir() and item.name.startswith("android-"):
api_level = item.name.replace("android-", "")
versions.append(api_level)
return sorted(versions, key=lambda x: int(x) if x.isdigit() else 0, reverse=True)
module = AndroidModule()

132
aide-program/aide/env/modules/flutter.py vendored Normal file
View File

@@ -0,0 +1,132 @@
"""Flutter SDK 检测模块。"""
from __future__ import annotations
import subprocess
from pathlib import Path
from typing import Any
from aide.env.modules.base import BaseModule, CheckResult, ModuleInfo
class FlutterModule(BaseModule):
"""Flutter SDK 检测模块类型A无需配置"""
@property
def info(self) -> ModuleInfo:
return ModuleInfo(
name="flutter",
description="Flutter SDK",
capabilities=["check"],
requires_config=False,
)
def check(self, config: dict[str, Any], root: Path) -> CheckResult:
"""检测 Flutter SDK。"""
flutter_version = self._get_flutter_version()
dart_version = self._get_dart_version()
if not flutter_version:
return CheckResult(
success=False,
message="flutter 未安装",
can_ensure=False,
)
# 检查最低版本要求(如果配置了)
min_version = config.get("min_version")
if min_version:
if not self._version_satisfies(flutter_version, min_version):
return CheckResult(
success=False,
version=flutter_version,
message=f"版本不足,要求>={min_version},当前 {flutter_version}",
can_ensure=False,
)
# 构建版本信息
extra = f"dart {dart_version}" if dart_version else ""
return CheckResult(
success=True,
version=flutter_version,
message=extra,
)
def _get_flutter_version(self) -> str | None:
"""获取 Flutter 版本。"""
try:
result = subprocess.run(
["flutter", "--version", "--machine"],
capture_output=True,
text=True,
timeout=30,
)
if result.returncode == 0:
# 尝试解析 JSON 输出
import json
try:
data = json.loads(result.stdout)
return data.get("frameworkVersion")
except json.JSONDecodeError:
pass
# 回退到普通版本输出
result = subprocess.run(
["flutter", "--version"],
capture_output=True,
text=True,
timeout=30,
)
if result.returncode == 0:
# Flutter 3.16.0 • channel stable • ...
output = result.stdout.strip()
lines = output.split("\n")
if lines:
parts = lines[0].split()
if len(parts) >= 2 and parts[0] == "Flutter":
return parts[1]
return None
except (FileNotFoundError, subprocess.TimeoutExpired):
return None
def _get_dart_version(self) -> str | None:
"""获取 Dart 版本。"""
try:
result = subprocess.run(
["dart", "--version"],
capture_output=True,
text=True,
timeout=10,
)
if result.returncode == 0:
# Dart SDK version: 3.2.0 (stable) ...
output = result.stdout.strip()
if not output:
output = result.stderr.strip() # dart 有时输出到 stderr
parts = output.split()
for i, part in enumerate(parts):
if part == "version:" and i + 1 < len(parts):
return parts[i + 1]
return None
except (FileNotFoundError, subprocess.TimeoutExpired):
return None
def _version_satisfies(self, current: str, minimum: str) -> bool:
"""检查版本是否满足最低要求。"""
current_parts = self._parse_version(current)
min_parts = self._parse_version(minimum)
return current_parts >= min_parts
@staticmethod
def _parse_version(version: str) -> tuple[int, ...]:
"""解析版本号字符串。"""
parts = []
for part in version.split("."):
try:
parts.append(int(part))
except ValueError:
break
return tuple(parts)
module = FlutterModule()

93
aide-program/aide/env/modules/node.py vendored Normal file
View File

@@ -0,0 +1,93 @@
"""Node.js 环境检测模块。"""
from __future__ import annotations
import subprocess
from pathlib import Path
from typing import Any
from aide.env.modules.base import BaseModule, CheckResult, ModuleInfo
class NodeModule(BaseModule):
"""Node.js 检测模块类型A无需配置"""
@property
def info(self) -> ModuleInfo:
return ModuleInfo(
name="node",
description="Node.js 运行时",
capabilities=["check"],
requires_config=False,
)
def check(self, config: dict[str, Any], root: Path) -> CheckResult:
"""检测 Node.js 版本。"""
node_version = self._get_version("node")
npm_version = self._get_version("npm")
if not node_version:
return CheckResult(
success=False,
message="node 未安装",
can_ensure=False,
)
# 检查最低版本要求(如果配置了)
min_version = config.get("min_version")
if min_version:
if not self._version_satisfies(node_version, min_version):
return CheckResult(
success=False,
version=node_version,
message=f"版本不足,要求>={min_version},当前 {node_version}",
can_ensure=False,
)
# 构建版本信息
extra = f"npm {npm_version}" if npm_version else "npm 未安装"
return CheckResult(
success=True,
version=node_version,
message=extra,
)
def _get_version(self, cmd: str) -> str | None:
"""获取命令版本。"""
try:
result = subprocess.run(
[cmd, "--version"],
capture_output=True,
text=True,
timeout=10,
)
if result.returncode == 0:
# node: v20.10.0 -> 20.10.0
# npm: 10.2.3 -> 10.2.3
output = result.stdout.strip()
if output.startswith("v"):
output = output[1:]
return output
return None
except (FileNotFoundError, subprocess.TimeoutExpired):
return None
def _version_satisfies(self, current: str, minimum: str) -> bool:
"""检查版本是否满足最低要求。"""
current_parts = self._parse_version(current)
min_parts = self._parse_version(minimum)
return current_parts >= min_parts
@staticmethod
def _parse_version(version: str) -> tuple[int, ...]:
"""解析版本号字符串。"""
parts = []
for part in version.split("."):
try:
parts.append(int(part))
except ValueError:
break
return tuple(parts)
module = NodeModule()

View File

@@ -0,0 +1,141 @@
"""Node.js 项目依赖检测模块。"""
from __future__ import annotations
import subprocess
from pathlib import Path
from typing import Any
from aide.env.modules.base import BaseModule, CheckResult, ModuleInfo
class NodeDepsModule(BaseModule):
"""Node.js 项目依赖检测模块类型B需要配置
支持多种包管理器npm, pnpm, yarn, bun
自动根据锁文件检测包管理器类型
"""
# 锁文件到包管理器的映射
LOCK_FILES = {
"pnpm-lock.yaml": "pnpm",
"yarn.lock": "yarn",
"bun.lockb": "bun",
"package-lock.json": "npm",
}
# 包管理器安装命令
INSTALL_COMMANDS = {
"npm": ["npm", "install"],
"pnpm": ["pnpm", "install"],
"yarn": ["yarn", "install"],
"bun": ["bun", "install"],
}
@property
def info(self) -> ModuleInfo:
return ModuleInfo(
name="node_deps",
description="Node.js 项目依赖",
capabilities=["check", "ensure"],
requires_config=True,
config_keys=["path"],
)
def check(self, config: dict[str, Any], root: Path) -> CheckResult:
"""检测 Node.js 项目依赖。"""
project_path = root / config["path"]
# 检测 package.json 是否存在
package_json = project_path / "package.json"
if not package_json.exists():
return CheckResult(
success=False,
message=f"package.json 不存在: {config['path']}",
can_ensure=False,
)
# 检测 node_modules 是否存在
node_modules = project_path / "node_modules"
if not node_modules.exists():
manager = self._detect_manager(project_path, config)
return CheckResult(
success=False,
message=f"node_modules 不存在",
can_ensure=True,
)
# 检测包管理器
manager = self._detect_manager(project_path, config)
return CheckResult(
success=True,
version=config["path"],
message=manager,
)
def ensure(self, config: dict[str, Any], root: Path) -> CheckResult:
"""安装 Node.js 项目依赖。"""
project_path = root / config["path"]
manager = self._detect_manager(project_path, config)
# 检测包管理器是否已安装
if not self._is_manager_installed(manager):
return CheckResult(
success=False,
message=f"{manager} 未安装",
)
# 运行安装命令
install_cmd = self.INSTALL_COMMANDS.get(manager, ["npm", "install"])
try:
subprocess.run(
install_cmd,
cwd=project_path,
check=True,
capture_output=True,
)
return CheckResult(
success=True,
version=config["path"],
message=f"已安装 ({manager})",
)
except subprocess.CalledProcessError as exc:
error_msg = exc.stderr.decode() if exc.stderr else str(exc)
return CheckResult(
success=False,
message=f"安装失败: {error_msg[:100]}",
)
def _detect_manager(self, project_path: Path, config: dict[str, Any]) -> str:
"""检测包管理器类型。
优先使用配置指定的 manager否则根据锁文件自动检测。
"""
# 优先使用配置指定的 manager
if "manager" in config:
return config["manager"]
# 根据锁文件检测
for lock_file, manager in self.LOCK_FILES.items():
if (project_path / lock_file).exists():
return manager
# 默认使用 npm
return "npm"
def _is_manager_installed(self, manager: str) -> bool:
"""检测包管理器是否已安装。"""
try:
subprocess.run(
[manager, "--version"],
capture_output=True,
timeout=10,
)
return True
except (FileNotFoundError, subprocess.TimeoutExpired):
return False
module = NodeDepsModule()

98
aide-program/aide/env/modules/rust.py vendored Normal file
View File

@@ -0,0 +1,98 @@
"""Rust 工具链检测模块。"""
from __future__ import annotations
import subprocess
from pathlib import Path
from typing import Any
from aide.env.modules.base import BaseModule, CheckResult, ModuleInfo
class RustModule(BaseModule):
"""Rust 工具链检测模块类型A无需配置"""
@property
def info(self) -> ModuleInfo:
return ModuleInfo(
name="rust",
description="Rust 工具链",
capabilities=["check"],
requires_config=False,
)
def check(self, config: dict[str, Any], root: Path) -> CheckResult:
"""检测 Rust 工具链rustc 和 cargo"""
rustc_version = self._get_version("rustc")
cargo_version = self._get_version("cargo")
if not rustc_version:
return CheckResult(
success=False,
message="rustc 未安装",
can_ensure=False,
)
if not cargo_version:
return CheckResult(
success=False,
message="cargo 未安装",
can_ensure=False,
)
# 检查最低版本要求(如果配置了)
min_version = config.get("min_version")
if min_version:
if not self._version_satisfies(rustc_version, min_version):
return CheckResult(
success=False,
version=rustc_version,
message=f"版本不足,要求>={min_version},当前 {rustc_version}",
can_ensure=False,
)
return CheckResult(
success=True,
version=rustc_version,
message=f"cargo {cargo_version}",
)
def _get_version(self, cmd: str) -> str | None:
"""获取命令版本。"""
try:
result = subprocess.run(
[cmd, "--version"],
capture_output=True,
text=True,
timeout=10,
)
if result.returncode == 0:
# rustc 1.75.0 (xxx) -> 1.75.0
# cargo 1.75.0 (xxx) -> 1.75.0
output = result.stdout.strip()
parts = output.split()
if len(parts) >= 2:
return parts[1]
return None
except (FileNotFoundError, subprocess.TimeoutExpired):
return None
def _version_satisfies(self, current: str, minimum: str) -> bool:
"""检查版本是否满足最低要求。"""
current_parts = self._parse_version(current)
min_parts = self._parse_version(minimum)
return current_parts >= min_parts
@staticmethod
def _parse_version(version: str) -> tuple[int, ...]:
"""解析版本号字符串。"""
parts = []
for part in version.split("."):
try:
parts.append(int(part))
except ValueError:
break
return tuple(parts)
module = RustModule()

View File

@@ -43,8 +43,12 @@ class ModuleRegistry:
def register_builtin_modules() -> None:
"""注册内置模块。"""
from aide.env.modules import python, uv, venv, requirements
from aide.env.modules import (
python, uv, venv, requirements,
rust, node, flutter,
node_deps, android,
)
for mod in [python, uv, venv, requirements]:
for mod in [python, uv, venv, requirements, rust, node, flutter, node_deps, android]:
if hasattr(mod, "module"):
ModuleRegistry.register(mod.module)

View File

@@ -69,6 +69,12 @@ def build_parser() -> argparse.ArgumentParser:
list_parser = env_sub.add_parser("list", help="列出所有可用模块")
list_parser.set_defaults(func=handle_env_list)
# aide env set
set_parser = env_sub.add_parser("set", help="设置环境配置(带验证)")
set_parser.add_argument("key", help="配置键modules 或 模块名.配置项")
set_parser.add_argument("value", help="配置值")
set_parser.set_defaults(func=handle_env_set)
# aide env无子命令时等同于 ensure
env_parser.set_defaults(func=handle_env_default)
@@ -137,6 +143,34 @@ def handle_env_list(args: argparse.Namespace) -> bool:
return True
def handle_env_set(args: argparse.Namespace) -> bool:
"""aide env set 处理。"""
root = Path.cwd()
cfg = ConfigManager(root)
manager = EnvManager(root, cfg)
key = args.key
value = args.value
if key == "modules":
# 设置启用的模块列表
module_names = [m.strip() for m in value.split(",") if m.strip()]
return manager.set_modules(module_names)
elif "." in key:
# 设置模块配置,如 venv.path
parts = key.split(".", 1)
module_name = parts[0]
config_key = parts[1]
parsed_value = _parse_value(value)
return manager.set_module_config(module_name, config_key, parsed_value)
else:
# 无效的键格式
output.err(f"无效的配置键: {key}")
output.info("用法: aide env set modules <模块列表>")
output.info(" aide env set <模块名>.<配置项> <值>")
return False
def handle_config_get(args: argparse.Namespace) -> bool:
root = Path.cwd()
cfg = ConfigManager(root)