插件直接采用程序的定时任务模块, 可显示在前端页面

app/api/endpoints/plugin.py

@@ -8,6 +8,7 @@ from app.core.security import verify_token
 from app.db.systemconfig_oper import SystemConfigOper
 from app.helper.plugin import PluginHelper
 from app.schemas.types import SystemConfigKey
+from app.scheduler import Scheduler
 
 router = APIRouter()
 
@@ -90,6 +91,8 @@ def install(plugin_id: str,
         SystemConfigOper().set(SystemConfigKey.UserInstalledPlugins, install_plugins)
     # 重载插件管理器
     PluginManager().init_config()
+    # 注册插件服务
+    Scheduler().update_plugin_job(plugin_id)
     return schemas.Response(success=True)
 
 
@@ -123,6 +126,8 @@ def reset_plugin(plugin_id: str, _: schemas.TokenPayload = Depends(verify_token)
     PluginManager().delete_plugin_config(plugin_id)
     # 重新生效插件
     PluginManager().reload_plugin(plugin_id, {})
+    # 注册插件服务
+    Scheduler().update_plugin_job(plugin_id)
     return schemas.Response(success=True)
 
 
@@ -144,6 +149,8 @@ def set_plugin_config(plugin_id: str, conf: dict,
     PluginManager().save_plugin_config(plugin_id, conf)
     # 重新生效插件
     PluginManager().reload_plugin(plugin_id, conf)
+    # 注册插件服务
+    Scheduler().update_plugin_job(plugin_id)
     return schemas.Response(success=True)
 
 
@@ -163,6 +170,8 @@ def uninstall_plugin(plugin_id: str,
     SystemConfigOper().set(SystemConfigKey.UserInstalledPlugins, install_plugins)
     # 重载插件管理器
     PluginManager().init_config()
+    # 移除插件服务
+    Scheduler().remove_plugin_job(plugin_id)
     return schemas.Response(success=True)
 
 

app/core/plugin.py

@@ -262,6 +262,12 @@ class PluginManager(metaclass=Singleton):
         """
         return list(self._plugins.keys())
 
+    def get_running_plugin_ids(self) -> List[str]:
+        """
+        获取所有运行态插件ID
+        """
+        return list(self._running_plugins.keys())
+
     def get_online_plugins(self) -> List[dict]:
         """
         获取所有在线插件信息

app/scheduler.py

@@ -43,6 +43,8 @@ class Scheduler(metaclass=Singleton):
                                      })
     # 退出事件
     _event = threading.Event()
+    # 锁
+    _lock = threading.Lock()
 
     def __init__(self):
 
@@ -232,25 +234,8 @@ class Scheduler(metaclass=Singleton):
         )
 
         # 注册插件公共服务
-        plugin_services = PluginManager().get_plugin_services()
+        for pid in PluginManager().get_running_plugin_ids():
-        for service in plugin_services:
+            self.update_plugin_job(pid)
-            try:
-                self._jobs[service["id"]] = {
-                    "func": service["func"],
-                    "running": False,
-                }
-                self._scheduler.add_job(
-                    self.start,
-                    service["trigger"],
-                    id=service["id"],
-                    name=service["name"],
-                    **service["kwargs"],
-                    kwargs={
-                        'job_id': service["id"]
-                    }
-                )
-            except Exception as e:
-                logger.error(f"注册插件服务失败：{str(e)} - {service}")
 
         # 打印服务
         logger.debug(self._scheduler.print_jobs())
@@ -263,51 +248,127 @@ class Scheduler(metaclass=Singleton):
         启动定时服务
         """
         # 处理job_id格式
-        job = self._jobs.get(job_id)
+        with self._lock:
-        if not job:
+            job = self._jobs.get(job_id)
-            return
+            if not job:
-        if job.get("running"):
+                return
-            logger.warning(f"定时任务 {job_id} 正在运行 ...")
+            if job.get("running"):
-            return
+                logger.warning(f"定时任务 {job_id} 正在运行 ...")
-        self._jobs[job_id]["running"] = True
+                return
+            self._jobs[job_id]["running"] = True
         try:
             if not kwargs:
                 kwargs = job.get("kwargs") or {}
             job["func"](*args, **kwargs)
         except Exception as e:
             logger.error(f"定时任务 {job_id} 执行失败：{str(e)}")
-        self._jobs[job_id]["running"] = False
+        # 如果在job["func"]()运行时, 编辑配置导致任务被移除时, 该id已经不存在, 忽略错误
+        with self._lock:
+            try:
+                self._jobs[job_id]["running"] = False
+            except:
+                pass
+            # 如果是单次任务, 应立即移除缓存
+            if not self._scheduler.get_job(job_id):
+                self._jobs.pop(job_id, None)
+
+    def update_plugin_job(self, pid: str):
+        """
+        更新插件定时服务
+        """
+        # 移除该插件的全部服务
+        self.remove_plugin_job(pid)
+        # 获取插件服务列表
+        with self._lock:
+            try:
+                plugin_services = PluginManager().run_plugin_method(pid, "get_service") or []
+            except:
+                return
+            # 开始注册插件服务
+            for service in plugin_services:
+                try:
+                    sid = f"{pid}_{service['id']}"
+                    self._jobs[sid] = {
+                        "func": service["func"],
+                        "name": service["name"],
+                        "pid": pid,
+                        "running": False,
+                    }
+                    self._scheduler.add_job(
+                        self.start,
+                        service["trigger"],
+                        id=sid,
+                        name=service["name"],
+                        **service["kwargs"],
+                        kwargs={
+                            'job_id': sid
+                        }
+                    )
+                    logger.info(f"注册插件服务({pid})：{service['name']}")
+                except Exception as e:
+                    logger.error(f"注册插件服务失败：{str(e)} - {service}")
+
+    def remove_plugin_job(self, pid: str):
+        """
+        移除插件定时服务
+        """
+        with self._lock:
+            for job_id, service in self._jobs.copy().items():
+                try:
+                    if service.get("pid") == pid:
+                        self._jobs.pop(job_id, None)
+                        try:
+                            self._scheduler.remove_job(job_id)
+                        except:
+                            pass
+                        logger.info(f"移除插件服务({pid})：{service.get('name')}")
+                except Exception as e:
+                    logger.error(f"移除插件服务失败：{str(e)} - {job_id}: {service}")
 
     def list(self) -> List[schemas.ScheduleInfo]:
         """
         当前所有任务
         """
-        # 返回计时任务
+        with self._lock:
-        schedulers = []
+            # 返回计时任务
-        # 去重
+            schedulers = []
-        added = []
+            # 去重
-        jobs = self._scheduler.get_jobs()
+            added = []
-        # 按照下次运行时间排序
+            jobs = self._scheduler.get_jobs()
-        jobs.sort(key=lambda x: x.next_run_time)
+            # 按照下次运行时间排序
-        for job in jobs:
+            jobs.sort(key=lambda x: x.next_run_time)
-            if job.name not in added:
+            # 将正在运行的任务提取出来 (保障一次性任务正常显示)
-                added.append(job.name)
+            for job_id, value in self._jobs.items():
-            else:
+                name = value.get("name")
-                continue
+                if value.get("running") and name:
-            job_id = job.id.split("|")[0]
+                    if name not in added:
-            if not self._jobs.get(job_id):
+                        added.append(name)
-                continue
+                    else:
-            # 任务状态
+                        continue
-            status = "正在运行" if self._jobs[job_id].get("running") else "等待"
+                    schedulers.append(schemas.ScheduleInfo(
-            # 下次运行时间
+                        id=job_id,
-            next_run = TimerUtils.time_difference(job.next_run_time)
+                        name=name,
-            schedulers.append(schemas.ScheduleInfo(
+                        status="正在运行",
-                id=job_id,
+                        next_run="~"
-                name=job.name,
+                    ))
-                status=status,
+            # 获取其他待执行任务
-                next_run=next_run
+            for job in jobs:
-            ))
+                if job.name not in added:
-        return schedulers
+                    added.append(job.name)
+                else:
+                    continue
+                job_id = job.id.split("|")[0]
+                if not self._jobs.get(job_id):
+                    continue
+                # 下次运行时间
+                next_run = TimerUtils.time_difference(job.next_run_time)
+                schedulers.append(schemas.ScheduleInfo(
+                    id=job_id,
+                    name=job.name,
+                    status="等待",
+                    next_run=next_run
+                ))
+            return schedulers
 
     def stop(self):
         """