qinglong
fca2aa5a05
PluginIndex 服务: - 从远程 JSON 索引获取可用插件列表 - 支持 zip 包和单文件 .py 两种格式安装 - 自动解压/移动到 plugins/ 目录 - 默认索引地址: GitHub Pages install 内置命令: - install --list → 列出远程可用插件 - install <name> → 下载并安装 - 自动缓存索引，避免重复请求 Co-Authored-By: Claude <noreply@anthropic.com>
109 lines
4.4 KiB
Python
109 lines
4.4 KiB
Python
#!/usr/bin/env python3
|
|
"""插件索引仓库 — 在线 JSON 索引 + 一键安装"""
|
|
import logging, json, os, asyncio, tempfile, zipfile, shutil
|
|
from pathlib import Path
|
|
from typing import Optional, List, Dict
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
DEFAULT_INDEX_URL = "https://raw.githubusercontent.com/AskaEth/SenSu-plugins/main/index.json"
|
|
|
|
|
|
class PluginIndex:
|
|
def __init__(self, index_url: str = DEFAULT_INDEX_URL):
|
|
self.index_url = index_url
|
|
self._cache: Optional[List[Dict]] = None
|
|
|
|
async def fetch_index(self, force: bool = False) -> List[Dict]:
|
|
"""获取远程插件索引"""
|
|
if self._cache is not None and not force:
|
|
return self._cache
|
|
try:
|
|
import aiohttp
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(self.index_url, timeout=aiohttp.ClientTimeout(total=15)) as resp:
|
|
if resp.status == 200:
|
|
data = await resp.json()
|
|
self._cache = data if isinstance(data, list) else data.get("plugins", [])
|
|
logger.info(f"插件索引已加载: {len(self._cache)} 个可用插件")
|
|
return self._cache
|
|
else:
|
|
logger.warning(f"插件索引请求失败: HTTP {resp.status}")
|
|
return []
|
|
except Exception as e:
|
|
logger.warning(f"无法获取插件索引 ({self.index_url}): {e}")
|
|
return []
|
|
|
|
def search(self, name: str) -> Optional[Dict]:
|
|
"""在缓存中搜索插件"""
|
|
if not self._cache:
|
|
return None
|
|
for p in self._cache:
|
|
if p.get("name") == name:
|
|
return p
|
|
return None
|
|
|
|
def list_plugins(self) -> List[str]:
|
|
if not self._cache:
|
|
return []
|
|
return [f"{p.get('name','?')} v{p.get('version','?')} — {p.get('description','?')[:60]}"
|
|
for p in self._cache]
|
|
|
|
async def install(self, name: str, target_dir: str = "plugins") -> bool:
|
|
"""下载并安装指定插件到 plugins/ 目录"""
|
|
plugin = self.search(name)
|
|
if not plugin:
|
|
# Try to refresh cache
|
|
await self.fetch_index(force=True)
|
|
plugin = self.search(name)
|
|
if not plugin:
|
|
logger.error(f"插件未在索引中找到: {name}")
|
|
return False
|
|
|
|
download_url = plugin.get("download_url") or plugin.get("url")
|
|
if not download_url:
|
|
logger.error(f"插件 {name} 缺少下载地址")
|
|
return False
|
|
|
|
target = Path(target_dir) / name
|
|
if target.exists():
|
|
logger.warning(f"插件目录已存在: {target}")
|
|
return False
|
|
|
|
try:
|
|
import aiohttp
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(download_url, timeout=aiohttp.ClientTimeout(total=120)) as resp:
|
|
if resp.status != 200:
|
|
logger.error(f"下载失败: HTTP {resp.status}")
|
|
return False
|
|
content = await resp.read()
|
|
|
|
# Handle zip archives
|
|
if download_url.endswith(".zip"):
|
|
with tempfile.NamedTemporaryFile(suffix=".zip", delete=False) as tmp:
|
|
tmp.write(content)
|
|
tmp.flush()
|
|
with zipfile.ZipFile(tmp.name, "r") as zf:
|
|
# Zip 内第一层目录名可能不同, 提取到临时位置再移动
|
|
extract_tmp = Path(tempfile.mkdtemp())
|
|
zf.extractall(extract_tmp)
|
|
# 如果 zip 内只有一个顶层目录, 直接用它
|
|
members = list(extract_tmp.iterdir())
|
|
if len(members) == 1 and members[0].is_dir():
|
|
shutil.move(str(members[0]), str(target))
|
|
else:
|
|
extract_tmp.rename(target)
|
|
os.unlink(tmp.name)
|
|
else:
|
|
# Assume single .py file plugin
|
|
target.mkdir(parents=True, exist_ok=True)
|
|
(target / "__init__.py").write_bytes(content)
|
|
|
|
logger.info(f"✅ 插件安装完成: {name} → {target}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"安装插件 {name} 失败: {e}")
|
|
return False
|