codfrm/MoviePilot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat：插件仪表板API

Browse Source
This commit is contained in:
jxxghp
2024-05-08 20:58:28 +08:00
parent cce2e13e21
commit 5c9039e6d0
5 changed files with 133 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
app/api/endpoints/plugin.py
View File

@ -125,6 +125,22 @@ def plugin_page(plugin_id: str, _: schemas.TokenPayload = Depends(verify_token))
return PluginManager().get_plugin_page(plugin_id)
@router.get("/dashboards", summary="获取有仪表板的插件清单")
def dashboard_plugins(_: schemas.TokenPayload = Depends(verify_token)) -> List[dict]:
"""
获取所有插件仪表板
"""
return PluginManager().get_dashboard_plugins()
@router.get("/dashboard/{plugin_id}", summary="获取插件仪表板配置")
def plugin_dashboard(plugin_id: str, _: schemas.TokenPayload = Depends(verify_token)) -> schemas.PluginDashboard:
"""
根据插件ID获取插件仪表板
"""
return PluginManager().get_plugin_dashboard(plugin_id)
@router.get("/reset/{plugin_id}", summary="重置插件配置", response_model=schemas.Response)
def reset_plugin(plugin_id: str, _: schemas.TokenPayload = Depends(verify_token)) -> Any:
"""

66
app/core/plugin.py
View File

@ -217,10 +217,11 @@ class PluginManager(metaclass=Singleton):
获取插件表单
:param pid: 插件ID
"""
if not self._running_plugins.get(pid):
plugin = self._running_plugins.get(pid)
if not plugin:
return [], {}
if hasattr(self._running_plugins[pid], "get_form"):
return self._running_plugins[pid].get_form() or ([], {})
if hasattr(plugin, "get_form"):
return plugin.get_form() or ([], {})
return [], {}
def get_plugin_page(self, pid: str) -> List[dict]:
@ -228,12 +229,34 @@ class PluginManager(metaclass=Singleton):
获取插件页面
:param pid: 插件ID
"""
if not self._running_plugins.get(pid):
plugin = self._running_plugins.get(pid)
if not plugin:
return []
if hasattr(self._running_plugins[pid], "get_page"):
return self._running_plugins[pid].get_page() or []
if hasattr(plugin, "get_page"):
return plugin.get_page() or []
return []
def get_plugin_dashboard(self, pid: str) -> Optional[schemas.PluginDashboard]:
"""
获取插件仪表盘
:param pid: 插件ID
"""
plugin = self._running_plugins.get(pid)
if not plugin:
return None
if hasattr(plugin, "get_dashboard"):
dashboard: Tuple = plugin.get_dashboard()
if dashboard:
cols, attrs, elements = dashboard
return schemas.PluginDashboard(
id=pid,
name=plugin.plugin_name,
cols=cols or {},
elements=elements,
attrs=attrs or {}
)
return None
def get_plugin_commands(self) -> List[Dict[str, Any]]:
"""
获取插件命令
@ -301,17 +324,35 @@ class PluginManager(metaclass=Singleton):
logger.error(f"获取插件 {pid} 服务出错：{str(e)}")
return ret_services
def get_dashboard_plugins(self) -> List[dict]:
"""
获取有仪表盘的插件列表
"""
dashboards = []
for pid, plugin in self._running_plugins.items():
if hasattr(plugin, "get_dashboard") \
and ObjectUtils.check_method(plugin.get_dashboard):
try:
dashboards.append({
"id": pid,
"name": plugin.plugin_name
})
except Exception as e:
logger.error(f"获取有仪表盘的插件出错：{str(e)}")
return dashboards
def get_plugin_attr(self, pid: str, attr: str) -> Any:
"""
获取插件属性
:param pid: 插件ID
:param attr: 属性名
"""
if not self._running_plugins.get(pid):
plugin = self._running_plugins.get(pid)
if not plugin:
return None
if not hasattr(self._running_plugins[pid], attr):
if not hasattr(plugin, attr):
return None
return getattr(self._running_plugins[pid], attr)
return getattr(plugin, attr)
def run_plugin_method(self, pid: str, method: str, *args, **kwargs) -> Any:
"""
@ -321,11 +362,12 @@ class PluginManager(metaclass=Singleton):
:param args: 参数
:param kwargs: 关键字参数
"""
if not self._running_plugins.get(pid):
plugin = self._running_plugins.get(pid)
if not plugin:
return None
if not hasattr(self._running_plugins[pid], method):
if not hasattr(plugin, method):
return None
return getattr(self._running_plugins[pid], method)(*args, **kwargs)
return getattr(plugin, method)(*args, **kwargs)
def get_plugin_ids(self) -> List[str]:
"""

49
app/plugins/__init__.py
View File

@ -1,6 +1,6 @@
from abc import ABCMeta, abstractmethod
from pathlib import Path
from typing import Any, List, Dict, Tuple
from typing import Any, List, Dict, Tuple, Optional
from app.chain import ChainBase
from app.core.config import settings
@ -55,6 +55,13 @@ class _PluginBase(metaclass=ABCMeta):
"""
pass
@abstractmethod
def get_state(self) -> bool:
"""
获取插件运行状态
"""
pass
@staticmethod
@abstractmethod
def get_command() -> List[Dict[str, Any]]:
@ -84,19 +91,6 @@ class _PluginBase(metaclass=ABCMeta):
"""
pass
def get_service(self) -> List[Dict[str, Any]]:
"""
注册插件公共服务
[{
"id": "服务ID",
"name": "服务名称",
"trigger": "触发器：cron/interval/date/CronTrigger.from_crontab()",
"func": self.xxx,
"kwargs": {} # 定时器参数
}]
"""
pass
@abstractmethod
def get_form(self) -> Tuple[List[dict], Dict[str, Any]]:
"""
@ -113,10 +107,31 @@ class _PluginBase(metaclass=ABCMeta):
"""
pass
@abstractmethod
def get_state(self) -> bool:
def get_service(self) -> List[Dict[str, Any]]:
"""
获取插件运行状态
注册插件公共服务
[{
"id": "服务ID",
"name": "服务名称",
"trigger": "触发器：cron/interval/date/CronTrigger.from_crontab()",
"func": self.xxx,
"kwargs": {} # 定时器参数
}]
"""
pass
def get_dashboard(self) -> Optional[Tuple[Dict[str, Any], Dict[str, Any], List[dict]]]:
"""
获取插件仪表盘页面，需要返回：1、仪表板col配置字典；2、全局配置（自动刷新等）；3、仪表板页面元素配置json（含数据）
1、col配置参考：
{
"cols": 12, "md": 6
}
2、全局配置参考：
{
"refresh": 10 // 自动刷新时间，单位秒
}
3、页面配置使用Vuetify组件拼装，参考：https://vuetifyjs.com/
"""
pass

17
app/schemas/plugin.py
View File

@ -1,4 +1,4 @@
from typing import Optional
from typing import Optional, List
from pydantic import BaseModel
@ -46,3 +46,18 @@ class Plugin(BaseModel):
history: Optional[dict] = {}
# 添加时间，值越小表示越靠后发布
add_time: Optional[int] = 0
class PluginDashboard(Plugin):
"""
插件仪表盘
"""
id: Optional[str] = None
# 名称
name: Optional[str] = None
# 全局配置
attrs: Optional[dict] = {}
# col列数
cols: Optional[dict] = {}
# 页面元素
elements: Optional[List[dict]] = []

16
app/utils/object.py
View File

@ -35,7 +35,21 @@ class ObjectUtils:
"""
检查函数是否已实现
"""
return func.__code__.co_code not in [b'd\x01S\x00', b'\x97\x00d\x00S\x00']
source = inspect.getsource(func)
in_comment = False
for line in source.split('\n'):
line = line.strip()
if not line:
continue
if line.startswith('"""') or line.startswith("'''"):
in_comment = not in_comment
continue
if not in_comment and not (line.startswith('#')
or line == "pass"
or line.startswith('@')
or line.startswith('def ')):
return True
return False
@staticmethod
def check_signature(func: FunctionType, *args) -> bool: