mirror of
https://github.com/jxxghp/MoviePilot-Plugins.git
synced 2026-03-27 10:05:57 +00:00
235 lines
7.9 KiB
Python
235 lines
7.9 KiB
Python
import shutil
|
||
from pathlib import Path
|
||
from typing import Any, Dict, List, Tuple
|
||
|
||
from app.core.config import settings
|
||
from app.core.plugin import PluginManager
|
||
from app.db.systemconfig_oper import SystemConfigOper
|
||
from app.log import logger
|
||
from app.plugins import _PluginBase
|
||
from app.schemas.types import SystemConfigKey
|
||
|
||
|
||
class CleanInvalidPlugin(_PluginBase):
|
||
# 插件名称
|
||
plugin_name = "清理无效插件"
|
||
# 插件描述
|
||
plugin_desc = "删除数据库中无法安装的插件记录，避免反复重装。"
|
||
# 插件图标
|
||
plugin_icon = "delete.jpg"
|
||
# 插件版本
|
||
plugin_version = "1.0"
|
||
# 插件作者
|
||
plugin_author = "cddjr"
|
||
# 作者主页
|
||
author_url = "https://github.com/cddjr"
|
||
# 插件配置项ID前缀
|
||
plugin_config_prefix = "cleaninvalidplugin_"
|
||
# 加载顺序
|
||
plugin_order = 999
|
||
# 可使用的用户级别
|
||
auth_level = 1
|
||
|
||
# 需要清理的插件
|
||
_invalid_plugin_ids = []
|
||
|
||
def init_plugin(self, config: dict = None):
|
||
"""
|
||
生效配置信息
|
||
|
||
:param config: 配置信息字典
|
||
"""
|
||
try:
|
||
if not config:
|
||
return
|
||
self._invalid_plugin_ids = config.get("invalid_plugin_ids") or []
|
||
if not self._invalid_plugin_ids:
|
||
return
|
||
|
||
config_oper = SystemConfigOper()
|
||
plugin_manager = PluginManager()
|
||
|
||
valid_plugins = set(plugin_manager.get_plugin_ids() or [])
|
||
all_plugins: List[str] = (
|
||
config_oper.get(SystemConfigKey.UserInstalledPlugins) or []
|
||
)
|
||
|
||
all_plugins_modified = []
|
||
for plugin_id in all_plugins:
|
||
if plugin_id not in self._invalid_plugin_ids:
|
||
all_plugins_modified.append(plugin_id)
|
||
continue
|
||
try:
|
||
# 再一次确保是无效
|
||
if plugin_id in valid_plugins:
|
||
all_plugins_modified.append(plugin_id)
|
||
logger.warn(f"{plugin_id} 是有效插件")
|
||
continue
|
||
logger.info(f"正在清理无效插件 {plugin_id}")
|
||
plugin_dir = (
|
||
Path(settings.ROOT_PATH) / "app" / "plugins" / plugin_id.lower()
|
||
)
|
||
if plugin_dir.exists():
|
||
shutil.rmtree(plugin_dir, ignore_errors=True)
|
||
else:
|
||
logger.warn(f"插件目录 {plugin_dir} 不存在")
|
||
except Exception as e:
|
||
logger.warn(
|
||
f"清理无效插件 {plugin_id} 产生异常: {e}", exc_info=True
|
||
)
|
||
|
||
# 更新安装记录
|
||
config_oper.set(SystemConfigKey.UserInstalledPlugins, all_plugins_modified)
|
||
self._invalid_plugin_ids = []
|
||
self.update_config(
|
||
{
|
||
"invalid_plugin_ids": [],
|
||
}
|
||
)
|
||
except Exception as e:
|
||
logger.error(f"异常: {e}", exc_info=True)
|
||
|
||
def get_state(self) -> bool:
|
||
"""
|
||
获取插件运行状态
|
||
"""
|
||
return False
|
||
|
||
@staticmethod
|
||
def get_command() -> List[Dict[str, Any]]:
|
||
"""
|
||
注册插件远程命令
|
||
|
||
[{
|
||
"cmd": "/xx",
|
||
"event": EventType.xx,
|
||
"desc": "名称",
|
||
"category": "分类，需要注册到Wechat时必须有分类",
|
||
"data": {}
|
||
}]
|
||
"""
|
||
pass
|
||
|
||
def get_api(self) -> List[Dict[str, Any]]:
|
||
"""
|
||
注册插件API
|
||
|
||
[{
|
||
"path": "/xx",
|
||
"endpoint": self.xxx,
|
||
"methods": ["GET", "POST"],
|
||
"auth: "apikey", # 鉴权类型：apikey/bear
|
||
"summary": "API名称",
|
||
"description": "API说明"
|
||
}]
|
||
"""
|
||
pass
|
||
|
||
def get_form(self) -> Tuple[List[dict], Dict[str, Any]]:
|
||
"""
|
||
拼装插件配置页面，插件配置页面使用Vuetify组件拼装，参考：https://vuetifyjs.com/
|
||
|
||
:return: 1、页面配置（vuetify模式）或 None（vue模式）；2、默认数据结构
|
||
"""
|
||
invalid_items = self.get_invalid_plugins()
|
||
return [
|
||
{
|
||
"component": "VForm",
|
||
"content": [
|
||
{
|
||
"component": "VRow",
|
||
"content": [
|
||
{
|
||
"component": "VCol",
|
||
"props": {"cols": 12, "md": 6},
|
||
"content": [
|
||
{
|
||
"component": "VSelect",
|
||
"props": {
|
||
"multiple": True,
|
||
"chips": True,
|
||
"model": "invalid_plugin_ids",
|
||
"label": "插件ID",
|
||
"items": invalid_items,
|
||
},
|
||
}
|
||
],
|
||
},
|
||
],
|
||
},
|
||
{
|
||
"component": "VRow",
|
||
"content": [
|
||
{
|
||
"component": "VCol",
|
||
"props": {
|
||
"cols": 12,
|
||
},
|
||
"content": [
|
||
{
|
||
"component": "VAlert",
|
||
"props": {
|
||
"type": "info",
|
||
"variant": "tonal",
|
||
"text": (
|
||
f"当前有{len(invalid_items)}个插件无法安装，选择需要清理的插件，点击【保存】"
|
||
if invalid_items
|
||
else "所有插件均已成功安装，无需清理"
|
||
),
|
||
},
|
||
}
|
||
],
|
||
}
|
||
],
|
||
},
|
||
],
|
||
}
|
||
], {
|
||
"invalid_plugin_ids": self._invalid_plugin_ids,
|
||
}
|
||
|
||
def get_page(self) -> List[dict]:
|
||
"""
|
||
拼装插件详情页面，需要返回页面配置，同时附带数据
|
||
插件详情页面使用Vuetify组件拼装，参考：https://vuetifyjs.com/
|
||
|
||
:return: 页面配置（vuetify模式）或 None（vue模式）
|
||
"""
|
||
pass
|
||
|
||
def stop_service(self):
|
||
"""
|
||
停止插件
|
||
"""
|
||
pass
|
||
|
||
@staticmethod
|
||
def get_invalid_plugins():
|
||
"""
|
||
获取本地无效插件
|
||
"""
|
||
try:
|
||
# 已安装插件
|
||
config_oper = SystemConfigOper()
|
||
plugin_manager = PluginManager()
|
||
|
||
# 所有安装的插件（包括安装后未能成功加载的插件）
|
||
all_plugins = set(
|
||
config_oper.get(SystemConfigKey.UserInstalledPlugins) or []
|
||
)
|
||
|
||
# 获取有效插件(成功安装并能正常加载)
|
||
valid_plugins = set(plugin_manager.get_plugin_ids() or [])
|
||
|
||
# 过滤无效的插件
|
||
invalid_plugins = all_plugins - valid_plugins
|
||
|
||
# 构建 VSelect 数据
|
||
return [
|
||
{"title": f"{plugin_id}", "value": plugin_id}
|
||
for plugin_id in invalid_plugins
|
||
]
|
||
except Exception as e:
|
||
logger.error(f"异常: {e}", exc_info=True)
|
||
return []
|