♻️ refactor(scheduler): 重构定时任务服务架构并增强用户体验 (#1967)

**架构重构**
- 拆分为 Service、Repository、Adapter 三层架构，提升模块化
- 统一 APScheduler Job ID 生成方式，优化 ScheduleTargeter 逻辑

**新增功能**
- 支持定时任务时区配置
- 新增"运行中"任务状态显示
- 为"所有群组"任务增加随机延迟，分散并发压力

**用户体验优化**
- 重构操作反馈消息，提供详细的成功提示卡片
- 优化任务查看命令的筛选逻辑
- 统一删除、暂停、恢复、执行、更新操作的响应格式

Co-authored-by: webjoin111 <455457521@qq.com>

zhenxun/services/scheduler/__init__.py

@@ -0,0 +1,12 @@
+"""
+定时调度服务模块
+
+提供一个统一的、持久化的定时任务管理器，供所有插件使用。
+"""
+
+from .lifecycle import _load_schedules_from_db
+from .service import scheduler_manager
+
+_ = _load_schedules_from_db
+
+__all__ = ["scheduler_manager"]

zhenxun/services/scheduler/adapter.py

@@ -0,0 +1,102 @@
+"""
+引擎适配层 (Adapter)
+
+封装所有对具体调度器引擎 (APScheduler) 的操作，
+使上层服务与调度器实现解耦。
+"""
+
+from nonebot_plugin_apscheduler import scheduler
+
+from zhenxun.models.schedule_info import ScheduleInfo
+from zhenxun.services.log import logger
+
+from .job import _execute_job
+
+JOB_PREFIX = "zhenxun_schedule_"
+
+
+class APSchedulerAdapter:
+    """封装对 APScheduler 的操作"""
+
+    @staticmethod
+    def _get_job_id(schedule_id: int) -> str:
+        """生成 APScheduler 的 Job ID"""
+        return f"{JOB_PREFIX}{schedule_id}"
+
+    @staticmethod
+    def add_or_reschedule_job(schedule: ScheduleInfo):
+        """根据 ScheduleInfo 添加或重新调度一个 APScheduler 任务"""
+        job_id = APSchedulerAdapter._get_job_id(schedule.id)
+
+        if not isinstance(schedule.trigger_config, dict):
+            logger.error(
+                f"任务 {schedule.id} 的 trigger_config 不是字典类型: "
+                f"{type(schedule.trigger_config)}"
+            )
+            return
+
+        job = scheduler.get_job(job_id)
+        if job:
+            scheduler.reschedule_job(
+                job_id, trigger=schedule.trigger_type, **schedule.trigger_config
+            )
+            logger.debug(f"已更新APScheduler任务: {job_id}")
+        else:
+            scheduler.add_job(
+                _execute_job,
+                trigger=schedule.trigger_type,
+                id=job_id,
+                misfire_grace_time=300,
+                args=[schedule.id],
+                **schedule.trigger_config,
+            )
+            logger.debug(f"已添加新的APScheduler任务: {job_id}")
+
+    @staticmethod
+    def remove_job(schedule_id: int):
+        """移除一个 APScheduler 任务"""
+        job_id = APSchedulerAdapter._get_job_id(schedule_id)
+        try:
+            scheduler.remove_job(job_id)
+            logger.debug(f"已从APScheduler中移除任务: {job_id}")
+        except Exception:
+            pass
+
+    @staticmethod
+    def pause_job(schedule_id: int):
+        """暂停一个 APScheduler 任务"""
+        job_id = APSchedulerAdapter._get_job_id(schedule_id)
+        try:
+            scheduler.pause_job(job_id)
+        except Exception:
+            pass
+
+    @staticmethod
+    def resume_job(schedule_id: int):
+        """恢复一个 APScheduler 任务"""
+        job_id = APSchedulerAdapter._get_job_id(schedule_id)
+        try:
+            scheduler.resume_job(job_id)
+        except Exception:
+            import asyncio
+
+            from .repository import ScheduleRepository
+
+            async def _re_add_job():
+                schedule = await ScheduleRepository.get_by_id(schedule_id)
+                if schedule:
+                    APSchedulerAdapter.add_or_reschedule_job(schedule)
+
+            asyncio.create_task(_re_add_job())  # noqa: RUF006
+
+    @staticmethod
+    def get_job_status(schedule_id: int) -> dict:
+        """获取 APScheduler Job 的状态"""
+        job_id = APSchedulerAdapter._get_job_id(schedule_id)
+        job = scheduler.get_job(job_id)
+        return {
+            "next_run_time": job.next_run_time.strftime("%Y-%m-%d %H:%M:%S")
+            if job and job.next_run_time
+            else "N/A",
+            "is_paused_in_scheduler": not bool(job.next_run_time) if job else "N/A",
+        }

zhenxun/services/scheduler/job.py

@@ -0,0 +1,192 @@
+"""
+定时任务的执行逻辑
+
+包含被 APScheduler 实际调度的函数，以及处理不同目标（单个、所有群组）的执行策略。
+"""
+
+import asyncio
+import copy
+import inspect
+import random
+
+import nonebot
+
+from zhenxun.configs.config import Config
+from zhenxun.models.schedule_info import ScheduleInfo
+from zhenxun.services.log import logger
+from zhenxun.utils.common_utils import CommonUtils
+from zhenxun.utils.decorator.retry import Retry
+from zhenxun.utils.platform import PlatformUtils
+
+SCHEDULE_CONCURRENCY_KEY = "all_groups_concurrency_limit"
+
+
+async def _execute_job(schedule_id: int):
+    """
+    APScheduler 调度的入口函数。
+    根据 schedule_id 处理特定任务、所有群组任务或全局任务。
+    """
+    from .repository import ScheduleRepository
+    from .service import scheduler_manager
+
+    scheduler_manager._running_tasks.add(schedule_id)
+    try:
+        schedule = await ScheduleRepository.get_by_id(schedule_id)
+        if not schedule or not schedule.is_enabled:
+            logger.warning(f"定时任务 {schedule_id} 不存在或已禁用，跳过执行。")
+            return
+
+        plugin_name = schedule.plugin_name
+
+        task_meta = scheduler_manager._registered_tasks.get(plugin_name)
+        if not task_meta:
+            logger.error(
+                f"无法执行定时任务：插件 '{plugin_name}' 未注册或已卸载。将禁用该任务。"
+            )
+            schedule.is_enabled = False
+            await ScheduleRepository.save(schedule, update_fields=["is_enabled"])
+            from .adapter import APSchedulerAdapter
+
+            APSchedulerAdapter.remove_job(schedule.id)
+            return
+
+        try:
+            if schedule.bot_id:
+                bot = nonebot.get_bot(schedule.bot_id)
+            else:
+                bot = nonebot.get_bot()
+                logger.debug(
+                    f"任务 {schedule_id} 未关联特定Bot，使用默认Bot {bot.self_id}"
+                )
+        except KeyError:
+            logger.warning(
+                f"定时任务 {schedule_id} 需要的 Bot {schedule.bot_id} "
+                f"不在线，本次执行跳过。"
+            )
+            return
+        except ValueError:
+            logger.warning(f"当前没有Bot在线，定时任务 {schedule_id} 跳过。")
+            return
+
+        if schedule.group_id == scheduler_manager.ALL_GROUPS:
+            await _execute_for_all_groups(schedule, task_meta, bot)
+        else:
+            await _execute_for_single_target(schedule, task_meta, bot)
+    finally:
+        scheduler_manager._running_tasks.discard(schedule_id)
+
+
+async def _execute_for_all_groups(schedule: ScheduleInfo, task_meta: dict, bot):
+    """为所有群组执行任务，并处理优先级覆盖。"""
+    plugin_name = schedule.plugin_name
+
+    concurrency_limit = Config.get_config(
+        "SchedulerManager", SCHEDULE_CONCURRENCY_KEY, 5
+    )
+    if not isinstance(concurrency_limit, int) or concurrency_limit <= 0:
+        logger.warning(
+            f"无效的定时任务并发限制配置 '{concurrency_limit}'，将使用默认值 5。"
+        )
+        concurrency_limit = 5
+
+    logger.info(
+        f"开始执行针对 [所有群组] 的任务 "
+        f"(ID: {schedule.id}, 插件: {plugin_name}, Bot: {bot.self_id})，"
+        f"并发限制: {concurrency_limit}"
+    )
+
+    all_gids = set()
+    try:
+        group_list, _ = await PlatformUtils.get_group_list(bot)
+        all_gids.update(
+            g.group_id for g in group_list if g.group_id and not g.channel_id
+        )
+    except Exception as e:
+        logger.error(f"为 'all' 任务获取 Bot {bot.self_id} 的群列表失败", e=e)
+        return
+
+    specific_tasks_gids = set(
+        await ScheduleInfo.filter(
+            plugin_name=plugin_name, group_id__in=list(all_gids)
+        ).values_list("group_id", flat=True)
+    )
+
+    semaphore = asyncio.Semaphore(concurrency_limit)
+
+    async def worker(gid: str):
+        """使用 Semaphore 包装单个群组的任务执行"""
+        await asyncio.sleep(random.uniform(0, 59))
+        async with semaphore:
+            temp_schedule = copy.deepcopy(schedule)
+            temp_schedule.group_id = gid
+            await _execute_for_single_target(temp_schedule, task_meta, bot)
+            await asyncio.sleep(random.uniform(0.1, 0.5))
+
+    tasks_to_run = []
+    for gid in all_gids:
+        if gid in specific_tasks_gids:
+            logger.debug(f"群组 {gid} 已有特定任务，跳过 'all' 任务的执行。")
+            continue
+        tasks_to_run.append(worker(gid))
+
+    if tasks_to_run:
+        await asyncio.gather(*tasks_to_run)
+
+
+async def _execute_for_single_target(schedule: ScheduleInfo, task_meta: dict, bot):
+    """为单个目标（具体群组或全局）执行任务。"""
+
+    plugin_name = schedule.plugin_name
+    group_id = schedule.group_id
+
+    try:
+        is_blocked = await CommonUtils.task_is_block(bot, plugin_name, group_id)
+        if is_blocked:
+            target_desc = f"群 {group_id}" if group_id else "全局"
+            logger.info(
+                f"插件 '{plugin_name}' 的定时任务在目标 [{target_desc}]"
+                "因功能被禁用而跳过执行。"
+            )
+            return
+
+        max_retries = Config.get_config("SchedulerManager", "JOB_MAX_RETRIES", 2)
+        retry_delay = Config.get_config("SchedulerManager", "JOB_RETRY_DELAY", 10)
+
+        @Retry.simple(
+            stop_max_attempt=max_retries + 1,
+            wait_fixed_seconds=retry_delay,
+            log_name=f"定时任务执行:{schedule.plugin_name}",
+        )
+        async def _execute_task_with_retry():
+            task_func = task_meta["func"]
+            job_kwargs = schedule.job_kwargs
+            if not isinstance(job_kwargs, dict):
+                logger.error(
+                    f"任务 {schedule.id} 的 job_kwargs 不是字典类型: {type(job_kwargs)}"
+                )
+                return
+
+            sig = inspect.signature(task_func)
+            if "bot" in sig.parameters:
+                job_kwargs["bot"] = bot
+
+            await task_func(group_id, **job_kwargs)
+
+        try:
+            logger.info(
+                f"插件 '{schedule.plugin_name}' 开始为目标 "
+                f"[{schedule.group_id or '全局'}] 执行定时任务 (ID: {schedule.id})。"
+            )
+            await _execute_task_with_retry()
+        except Exception as e:
+            logger.error(
+                f"执行定时任务 (ID: {schedule.id}, 插件: {schedule.plugin_name}, "
+                f"目标: {schedule.group_id or '全局'}) 在所有重试后最终失败",
+                e=e,
+            )
+    except Exception as e:
+        logger.error(
+            f"执行定时任务 (ID: {schedule.id}, 插件: {plugin_name}, "
+            f"目标: {group_id or '全局'}) 时发生异常",
+            e=e,
+        )

zhenxun/services/scheduler/lifecycle.py

@@ -0,0 +1,62 @@
+"""
+定时任务的生命周期管理
+
+包含在机器人启动时加载和调度数据库中保存的任务的逻辑。
+"""
+
+from zhenxun.services.log import logger
+from zhenxun.utils.manager.priority_manager import PriorityLifecycle
+
+from .adapter import APSchedulerAdapter
+from .repository import ScheduleRepository
+from .service import scheduler_manager
+
+
+@PriorityLifecycle.on_startup(priority=90)
+async def _load_schedules_from_db():
+    """在服务启动时从数据库加载并调度所有任务。"""
+    logger.info("正在从数据库加载并调度所有定时任务...")
+    schedules = await ScheduleRepository.get_all_enabled()
+    count = 0
+    for schedule in schedules:
+        if schedule.plugin_name in scheduler_manager._registered_tasks:
+            APSchedulerAdapter.add_or_reschedule_job(schedule)
+            count += 1
+        else:
+            logger.warning(f"跳过加载定时任务：插件 '{schedule.plugin_name}' 未注册。")
+    logger.info(f"数据库定时任务加载完成，共成功加载 {count} 个任务。")
+
+    logger.info("正在检查并注册声明式默认任务...")
+    declared_count = 0
+    for task_info in scheduler_manager._declared_tasks:
+        plugin_name = task_info["plugin_name"]
+        group_id = task_info["group_id"]
+        bot_id = task_info["bot_id"]
+
+        query_kwargs = {
+            "plugin_name": plugin_name,
+            "group_id": group_id,
+            "bot_id": bot_id,
+        }
+        exists = await ScheduleRepository.exists(**query_kwargs)
+
+        if not exists:
+            logger.info(f"为插件 '{plugin_name}' 注册新的默认定时任务...")
+            schedule = await scheduler_manager.add_schedule(
+                plugin_name=plugin_name,
+                group_id=group_id,
+                trigger_type=task_info["trigger_type"],
+                trigger_config=task_info["trigger_config"],
+                job_kwargs=task_info["job_kwargs"],
+                bot_id=bot_id,
+            )
+            if schedule:
+                declared_count += 1
+                logger.debug(f"默认任务 '{plugin_name}' 注册成功 (ID: {schedule.id})")
+            else:
+                logger.error(f"默认任务 '{plugin_name}' 注册失败")
+        else:
+            logger.debug(f"插件 '{plugin_name}' 的默认任务已存在于数据库中，跳过注册。")
+
+    if declared_count > 0:
+        logger.info(f"声明式任务检查完成，新注册了 {declared_count} 个默认任务。")

zhenxun/services/scheduler/repository.py

@@ -0,0 +1,79 @@
+"""
+数据持久层 (Repository)
+
+封装所有对 ScheduleInfo 模型的数据库操作，将数据访问逻辑与业务逻辑分离。
+"""
+
+from typing import Any
+
+from tortoise.queryset import QuerySet
+
+from zhenxun.models.schedule_info import ScheduleInfo
+
+
+class ScheduleRepository:
+    """封装 ScheduleInfo 模型的数据库操作"""
+
+    @staticmethod
+    async def get_by_id(schedule_id: int) -> ScheduleInfo | None:
+        """通过ID获取任务"""
+        return await ScheduleInfo.get_or_none(id=schedule_id)
+
+    @staticmethod
+    async def get_all_enabled() -> list[ScheduleInfo]:
+        """获取所有启用的任务"""
+        return await ScheduleInfo.filter(is_enabled=True).all()
+
+    @staticmethod
+    async def get_all(plugin_name: str | None = None) -> list[ScheduleInfo]:
+        """获取所有任务，可按插件名过滤"""
+        if plugin_name:
+            return await ScheduleInfo.filter(plugin_name=plugin_name).all()
+        return await ScheduleInfo.all()
+
+    @staticmethod
+    async def save(schedule: ScheduleInfo, update_fields: list[str] | None = None):
+        """保存任务"""
+        await schedule.save(update_fields=update_fields)
+
+    @staticmethod
+    async def exists(**kwargs: Any) -> bool:
+        """检查任务是否存在"""
+        return await ScheduleInfo.exists(**kwargs)
+
+    @staticmethod
+    async def get_by_plugin_and_group(
+        plugin_name: str, group_ids: list[str]
+    ) -> list[ScheduleInfo]:
+        """根据插件和群组ID列表获取任务"""
+        return await ScheduleInfo.filter(
+            plugin_name=plugin_name, group_id__in=group_ids
+        ).all()
+
+    @staticmethod
+    async def update_or_create(
+        defaults: dict, **kwargs: Any
+    ) -> tuple[ScheduleInfo, bool]:
+        """更新或创建任务"""
+        return await ScheduleInfo.update_or_create(defaults=defaults, **kwargs)
+
+    @staticmethod
+    async def query_schedules(**filters: Any) -> list[ScheduleInfo]:
+        """
+        根据任意条件查询任务列表
+
+        参数:
+            **filters: 过滤条件，如 group_id="123", plugin_name="abc"
+
+        返回:
+            list[ScheduleInfo]: 任务列表
+        """
+        cleaned_filters = {k: v for k, v in filters.items() if v is not None}
+        if not cleaned_filters:
+            return await ScheduleInfo.all()
+        return await ScheduleInfo.filter(**cleaned_filters).all()
+
+    @staticmethod
+    def filter(**kwargs: Any) -> QuerySet[ScheduleInfo]:
+        """提供一个通用的过滤查询接口，供Targeter使用"""
+        return ScheduleInfo.filter(**kwargs)

zhenxun/services/scheduler/service.py

@@ -0,0 +1,448 @@
+"""
+服务层 (Service)
+
+定义 SchedulerManager 类作为定时任务服务的公共 API 入口。
+它负责编排业务逻辑，并调用 Repository 和 Adapter 层来完成具体工作。
+"""
+
+from collections.abc import Callable, Coroutine
+from datetime import datetime
+from typing import Any, ClassVar
+
+import nonebot
+from pydantic import BaseModel
+
+from zhenxun.configs.config import Config
+from zhenxun.models.schedule_info import ScheduleInfo
+from zhenxun.services.log import logger
+
+from .adapter import APSchedulerAdapter
+from .job import _execute_job
+from .repository import ScheduleRepository
+from .targeter import ScheduleTargeter
+
+
+class SchedulerManager:
+    ALL_GROUPS: ClassVar[str] = "__ALL_GROUPS__"
+    _registered_tasks: ClassVar[
+        dict[str, dict[str, Callable | type[BaseModel] | None]]
+    ] = {}
+    _declared_tasks: ClassVar[list[dict[str, Any]]] = []
+    _running_tasks: ClassVar[set] = set()
+
+    def target(self, **filters: Any) -> ScheduleTargeter:
+        """
+        创建目标选择器以执行批量操作
+
+        参数:
+            **filters: 过滤条件，支持plugin_name、group_id、bot_id等字段。
+
+        返回:
+            ScheduleTargeter: 目标选择器对象，可用于批量操作。
+        """
+        return ScheduleTargeter(self, **filters)
+
+    def task(
+        self,
+        trigger: str,
+        group_id: str | None = None,
+        bot_id: str | None = None,
+        **trigger_kwargs,
+    ):
+        """
+        声明式定时任务装饰器
+
+        参数:
+            trigger: 触发器类型，如'cron'、'interval'等。
+            group_id: 目标群组ID，None表示全局任务。
+            bot_id: 目标Bot ID，None表示使用默认Bot。
+            **trigger_kwargs: 触发器配置参数。
+
+        返回:
+            Callable: 装饰器函数。
+        """
+
+        def decorator(func: Callable[..., Coroutine]) -> Callable[..., Coroutine]:
+            try:
+                plugin = nonebot.get_plugin_by_module_name(func.__module__)
+                if not plugin:
+                    raise ValueError(f"函数 {func.__name__} 不在任何已加载的插件中。")
+                plugin_name = plugin.name
+
+                task_declaration = {
+                    "plugin_name": plugin_name,
+                    "func": func,
+                    "group_id": group_id,
+                    "bot_id": bot_id,
+                    "trigger_type": trigger,
+                    "trigger_config": trigger_kwargs,
+                    "job_kwargs": {},
+                }
+                self._declared_tasks.append(task_declaration)
+                logger.debug(
+                    f"发现声明式定时任务 '{plugin_name}'，将在启动时进行注册。"
+                )
+            except Exception as e:
+                logger.error(f"注册声明式定时任务失败: {func.__name__}, 错误: {e}")
+
+            return func
+
+        return decorator
+
+    def register(
+        self, plugin_name: str, params_model: type[BaseModel] | None = None
+    ) -> Callable:
+        """
+        注册可调度的任务函数
+
+        参数:
+            plugin_name: 插件名称，用于标识任务。
+            params_model: 参数验证模型，继承自BaseModel的类。
+
+        返回:
+            Callable: 装饰器函数。
+        """
+
+        def decorator(func: Callable[..., Coroutine]) -> Callable[..., Coroutine]:
+            if plugin_name in self._registered_tasks:
+                logger.warning(f"插件 '{plugin_name}' 的定时任务已被重复注册。")
+            self._registered_tasks[plugin_name] = {
+                "func": func,
+                "model": params_model,
+            }
+            model_name = params_model.__name__ if params_model else "无"
+            logger.debug(
+                f"插件 '{plugin_name}' 的定时任务已注册，参数模型: {model_name}"
+            )
+            return func
+
+        return decorator
+
+    def get_registered_plugins(self) -> list[str]:
+        """
+        获取已注册插件列表
+
+        返回:
+            list[str]: 已注册的插件名称列表。
+        """
+        return list(self._registered_tasks.keys())
+
+    async def add_daily_task(
+        self,
+        plugin_name: str,
+        group_id: str | None,
+        hour: int,
+        minute: int,
+        second: int = 0,
+        job_kwargs: dict | None = None,
+        bot_id: str | None = None,
+    ) -> "ScheduleInfo | None":
+        """
+        添加每日定时任务
+
+        参数:
+            plugin_name: 插件名称。
+            group_id: 目标群组ID，None表示全局任务。
+            hour: 执行小时（0-23）。
+            minute: 执行分钟（0-59）。
+            second: 执行秒数（0-59），默认为0。
+            job_kwargs: 任务参数字典。
+            bot_id: 目标Bot ID，None表示使用默认Bot。
+
+        返回:
+            ScheduleInfo | None: 创建的任务信息，失败时返回None。
+        """
+        trigger_config = {
+            "hour": hour,
+            "minute": minute,
+            "second": second,
+            "timezone": Config.get_config("SchedulerManager", "SCHEDULER_TIMEZONE"),
+        }
+        return await self.add_schedule(
+            plugin_name,
+            group_id,
+            "cron",
+            trigger_config,
+            job_kwargs=job_kwargs,
+            bot_id=bot_id,
+        )
+
+    async def add_interval_task(
+        self,
+        plugin_name: str,
+        group_id: str | None,
+        *,
+        weeks: int = 0,
+        days: int = 0,
+        hours: int = 0,
+        minutes: int = 0,
+        seconds: int = 0,
+        start_date: str | datetime | None = None,
+        job_kwargs: dict | None = None,
+        bot_id: str | None = None,
+    ) -> "ScheduleInfo | None":
+        """添加间隔性定时任务"""
+        trigger_config = {
+            "weeks": weeks,
+            "days": days,
+            "hours": hours,
+            "minutes": minutes,
+            "seconds": seconds,
+            "start_date": start_date,
+        }
+        trigger_config = {k: v for k, v in trigger_config.items() if v}
+        return await self.add_schedule(
+            plugin_name,
+            group_id,
+            "interval",
+            trigger_config,
+            job_kwargs=job_kwargs,
+            bot_id=bot_id,
+        )
+
+    def _validate_and_prepare_kwargs(
+        self, plugin_name: str, job_kwargs: dict | None
+    ) -> tuple[bool, str | dict]:
+        """验证并准备任务参数，应用默认值"""
+        from pydantic import ValidationError
+
+        task_meta = self._registered_tasks.get(plugin_name)
+        if not task_meta:
+            return False, f"插件 '{plugin_name}' 未注册。"
+
+        params_model = task_meta.get("model")
+        job_kwargs = job_kwargs if job_kwargs is not None else {}
+
+        if not params_model:
+            if job_kwargs:
+                logger.warning(
+                    f"插件 '{plugin_name}' 未定义参数模型，但收到了参数: {job_kwargs}"
+                )
+            return True, job_kwargs
+
+        if not (isinstance(params_model, type) and issubclass(params_model, BaseModel)):
+            logger.error(f"插件 '{plugin_name}' 的参数模型不是有效的 BaseModel 类")
+            return False, f"插件 '{plugin_name}' 的参数模型配置错误"
+
+        try:
+            model_validate = getattr(params_model, "model_validate", None)
+            if not model_validate:
+                return False, f"插件 '{plugin_name}' 的参数模型不支持验证"
+
+            validated_model = model_validate(job_kwargs)
+
+            model_dump = getattr(validated_model, "model_dump", None)
+            if not model_dump:
+                return False, f"插件 '{plugin_name}' 的参数模型不支持导出"
+
+            return True, model_dump()
+        except ValidationError as e:
+            errors = [f"  - {err['loc'][0]}: {err['msg']}" for err in e.errors()]
+            error_str = "\n".join(errors)
+            msg = f"插件 '{plugin_name}' 的任务参数验证失败:\n{error_str}"
+            return False, msg
+
+    async def add_schedule(
+        self,
+        plugin_name: str,
+        group_id: str | None,
+        trigger_type: str,
+        trigger_config: dict,
+        job_kwargs: dict | None = None,
+        bot_id: str | None = None,
+    ) -> "ScheduleInfo | None":
+        """
+        添加定时任务（通用方法）
+
+        参数:
+            plugin_name: 插件名称。
+            group_id: 目标群组ID，None表示全局任务。
+            trigger_type: 触发器类型，如'cron'、'interval'等。
+            trigger_config: 触发器配置字典。
+            job_kwargs: 任务参数字典。
+            bot_id: 目标Bot ID，None表示使用默认Bot。
+
+        返回:
+            ScheduleInfo | None: 创建的任务信息，失败时返回None。
+        """
+        if plugin_name not in self._registered_tasks:
+            logger.error(f"插件 '{plugin_name}' 没有注册可用的定时任务。")
+            return None
+
+        is_valid, result = self._validate_and_prepare_kwargs(plugin_name, job_kwargs)
+        if not is_valid:
+            logger.error(f"任务参数校验失败: {result}")
+            return None
+
+        search_kwargs = {"plugin_name": plugin_name, "group_id": group_id}
+        if bot_id and group_id == self.ALL_GROUPS:
+            search_kwargs["bot_id"] = bot_id
+        else:
+            search_kwargs["bot_id__isnull"] = True
+
+        defaults = {
+            "trigger_type": trigger_type,
+            "trigger_config": trigger_config,
+            "job_kwargs": result,
+            "is_enabled": True,
+        }
+
+        schedule, created = await ScheduleRepository.update_or_create(
+            defaults, **search_kwargs
+        )
+        APSchedulerAdapter.add_or_reschedule_job(schedule)
+
+        action = "设置" if created else "更新"
+        logger.info(
+            f"已成功{action}插件 '{plugin_name}' 的定时任务 (ID: {schedule.id})。"
+        )
+        return schedule
+
+    async def get_all_schedules(self) -> list[ScheduleInfo]:
+        """
+        获取所有定时任务信息
+        """
+        return await self.get_schedules()
+
+    async def get_schedules(
+        self,
+        plugin_name: str | None = None,
+        group_id: str | None = None,
+        bot_id: str | None = None,
+    ) -> list[ScheduleInfo]:
+        """
+        根据条件获取定时任务列表
+
+        参数:
+            plugin_name: 插件名称，None表示不限制。
+            group_id: 群组ID，None表示不限制。
+            bot_id: Bot ID，None表示不限制。
+
+        返回:
+            list[ScheduleInfo]: 符合条件的任务信息列表。
+        """
+        return await ScheduleRepository.query_schedules(
+            plugin_name=plugin_name, group_id=group_id, bot_id=bot_id
+        )
+
+    async def update_schedule(
+        self,
+        schedule_id: int,
+        trigger_type: str | None = None,
+        trigger_config: dict | None = None,
+        job_kwargs: dict | None = None,
+    ) -> tuple[bool, str]:
+        """
+        更新定时任务配置
+
+        参数:
+            schedule_id: 任务ID。
+            trigger_type: 新的触发器类型，None表示不更新。
+            trigger_config: 新的触发器配置，None表示不更新。
+            job_kwargs: 新的任务参数，None表示不更新。
+
+        返回:
+            tuple[bool, str]: (是否成功, 结果消息)。
+        """
+        schedule = await ScheduleRepository.get_by_id(schedule_id)
+        if not schedule:
+            return False, f"未找到 ID 为 {schedule_id} 的任务。"
+
+        updated_fields = []
+        if trigger_config is not None:
+            schedule.trigger_config = trigger_config
+            updated_fields.append("trigger_config")
+            if trigger_type is not None and schedule.trigger_type != trigger_type:
+                schedule.trigger_type = trigger_type
+                updated_fields.append("trigger_type")
+
+        if job_kwargs is not None:
+            existing_kwargs = (
+                schedule.job_kwargs.copy()
+                if isinstance(schedule.job_kwargs, dict)
+                else {}
+            )
+            existing_kwargs.update(job_kwargs)
+
+            is_valid, result = self._validate_and_prepare_kwargs(
+                schedule.plugin_name, existing_kwargs
+            )
+            if not is_valid:
+                return False, str(result)
+
+            assert isinstance(result, dict), "验证成功时 result 应该是字典类型"
+            schedule.job_kwargs = result
+            updated_fields.append("job_kwargs")
+
+        if not updated_fields:
+            return True, "没有任何需要更新的配置。"
+
+        await ScheduleRepository.save(schedule, update_fields=updated_fields)
+        APSchedulerAdapter.add_or_reschedule_job(schedule)
+        return True, f"成功更新了任务 ID: {schedule_id} 的配置。"
+
+    async def get_schedule_status(self, schedule_id: int) -> dict | None:
+        """获取定时任务的详细状态信息"""
+        schedule = await ScheduleRepository.get_by_id(schedule_id)
+        if not schedule:
+            return None
+
+        status_from_scheduler = APSchedulerAdapter.get_job_status(schedule.id)
+
+        status_text = (
+            "运行中"
+            if schedule_id in self._running_tasks
+            else ("启用" if schedule.is_enabled else "暂停")
+        )
+
+        return {
+            "id": schedule.id,
+            "bot_id": schedule.bot_id,
+            "plugin_name": schedule.plugin_name,
+            "group_id": schedule.group_id,
+            "is_enabled": status_text,
+            "trigger_type": schedule.trigger_type,
+            "trigger_config": schedule.trigger_config,
+            "job_kwargs": schedule.job_kwargs,
+            **status_from_scheduler,
+        }
+
+    async def pause_schedule(self, schedule_id: int) -> tuple[bool, str]:
+        """暂停指定的定时任务"""
+        schedule = await ScheduleRepository.get_by_id(schedule_id)
+        if not schedule or not schedule.is_enabled:
+            return False, "任务不存在或已暂停。"
+
+        schedule.is_enabled = False
+        await ScheduleRepository.save(schedule, update_fields=["is_enabled"])
+        APSchedulerAdapter.pause_job(schedule_id)
+        return True, f"已暂停任务 (ID: {schedule.id})。"
+
+    async def resume_schedule(self, schedule_id: int) -> tuple[bool, str]:
+        """恢复指定的定时任务"""
+        schedule = await ScheduleRepository.get_by_id(schedule_id)
+        if not schedule or schedule.is_enabled:
+            return False, "任务不存在或已启用。"
+
+        schedule.is_enabled = True
+        await ScheduleRepository.save(schedule, update_fields=["is_enabled"])
+        APSchedulerAdapter.resume_job(schedule_id)
+        return True, f"已恢复任务 (ID: {schedule.id})。"
+
+    async def trigger_now(self, schedule_id: int) -> tuple[bool, str]:
+        """立即手动触发指定的定时任务"""
+        schedule = await ScheduleRepository.get_by_id(schedule_id)
+        if not schedule:
+            return False, f"未找到 ID 为 {schedule_id} 的定时任务。"
+        if schedule.plugin_name not in self._registered_tasks:
+            return False, f"插件 '{schedule.plugin_name}' 没有注册可用的定时任务。"
+
+        try:
+            await _execute_job(schedule.id)
+            return True, f"已手动触发任务 (ID: {schedule.id})。"
+        except Exception as e:
+            logger.error(f"手动触发任务失败: {e}")
+            return False, f"手动触发任务失败: {e}"
+
+
+scheduler_manager = SchedulerManager()

zhenxun/services/scheduler/targeter.py

@@ -0,0 +1,109 @@
+"""
+目标选择器 (Targeter)
+
+提供链式API，用于构建和执行对多个定时任务的批量操作。
+"""
+
+from collections.abc import Callable, Coroutine
+from typing import Any
+
+from .adapter import APSchedulerAdapter
+from .repository import ScheduleRepository
+
+
+class ScheduleTargeter:
+    """
+    一个用于构建和执行定时任务批量操作的目标选择器。
+    """
+
+    def __init__(self, manager: Any, **filters: Any):
+        """初始化目标选择器"""
+        self._manager = manager
+        self._filters = {k: v for k, v in filters.items() if v is not None}
+
+    async def _get_schedules(self):
+        """根据过滤器获取任务"""
+        query = ScheduleRepository.filter(**self._filters)
+        return await query.all()
+
+    def _generate_target_description(self) -> str:
+        """根据过滤条件生成友好的目标描述"""
+        if "id" in self._filters:
+            return f"任务 ID {self._filters['id']} 的"
+
+        parts = []
+        if "group_id" in self._filters:
+            group_id = self._filters["group_id"]
+            if group_id == self._manager.ALL_GROUPS:
+                parts.append("所有群组中")
+            else:
+                parts.append(f"群 {group_id} 中")
+
+        if "plugin_name" in self._filters:
+            parts.append(f"插件 '{self._filters['plugin_name']}' 的")
+
+        if not parts:
+            return "所有"
+
+        return "".join(parts)
+
+    async def _apply_operation(
+        self,
+        operation_func: Callable[[int], Coroutine[Any, Any, tuple[bool, str]]],
+        operation_name: str,
+    ) -> tuple[int, str]:
+        """通用的操作应用模板"""
+        schedules = await self._get_schedules()
+        if not schedules:
+            target_desc = self._generate_target_description()
+            return 0, f"没有找到{target_desc}可供{operation_name}的任务。"
+
+        success_count = 0
+        for schedule in schedules:
+            success, _ = await operation_func(schedule.id)
+            if success:
+                success_count += 1
+
+        target_desc = self._generate_target_description()
+        return (
+            success_count,
+            f"成功{operation_name}了{target_desc} {success_count} 个任务。",
+        )
+
+    async def pause(self) -> tuple[int, str]:
+        """
+        暂停匹配的定时任务
+
+        返回:
+            tuple[int, str]: (成功暂停的任务数量, 操作结果消息)。
+        """
+        return await self._apply_operation(self._manager.pause_schedule, "暂停")
+
+    async def resume(self) -> tuple[int, str]:
+        """
+        恢复匹配的定时任务
+
+        返回:
+            tuple[int, str]: (成功恢复的任务数量, 操作结果消息)。
+        """
+        return await self._apply_operation(self._manager.resume_schedule, "恢复")
+
+    async def remove(self) -> tuple[int, str]:
+        """
+        移除匹配的定时任务
+
+        返回:
+            tuple[int, str]: (成功移除的任务数量, 操作结果消息)。
+        """
+        schedules = await self._get_schedules()
+        if not schedules:
+            target_desc = self._generate_target_description()
+            return 0, f"没有找到{target_desc}可供移除的任务。"
+
+        for schedule in schedules:
+            APSchedulerAdapter.remove_job(schedule.id)
+
+        query = ScheduleRepository.filter(**self._filters)
+        count = await query.delete()
+        target_desc = self._generate_target_description()
+        return count, f"成功移除了{target_desc} {count} 个任务。"