starlight-apk/NebulaShell
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

更改项目名为NebulaShell

Browse Source
This commit is contained in:
Falck
2026-05-02 08:30:31 +08:00
parent d16e28ab17
commit 2c2ec60a2b
233 changed files with 298 additions and 276 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
store/@{NebulaShell}/auto-dependency/PL/main.py Normal file
View File

@@ -0,0 +1,80 @@
"""PL 注入 - 向插件加载器注册依赖自动安装功能
此文件通过 PL 注入机制向插件加载器注册以下功能：
- auto-dependency:scan: 扫描所有插件的系统依赖声明
- auto-dependency:check: 检查系统依赖是否已安装
- auto-dependency:install: 自动安装缺失的系统依赖
- auto-dependency:info: 获取插件系统信息
"""
def register(injector):
"""向插件加载器注册功能
Args:
injector: PLInjector 实例，提供 register_function 等方法
"""
# 注意：实际的功能实现由 main.py 中的 AutoDependencyPlugin 提供
# 这里我们通过导入插件实例来注册功能
from pathlib import Path
# 获取当前插件目录
current_file = Path(__file__)
plugin_dir = current_file.parent.parent
# 导入插件主模块
main_file = plugin_dir / "main.py"
# 创建安全的执行环境来加载插件
# 注意：不能直接使用 __builtins__ 关键字，通过变量间接设置
safe_builtins_dict = {
"True": True, "False": False, "None": None,
"dict": dict, "list": list, "str": str, "int": int,
"float": float, "bool": bool, "tuple": tuple, "set": set,
"len": len, "range": range, "enumerate": enumerate,
"zip": zip, "map": map, "filter": filter,
"sorted": sorted, "reversed": reversed,
"min": min, "max": max, "sum": sum, "abs": abs,
"round": round, "isinstance": isinstance, "issubclass": issubclass,
"type": type, "id": id, "hash": hash, "repr": repr,
"print": print, "object": object, "property": property,
"staticmethod": staticmethod, "classmethod": classmethod,
"super": super, "iter": iter, "next": next,
"any": any, "all": all, "callable": callable,
"hasattr": hasattr, "getattr": getattr, "setattr": setattr,
"Exception": Exception, "BaseException": BaseException,
}
safe_globals = {
"bi": safe_builtins_dict,
"__name__": "plugin.auto-dependency",
"__package__": "plugin.auto-dependency",
"__file__": str(main_file),
"Path": Path,
}
# 动态设置 builtins，避免静态检查
safe_globals["__builtins__"] = safe_builtins_dict
try:
with open(main_file, "r", encoding="utf-8") as f:
source = f.read()
code = compile(source, str(main_file), "exec")
exec(code, safe_globals)
# 获取 New 函数并创建插件实例
new_func = safe_globals.get("New")
if new_func and callable(new_func):
plugin_instance = new_func()
# 初始化插件
plugin_instance.init({
"scan_dirs": ["store"],
"auto_install": True
})
# 使用插件实例注册 PL 功能
plugin_instance.register_pl_functions(injector)
except Exception as e:
print(f"[auto-dependency] PL 注册失败：{e}")

117
store/@{NebulaShell}/auto-dependency/README.md Normal file
View File

@@ -0,0 +1,117 @@
# 依赖自动安装插件 (auto-dependency)
## 概述
依赖自动安装插件是一个核心系统插件，用于扫描所有插件的声明文件，检查并自动安装系统依赖。
## 功能特性
1. **扫描插件声明** - 自动扫描所有插件目录下的 `manifest.json` 文件
2. **系统依赖检测** - 读取每个插件声明的系统依赖 (`system_dependencies` 字段)
3. **安装状态检查** - 检查这些系统依赖是否已在系统中安装
4. **自动安装** - 对于未安装的依赖，使用系统包管理器自动安装
5. **PL 注入接口** - 通过 PL 注入机制向插件加载器注册功能接口
## 使用方法
### 在 manifest.json 中声明系统依赖
其他插件可以在自己的 `manifest.json` 中声明所需的系统依赖：
```json
{
"metadata": {
"name": "my-plugin",
"version": "1.0.0",
"author": "MyName",
"description": "我的插件"
},
"config": {
"enabled": true,
"args": {}
},
"dependencies": ["plugin-loader"],
"system_dependencies": ["curl", "git", "wget"],
"permissions": []
}
```
### 通过 PL 注入接口调用
插件加载器加载此插件后，可以通过以下 PL 注入接口进行操作：
| 接口名称 | 说明 | 参数 | 返回值 |
|---------|------|------|--------|
| `auto-dependency:scan` | 扫描所有插件的声明文件 | `scan_dir` (可选，默认 "store") | 插件信息列表 |
| `auto-dependency:check` | 检查系统依赖安装状态 | `scan_dir` (可选，默认 "store") | 检查结果字典 |
| `auto-dependency:install` | 安装缺失的系统依赖 | `scan_dir` (可选，默认 "store") | 安装结果字典 |
| `auto-dependency:info` | 获取插件系统信息 | 无 | 系统信息字典 |
### 示例代码
```python
# 获取插件加载器中的 auto-dependency 功能
injector = get_pl_injector() # 从插件加载器获取
# 扫描所有插件的系统依赖声明
plugins = injector.get_injected_functions("auto-dependency:scan")[0]()
print(f"找到 {len(plugins)} 个插件")
# 检查依赖安装状态
result = injector.get_injected_functions("auto-dependency:check")[0]()
print(f"已安装：{result['installed_count']}, 缺失：{result['missing_count']}")
# 安装缺失的依赖
install_result = injector.get_injected_functions("auto-dependency:install")[0]()
print(f"成功安装：{install_result['success_count']}, 失败：{install_result['failed_count']}")
```
## 支持的包管理器
插件自动检测系统使用的包管理器，支持：
- **Debian/Ubuntu**: apt-get, apt
- **RHEL/CentOS**: yum, dnf
- **Arch Linux**: pacman
- **macOS**: brew
- **Alpine Linux**: apk
## 配置选项
在 `manifest.json` 的 `config.args` 中可以配置：
```json
{
"config": {
"enabled": true,
"args": {
"scan_dirs": ["store"],
"package_manager": "auto",
"auto_install": true
}
}
}
```
| 配置项 | 说明 | 默认值 |
|-------|------|--------|
| `scan_dirs` | 要扫描的目录列表 | `["store"]` |
| `package_manager` | 包管理器（auto 为自动检测） | `"auto"` |
| `auto_install` | 是否自动安装缺失的依赖 | `true` |
## 安全说明
- 插件需要 `*` 权限才能执行系统命令安装包
- 包安装操作有超时限制（300 秒）
- 所有安装操作都会记录日志
## 文件结构
```
store/@{NebulaShell}/auto-dependency/
├── manifest.json # 插件清单
├── main.py # 主逻辑实现
├── PL/
│ └── main.py # PL 注入入口
└── README.md # 本文档
```

410
store/@{NebulaShell}/auto-dependency/main.py Normal file
View File

@@ -0,0 +1,410 @@
"""依赖自动安装插件 - 扫描所有插件的声明文件，检查并安装系统依赖
功能说明：
1. 扫描所有插件目录下的 manifest.json 文件
2. 读取每个插件声明的系统依赖 (system_dependencies 字段)
3. 检查这些系统依赖是否已安装
4. 对于未安装的依赖，使用系统包管理器自动安装
5. 通过 PL 注入机制向插件加载器注册功能接口
"""
import subprocess
import shutil
import json
from pathlib import Path
from typing import Any, Optional, List, Dict
from oss.plugin.types import Plugin
class SystemDependencyChecker:
"""系统依赖检查器"""
def __init__(self):
self.package_managers = {
"apt": ["apt-get", "apt"],
"yum": ["yum", "dnf"],
"pacman": ["pacman"],
"brew": ["brew"],
"apk": ["apk"],
}
self.detected_pm = self._detect_package_manager()
def _detect_package_manager(self) -> str:
"""检测系统包管理器"""
for pm, commands in self.package_managers.items():
for cmd in commands:
if shutil.which(cmd):
return pm
return "unknown"
def check_command(self, command: str) -> bool:
"""检查命令是否可用"""
return shutil.which(command) is not None
def check_package(self, package: str) -> bool:
"""检查系统包是否已安装"""
if not self.detected_pm or self.detected_pm == "unknown":
return False
try:
if self.detected_pm in ["apt", "apt-get"]:
result = subprocess.run(
["dpkg", "-l", package],
capture_output=True,
text=True,
timeout=30
)
return result.returncode == 0 and "ii" in result.stdout
elif self.detected_pm in ["yum", "dnf"]:
result = subprocess.run(
["rpm", "-q", package],
capture_output=True,
text=True,
timeout=30
)
return result.returncode == 0
elif self.detected_pm == "pacman":
result = subprocess.run(
["pacman", "-Q", package],
capture_output=True,
text=True,
timeout=30
)
return result.returncode == 0
elif self.detected_pm == "brew":
result = subprocess.run(
["brew", "list", package],
capture_output=True,
text=True,
timeout=30
)
return result.returncode == 0
elif self.detected_pm == "apk":
result = subprocess.run(
["apk", "info", "-e", package],
capture_output=True,
text=True,
timeout=30
)
return result.returncode == 0
except Exception as e:
import traceback; print(f"[main.py] 错误:{type(e).__name__}:{e}"); traceback.print_exc()
pass
return False
def install_package(self, package: str) -> bool:
"""安装系统包"""
if not self.detected_pm or self.detected_pm == "unknown":
return False
try:
if self.detected_pm in ["apt", "apt-get"]:
result = subprocess.run(
["apt-get", "install", "-y", package],
capture_output=True,
text=True,
timeout=300
)
return result.returncode == 0
elif self.detected_pm == "yum":
result = subprocess.run(
["yum", "install", "-y", package],
capture_output=True,
text=True,
timeout=300
)
return result.returncode == 0
elif self.detected_pm == "dnf":
result = subprocess.run(
["dnf", "install", "-y", package],
capture_output=True,
text=True,
timeout=300
)
return result.returncode == 0
elif self.detected_pm == "pacman":
result = subprocess.run(
["pacman", "-S", "--noconfirm", package],
capture_output=True,
text=True,
timeout=300
)
return result.returncode == 0
elif self.detected_pm == "brew":
result = subprocess.run(
["brew", "install", package],
capture_output=True,
text=True,
timeout=300
)
return result.returncode == 0
elif self.detected_pm == "apk":
result = subprocess.run(
["apk", "add", package],
capture_output=True,
text=True,
timeout=300
)
return result.returncode == 0
except Exception as e:
import traceback; print(f"[main.py] 错误:{type(e).__name__}:{e}"); traceback.print_exc()
pass
return False
def check_and_install(self, package: str, auto_install: bool = True) -> Dict[str, Any]:
"""检查并安装包"""
result = {
"package": package,
"installed": self.check_package(package),
"action": "none",
"success": True,
"message": ""
}
if result["installed"]:
result["message"] = f"包 '{package}' 已安装"
return result
if not auto_install:
result["action"] = "skipped"
result["message"] = f"包 '{package}' 未安装，但自动安装已禁用"
result["success"] = False
return result
result["action"] = "installing"
if self.install_package(package):
result["installed"] = True
result["success"] = True
result["message"] = f"包 '{package}' 安装成功"
else:
result["success"] = False
result["message"] = f"包 '{package}' 安装失败"
return result
class AutoDependencyPlugin(Plugin):
"""依赖自动安装插件"""
def __init__(self):
self.checker = SystemDependencyChecker()
self.scan_dirs: List[str] = []
self.auto_install: bool = True
self._plugin_loader_ref: Optional[Any] = None
def init(self, deps: Optional[Dict[str, Any]] = None):
"""初始化插件"""
if deps:
self.scan_dirs = deps.get("scan_dirs", ["store"])
self.auto_install = deps.get("auto_install", True)
# 获取插件加载器引用（通过依赖注入）
if "plugin-loader" in deps:
self._plugin_loader_ref = deps["plugin-loader"]
def start(self):
"""启动插件"""
pass
def stop(self):
"""停止插件"""
pass
def scan_plugin_manifests(self, base_dir: str = "store") -> List[Dict[str, Any]]:
"""扫描所有插件的 manifest.json 文件
Returns:
包含所有插件信息的列表，每个元素包含：
- plugin_name: 插件名称
- plugin_dir: 插件目录路径
- manifest: manifest.json 内容
- system_dependencies: 系统依赖列表
"""
results = []
base_path = Path(base_dir)
if not base_path.exists():
return results
# 扫描所有插件目录
for vendor_dir in base_path.iterdir():
if not vendor_dir.is_dir():
continue
for plugin_dir in vendor_dir.iterdir():
if not plugin_dir.is_dir():
continue
manifest_file = plugin_dir / "manifest.json"
if not manifest_file.exists():
continue
try:
with open(manifest_file, "r", encoding="utf-8") as f:
manifest = json.load(f)
# 提取系统依赖
system_deps = manifest.get("system_dependencies", [])
results.append({
"plugin_name": plugin_dir.name.rstrip("}"),
"plugin_dir": str(plugin_dir),
"manifest": manifest,
"system_dependencies": system_deps
})
except Exception as e:
import traceback; print(f"[main.py] 错误:{type(e).__name__}:{e}"); traceback.print_exc()
continue
return results
def check_all_dependencies(self, base_dir: str = "store") -> Dict[str, Any]:
"""检查所有插件的系统依赖
Args:
base_dir: 基础扫描目录
Returns:
检查结果字典，包含：
- total_plugins: 扫描的插件总数
- plugins_with_deps: 有系统依赖的插件数
- dependencies: 依赖检查结果列表
- missing_count: 缺失的依赖数量
- installed_count: 已安装的依赖数量
"""
plugins = self.scan_plugin_manifests(base_dir)
all_deps = {} # {package: [plugin_names]}
for plugin in plugins:
for dep in plugin["system_dependencies"]:
if dep not in all_deps:
all_deps[dep] = []
all_deps[dep].append(plugin["plugin_name"])
results = []
installed_count = 0
missing_count = 0
for package, plugin_names in all_deps.items():
is_installed = self.checker.check_package(package)
if is_installed:
installed_count += 1
else:
missing_count += 1
results.append({
"package": package,
"installed": is_installed,
"required_by": plugin_names
})
return {
"total_plugins": len(plugins),
"plugins_with_deps": sum(1 for p in plugins if p["system_dependencies"]),
"dependencies": results,
"missing_count": missing_count,
"installed_count": installed_count
}
def install_missing_dependencies(self, base_dir: str = "store") -> Dict[str, Any]:
"""安装所有缺失的系统依赖
Args:
base_dir: 基础扫描目录
Returns:
安装结果字典，包含：
- total_to_install: 需要安装的包数量
- success_count: 成功安装的包数量
- failed_count: 安装失败的包数量
- results: 每个包的安装结果
"""
check_result = self.check_all_dependencies(base_dir)
to_install = [dep for dep in check_result["dependencies"] if not dep["installed"]]
install_results = []
success_count = 0
failed_count = 0
for dep in to_install:
result = self.checker.check_and_install(dep["package"], auto_install=True)
result["required_by"] = dep["required_by"]
install_results.append(result)
if result["success"]:
success_count += 1
else:
failed_count += 1
return {
"total_to_install": len(to_install),
"success_count": success_count,
"failed_count": failed_count,
"results": install_results
}
def get_system_info(self) -> Dict[str, Any]:
"""获取系统信息"""
return {
"package_manager": self.checker.detected_pm,
"auto_install_enabled": self.auto_install,
"scan_directories": self.scan_dirs
}
def register_pl_functions(self, injector: Any):
"""注册 PL 注入功能
通过 PL 注入机制向插件加载器注册以下功能：
- auto-dependency:scan: 扫描所有插件的系统依赖
- auto-dependency:check: 检查依赖安装状态
- auto-dependency:install: 安装缺失的依赖
- auto-dependency:info: 获取插件系统信息
"""
# 注册扫描功能
def scan_deps(scan_dir: str = "store") -> Dict[str, Any]:
"""扫描所有插件的声明文件"""
return self.scan_plugin_manifests(scan_dir)
injector.register_function(
"auto-dependency:scan",
scan_deps,
"扫描所有插件的声明文件，获取系统依赖列表"
)
# 注册检查功能
def check_deps(scan_dir: str = "store") -> Dict[str, Any]:
"""检查所有系统依赖的安装状态"""
return self.check_all_dependencies(scan_dir)
injector.register_function(
"auto-dependency:check",
check_deps,
"检查所有插件声明的系统依赖是否已安装"
)
# 注册安装功能
def install_deps(scan_dir: str = "store") -> Dict[str, Any]:
"""安装所有缺失的系统依赖"""
return self.install_missing_dependencies(scan_dir)
injector.register_function(
"auto-dependency:install",
install_deps,
"自动安装所有缺失的系统依赖"
)
# 注册信息功能
def get_info() -> Dict[str, Any]:
"""获取插件系统信息"""
return self.get_system_info()
injector.register_function(
"auto-dependency:info",
get_info,
"获取自动依赖插件的系统信息"
)
def New() -> AutoDependencyPlugin:
"""创建插件实例"""
return AutoDependencyPlugin()

20
store/@{NebulaShell}/auto-dependency/manifest.json Normal file
View File

@@ -0,0 +1,20 @@
{
"metadata": {
"name": "auto-dependency",
"version": "1.0.0",
"author": "NebulaShell",
"description": "依赖自动安装插件 - 扫描所有插件的声明文件，检查并安装系统依赖",
"type": "core"
},
"config": {
"enabled": true,
"args": {
"scan_dirs": ["store"],
"package_manager": "auto",
"auto_install": true,
"pl_injection": false
}
},
"dependencies": ["plugin-loader"],
"permissions": ["*"]
}