486 lines
20 KiB
Python
486 lines
20 KiB
Python
"""包管理插件 - 提供插件配置管理和商店界面"""
|
||
import os
|
||
import sys
|
||
import json
|
||
import urllib.request
|
||
from pathlib import Path
|
||
from oss.logger.logger import Log
|
||
from oss.plugin.types import Plugin, Response, register_plugin_type
|
||
|
||
|
||
# Gitee 仓库配置
|
||
GITEE_OWNER = "starlight-apk"
|
||
GITEE_REPO = "future-oss"
|
||
GITEE_BRANCH = "main"
|
||
# 使用 raw 文件 URL（不走 API，无频率限制）
|
||
GITEE_RAW_BASE = f"https://gitee.com/{GITEE_OWNER}/{GITEE_REPO}/raw/{GITEE_BRANCH}"
|
||
GITEE_API_BASE = f"https://gitee.com/api/v5/repos/{GITEE_OWNER}/{GITEE_REPO}/contents"
|
||
# Gitee Token（从环境变量读取，可选）
|
||
GITEE_TOKEN = os.environ.get("GITEE_TOKEN", "")
|
||
|
||
|
||
def _gitee_request(url: str, timeout: int = 15):
|
||
"""Gitee 请求"""
|
||
req = urllib.request.Request(url)
|
||
req.add_header("User-Agent", "FutureOSS-PkgManager")
|
||
if GITEE_TOKEN:
|
||
# Gitee 使用私人令牌认证
|
||
req.add_header("Authorization", f"token {GITEE_TOKEN}")
|
||
return urllib.request.urlopen(req, timeout=timeout)
|
||
|
||
|
||
class PkgManagerPlugin(Plugin):
|
||
"""包管理插件"""
|
||
|
||
def __init__(self):
|
||
self.webui = None
|
||
self.storage = None
|
||
self.store_dir = Path("./store")
|
||
self._remote_cache = None
|
||
self._cache_time = 0
|
||
self._cache_ttl = 300 # 5分钟缓存
|
||
|
||
def meta(self):
|
||
from oss.plugin.types import Metadata, PluginConfig, Manifest
|
||
return Manifest(
|
||
metadata=Metadata(
|
||
name="pkg-manager",
|
||
version="1.0.0",
|
||
author="FutureOSS",
|
||
description="插件包管理器 - 配置管理和商店"
|
||
),
|
||
config=PluginConfig(enabled=True, args={}),
|
||
dependencies=["http-api", "webui", "plugin-storage"]
|
||
)
|
||
|
||
def set_webui(self, webui):
|
||
self.webui = webui
|
||
|
||
def set_plugin_storage(self, storage):
|
||
self.storage = storage
|
||
|
||
def init(self, deps: dict = None):
|
||
"""init 阶段：注册页面到 WebUI"""
|
||
if not self.webui:
|
||
Log.warn("pkg-manager", "警告: 未找到 WebUI 依赖")
|
||
return
|
||
|
||
self.webui.register_page(
|
||
path='/packages',
|
||
content_provider=self._packages_content,
|
||
nav_item={'icon': 'ri-apps-line', 'text': '插件管理'}
|
||
)
|
||
self.webui.register_page(
|
||
path='/store',
|
||
content_provider=self._store_content,
|
||
nav_item={'icon': 'ri-store-2-line', 'text': '插件商店'}
|
||
)
|
||
Log.info("pkg-manager", "已注册到 WebUI 导航")
|
||
|
||
def start(self):
|
||
"""启动阶段：注册 API 路由"""
|
||
if not self.webui or not hasattr(self.webui, 'server') or not self.webui.server:
|
||
Log.warn("pkg-manager", "警告: WebUI 服务器未就绪")
|
||
return
|
||
|
||
router = self.webui.server.router
|
||
|
||
# API - 已安装插件
|
||
router.get("/api/plugins", self._handle_list_plugins)
|
||
router.get("/api/plugins/:name/config", self._handle_get_config)
|
||
router.post("/api/plugins/:name/config", self._handle_save_config)
|
||
router.get("/api/plugins/:name/info", self._handle_get_plugin_info)
|
||
router.post("/api/plugins/:name/uninstall", self._handle_uninstall)
|
||
|
||
# API - 远程商店
|
||
router.get("/api/store/remote", self._handle_remote_store)
|
||
router.post("/api/store/install", self._handle_store_install)
|
||
|
||
Log.info("pkg-manager", "包管理器已启动")
|
||
|
||
def stop(self):
|
||
Log.error("pkg-manager", "包管理器已停止")
|
||
|
||
# ==================== 页面渲染 ====================
|
||
|
||
def _packages_content(self) -> str:
|
||
return self._render_php_view('packages.php', {'pageTitle': '插件管理'})
|
||
|
||
def _store_content(self) -> str:
|
||
return self._render_php_view('store.php', {'pageTitle': '插件商店'})
|
||
|
||
def _render_php_view(self, view_name: str, variables: dict) -> str:
|
||
import subprocess
|
||
|
||
views_dir = os.path.join(os.path.dirname(__file__), 'views')
|
||
php_file = os.path.join(views_dir, view_name)
|
||
if not os.path.exists(php_file):
|
||
return f"<h1>错误: 找不到 {view_name}</h1>"
|
||
|
||
php_vars = ""
|
||
for key, value in variables.items():
|
||
if isinstance(value, str):
|
||
php_vars += f"${key} = '{value}';\n"
|
||
else:
|
||
php_vars += f"${key} = {json.dumps(value)};\n"
|
||
|
||
with open(php_file, 'r', encoding='utf-8') as f:
|
||
php_content = f.read()
|
||
|
||
tmp_file = os.path.join(views_dir, '.temp_pkg.php')
|
||
try:
|
||
with open(tmp_file, 'w', encoding='utf-8') as f:
|
||
f.write(f"<?php\n{php_vars}\n?>\n{php_content}")
|
||
|
||
result = subprocess.run(
|
||
["php", "-f", tmp_file],
|
||
capture_output=True, text=True, timeout=10, cwd=views_dir
|
||
)
|
||
return result.stdout if result.returncode == 0 else f"<pre>{result.stderr}</pre>"
|
||
finally:
|
||
try:
|
||
if os.path.exists(tmp_file):
|
||
os.unlink(tmp_file)
|
||
except:
|
||
pass
|
||
|
||
# ==================== API 处理 ====================
|
||
|
||
def _handle_list_plugins(self, request):
|
||
"""列出所有已安装插件"""
|
||
plugins = self._scan_all_plugins()
|
||
return Response(status=200, headers={"Content-Type": "application/json"}, body=json.dumps(plugins, ensure_ascii=False))
|
||
|
||
def _handle_get_config(self, request):
|
||
"""获取插件配置 schema + 当前值"""
|
||
plugin_name = request.path_params.get('name', '')
|
||
schema = self._load_config_schema(plugin_name)
|
||
current = self._load_plugin_config(plugin_name)
|
||
return Response(status=200, headers={"Content-Type": "application/json"}, body=json.dumps({
|
||
"schema": schema,
|
||
"current": current
|
||
}, ensure_ascii=False))
|
||
|
||
def _handle_save_config(self, request):
|
||
"""保存插件配置"""
|
||
import json as json_mod
|
||
try:
|
||
body = json_mod.loads(request.body)
|
||
plugin_name = request.path_params.get('name', '')
|
||
self._save_plugin_config(plugin_name, body)
|
||
return Response(status=200, headers={"Content-Type": "application/json"}, body='{"ok":true}')
|
||
except Exception as e:
|
||
return Response(status=500, headers={"Content-Type": "application/json"}, body=json.dumps({"error": str(e)}))
|
||
|
||
def _handle_get_plugin_info(self, request):
|
||
"""获取插件详细信息"""
|
||
plugin_name = request.path_params.get('name', '')
|
||
info = self._get_plugin_detailed_info(plugin_name)
|
||
return Response(status=200, headers={"Content-Type": "application/json"}, body=json.dumps(info, ensure_ascii=False))
|
||
|
||
def _handle_uninstall(self, request):
|
||
"""卸载插件"""
|
||
import shutil
|
||
plugin_name = request.path_params.get('name', '')
|
||
# 查找插件目录
|
||
plugin_dir = self._find_plugin_dir(plugin_name)
|
||
if not plugin_dir:
|
||
return Response(status=404, body='{"error":"插件未安装"}')
|
||
try:
|
||
shutil.rmtree(plugin_dir)
|
||
return Response(status=200, body='{"ok":true}')
|
||
except Exception as e:
|
||
return Response(status=500, body=json.dumps({"error": str(e)}))
|
||
|
||
def _handle_remote_store(self, request):
|
||
"""从 Gitee API 获取远程插件列表"""
|
||
try:
|
||
plugins = self._fetch_remote_plugins()
|
||
return Response(status=200, headers={"Content-Type": "application/json"}, body=json.dumps(plugins, ensure_ascii=False))
|
||
except Exception as e:
|
||
return Response(status=500, body=json.dumps({"error": str(e)}))
|
||
|
||
def _handle_store_install(self, request):
|
||
"""安装插件"""
|
||
import json as json_mod
|
||
try:
|
||
body = json_mod.loads(request.body)
|
||
name = body.get("name", "")
|
||
author = body.get("author", "FutureOSS")
|
||
success = self._install_from_gitee(name, author)
|
||
return Response(status=200, body=json.dumps({"ok": success}))
|
||
except Exception as e:
|
||
return Response(status=500, body=json.dumps({"error": str(e)}))
|
||
|
||
# ==================== Gitee 远程商店 ====================
|
||
|
||
def _fetch_remote_plugins(self) -> list:
|
||
"""从 Gitee 获取所有可用插件（带缓存+限速+重试）"""
|
||
import time
|
||
now = time.time()
|
||
if self._remote_cache and (now - self._cache_time) < self._cache_ttl:
|
||
return self._remote_cache
|
||
|
||
plugins = []
|
||
try:
|
||
store_url = f"{GITEE_API_BASE}/store"
|
||
# 重试 3 次，每次间隔增加
|
||
for attempt in range(3):
|
||
try:
|
||
with _gitee_request(store_url, timeout=15) as resp:
|
||
dirs = json.loads(resp.read().decode("utf-8"))
|
||
break
|
||
except Exception as e:
|
||
if attempt < 2:
|
||
time.sleep(1 + attempt)
|
||
continue
|
||
raise
|
||
|
||
time.sleep(0.5)
|
||
|
||
for dir_info in dirs:
|
||
if dir_info.get("type") != "dir":
|
||
continue
|
||
author = dir_info.get("name", "")
|
||
if not author.startswith("@{"):
|
||
continue
|
||
|
||
author_url = f"{GITEE_API_BASE}/store/{urllib.parse.quote(author, safe='')}"
|
||
for attempt in range(3):
|
||
try:
|
||
with _gitee_request(author_url, timeout=15) as resp:
|
||
plugin_dirs = json.loads(resp.read().decode("utf-8"))
|
||
break
|
||
except Exception:
|
||
if attempt < 2:
|
||
time.sleep(1 + attempt)
|
||
continue
|
||
raise
|
||
|
||
time.sleep(0.5)
|
||
|
||
for plugin_dir in plugin_dirs:
|
||
if plugin_dir.get("type") != "dir":
|
||
continue
|
||
plugin_name = plugin_dir.get("name", "")
|
||
|
||
manifest_url = f"{GITEE_API_BASE}/store/{urllib.parse.quote(author, safe='')}/{plugin_name}/manifest.json"
|
||
manifest = {}
|
||
for attempt in range(3):
|
||
try:
|
||
with _gitee_request(manifest_url, timeout=15) as resp:
|
||
manifest = json.loads(resp.read().decode("utf-8"))
|
||
break
|
||
except Exception:
|
||
if attempt < 2:
|
||
time.sleep(1 + attempt)
|
||
continue
|
||
|
||
plugins.append({
|
||
"name": plugin_name,
|
||
"author": author,
|
||
"full_name": f"{author}/{plugin_name}",
|
||
"metadata": manifest.get("metadata", {}),
|
||
"dependencies": manifest.get("dependencies", []),
|
||
"has_config": False,
|
||
"is_installed": self._is_plugin_installed(plugin_name, author)
|
||
})
|
||
|
||
time.sleep(0.5)
|
||
|
||
self._remote_cache = plugins
|
||
self._cache_time = now
|
||
except Exception as e:
|
||
Log.error("pkg-manager", f"获取远程插件列表失败: {e}")
|
||
|
||
return plugins
|
||
|
||
def _install_from_gitee(self, plugin_name: str, author: str) -> bool:
|
||
"""从 Gitee 下载并安装插件（使用 raw URL）"""
|
||
import shutil, time
|
||
install_dir = self.store_dir / author / plugin_name
|
||
install_dir.mkdir(parents=True, exist_ok=True)
|
||
|
||
try:
|
||
# 获取目录结构（需要一次 API 调用）
|
||
api_url = f"{GITEE_API_BASE}/store/{author}/{plugin_name}"
|
||
with _gitee_request(api_url, timeout=15) as resp:
|
||
items = json.loads(resp.read().decode("utf-8"))
|
||
|
||
time.sleep(0.5)
|
||
|
||
for item in items:
|
||
if item.get("type") == "file":
|
||
# 使用 raw URL 下载文件（不走 API）
|
||
filename = item.get("name")
|
||
raw_url = f"{GITEE_RAW_BASE}/store/{author}/{plugin_name}/{filename}"
|
||
local_file = install_dir / filename
|
||
try:
|
||
with _gitee_request(raw_url, timeout=15) as resp:
|
||
content = resp.read()
|
||
with open(local_file, 'wb') as f:
|
||
f.write(content)
|
||
except:
|
||
pass
|
||
elif item.get("type") == "dir":
|
||
sub_dir = item.get("name")
|
||
self._download_dir_raw(author, plugin_name, sub_dir, install_dir / sub_dir)
|
||
time.sleep(0.3)
|
||
|
||
Log.info("pkg-manager", f"已安装: {author}/{plugin_name}")
|
||
return True
|
||
except Exception as e:
|
||
Log.error("pkg-manager", f"安装失败 {plugin_name}: {e}")
|
||
if install_dir.exists():
|
||
shutil.rmtree(install_dir)
|
||
return False
|
||
|
||
def _download_dir_raw(self, author: str, plugin: str, sub_dir: str, local_dir: Path):
|
||
"""使用 raw URL 递归下载子目录"""
|
||
import time
|
||
try:
|
||
api_url = f"{GITEE_API_BASE}/store/{author}/{plugin}/{sub_dir}"
|
||
with _gitee_request(api_url, timeout=15) as resp:
|
||
items = json.loads(resp.read().decode("utf-8"))
|
||
|
||
local_dir.mkdir(parents=True, exist_ok=True)
|
||
for item in items:
|
||
if item.get("type") == "file":
|
||
filename = item.get("name")
|
||
raw_url = f"{GITEE_RAW_BASE}/store/{author}/{plugin}/{sub_dir}/{filename}"
|
||
try:
|
||
with _gitee_request(raw_url, timeout=15) as resp:
|
||
content = resp.read()
|
||
with open(local_dir / filename, 'wb') as f:
|
||
f.write(content)
|
||
except:
|
||
pass
|
||
elif item.get("type") == "dir":
|
||
self._download_dir_raw(author, plugin, f"{sub_dir}/{item.get('name')}", local_dir / item.get("name"))
|
||
except:
|
||
pass
|
||
|
||
# ==================== 辅助方法 ====================
|
||
|
||
def _scan_all_plugins(self) -> list:
|
||
"""扫描本地已安装插件"""
|
||
plugins = []
|
||
if not self.store_dir.exists():
|
||
return plugins
|
||
|
||
for author_dir in self.store_dir.iterdir():
|
||
if author_dir.is_dir() and author_dir.name.startswith("@{"):
|
||
for plugin_dir in author_dir.iterdir():
|
||
if plugin_dir.is_dir() and (plugin_dir / "main.py").exists():
|
||
manifest_path = plugin_dir / "manifest.json"
|
||
if manifest_path.exists():
|
||
with open(manifest_path, 'r', encoding='utf-8') as f:
|
||
manifest = json.load(f)
|
||
plugins.append({
|
||
"name": plugin_dir.name,
|
||
"full_name": f"{author_dir.name}/{plugin_dir.name}",
|
||
"author": author_dir.name,
|
||
"metadata": manifest.get("metadata", {}),
|
||
"dependencies": manifest.get("dependencies", []),
|
||
"has_config": (plugin_dir / "config.json").exists(),
|
||
"is_installed": True
|
||
})
|
||
return plugins
|
||
|
||
def _is_plugin_installed(self, plugin_name: str, author: str) -> bool:
|
||
"""检查插件是否已安装"""
|
||
plugin_dir = self.store_dir / author / plugin_name
|
||
return (plugin_dir / "main.py").exists()
|
||
|
||
def _find_plugin_dir(self, plugin_name: str) -> Path | None:
|
||
"""查找插件目录"""
|
||
if not self.store_dir.exists():
|
||
return None
|
||
for author_dir in self.store_dir.iterdir():
|
||
if author_dir.is_dir():
|
||
plugin_dir = author_dir / plugin_name
|
||
if plugin_dir.exists() and (plugin_dir / "main.py").exists():
|
||
return plugin_dir
|
||
return None
|
||
|
||
def _load_config_schema(self, plugin_name: str) -> dict:
|
||
"""加载插件 config.json schema"""
|
||
plugin_dir = self._find_plugin_dir(plugin_name)
|
||
if not plugin_dir:
|
||
return {}
|
||
schema_path = plugin_dir / "config.json"
|
||
if not schema_path.exists():
|
||
return {}
|
||
with open(schema_path, 'r', encoding='utf-8') as f:
|
||
return json.load(f)
|
||
|
||
def _load_plugin_config(self, plugin_name: str) -> dict:
|
||
"""加载插件当前配置"""
|
||
schema = self._load_config_schema(plugin_name)
|
||
defaults = {}
|
||
for key, field_def in schema.items():
|
||
defaults[key] = field_def.get("default")
|
||
if self.storage:
|
||
storage_instance = self.storage.get_storage("pkg-manager")
|
||
user_config = storage_instance.get(f"plugin_config.{plugin_name}", {})
|
||
defaults.update(user_config)
|
||
return defaults
|
||
|
||
def _save_plugin_config(self, plugin_name: str, config: dict):
|
||
"""保存插件配置"""
|
||
if self.storage:
|
||
storage_instance = self.storage.get_storage("pkg-manager")
|
||
storage_instance.set(f"plugin_config.{plugin_name}", config)
|
||
|
||
def _get_plugin_detailed_info(self, plugin_name: str) -> dict:
|
||
"""获取插件的依赖、事件、页面信息"""
|
||
dependencies = []
|
||
events = [] # 事件 = 功能描述
|
||
plugin_dir = self._find_plugin_dir(plugin_name)
|
||
|
||
if plugin_dir:
|
||
manifest_path = plugin_dir / "manifest.json"
|
||
if manifest_path.exists():
|
||
with open(manifest_path, 'r', encoding='utf-8') as f:
|
||
manifest = json.load(f)
|
||
dependencies = manifest.get("dependencies", [])
|
||
# 从 manifest 的 metadata.description 或 type 中提取功能
|
||
metadata = manifest.get("metadata", {})
|
||
plugin_type = metadata.get("type", "")
|
||
if plugin_type:
|
||
events.append(f"类型: {plugin_type}")
|
||
# 从 manifest config 推断功能
|
||
config = manifest.get("config", {})
|
||
if config.get("enabled"):
|
||
events.append("已启用")
|
||
|
||
# 只返回该插件自己注册的页面（通过插件名匹配）
|
||
pages = []
|
||
if self.webui and hasattr(self.webui, 'server') and self.webui.server:
|
||
for path, provider in self.webui.server.pages.items():
|
||
# 检查 provider 是否属于该插件
|
||
provider_name = getattr(provider, '__self__', None)
|
||
if provider_name and isinstance(provider_name, PkgManagerPlugin):
|
||
continue # 跳过自己的页面
|
||
# 通过路径前缀判断（dashboard 注册 /dashboard）
|
||
if path == f'/{plugin_name}' or path.startswith(f'/{plugin_name}/'):
|
||
pages.append({"path": path})
|
||
# 特殊处理：首页
|
||
if plugin_name == 'webui' and path == '/':
|
||
pages.append({"path": path})
|
||
|
||
return {
|
||
"name": plugin_name,
|
||
"dependencies": dependencies,
|
||
"config_fields": list(self._load_config_schema(plugin_name).keys()),
|
||
"pages": pages,
|
||
"events": events
|
||
}
|
||
|
||
|
||
register_plugin_type("PkgManagerPlugin", PkgManagerPlugin)
|
||
|
||
|
||
def New():
|
||
return PkgManagerPlugin()
|