scout/automation/cron.py

Rama main · réplica de lectura del repositorio original.

"""定时任务 — 借鉴 OpenClaw 的一等公民 Cron 设计.

Cron 任务是 Agent 任务（不是 shell 任务），可以让 Agent 定时执行。

优化 (2026-08-05):
- 任务触发时通过 EventBus 广播 notification 事件，实现前端实时推送
"""

from __future__ import annotations

import asyncio
import re
from datetime import datetime, timedelta, time as dtime
from typing import Any, Callable, Coroutine


class CronTask:
    """定时任务定义."""

    def __init__(
        self,
        name: str,
        schedule: str,
        task: str,
        agent_id: str = "default",
    ):
        self.name = name
        self.schedule = schedule  # cron 表达式或自然语言
        self.task = task  # 任务描述（给 Agent 执行）
        self.agent_id = agent_id
        self.last_run: datetime | None = None
        self.next_run: datetime | None = None
        self.enabled = True

    def parse_interval(self) -> int | None:
        """解析简单间隔表达式，返回秒数."""
        # "每X分钟" / "每X小时" / "每X秒"
        m = re.match(r"每(\d+)?\s*(秒|分|小时|天)", self.schedule)
        if not m:
            return None
        num = int(m.group(1)) if m.group(1) else 1
        unit = m.group(2)
        if unit == "秒":
            return num
        elif unit == "分":
            return num * 60
        elif unit == "小时":
            return num * 3600
        elif unit == "天":
            return num * 86400
        return None

    def parse_daily_time(self) -> dtime | None:
        """解析每日定点表达式，返回 time 对象.

        支持格式（2026-08-13）:
        - "每天09:00" / "每天 9:30" / "每日21:15"
        - "09:00" (裸时间也按每日定点)
        """
        m = re.match(r"^(?:每天|每日)?\s*(\d{1,2}):(\d{2})$", self.schedule.strip())
        if not m:
            return None
        hour, minute = int(m.group(1)), int(m.group(2))
        if hour > 23 or minute > 59:
            return None
        return dtime(hour=hour, minute=minute)

    def to_dict(self) -> dict:
        return {
            "name": self.name,
            "schedule": self.schedule,
            "task": self.task,
            "agent_id": self.agent_id,
            "enabled": self.enabled,
            "last_run": self.last_run.isoformat() if self.last_run else None,
            "next_run": self.next_run.isoformat() if self.next_run else None,
        }


class CronManager:
    """定时任务管理器."""

    def __init__(self):
        self._tasks: dict[str, CronTask] = {}
        self._runner_task: asyncio.Task | None = None
        self._agent_callback: Callable[..., Coroutine] | None = None
        self._bus = None  # EventBus 引用

    def set_agent_callback(self, callback: Callable[..., Coroutine]):
        """设置 Agent 执行回调 — 当 Cron 任务触发时调用."""
        self._agent_callback = callback

    def set_bus(self, bus):
        """设置 EventBus 引用 — 用于广播通知事件."""
        self._bus = bus

    def add(self, task: CronTask) -> None:
        """添加定时任务."""
        interval = task.parse_interval()
        if interval:
            task.next_run = datetime.now() + timedelta(seconds=interval)
        elif task.parse_daily_time():
            task.next_run = self._next_daily_run(task.parse_daily_time())
        self._tasks[task.name] = task

    @staticmethod
    def _next_daily_run(t: dtime, after: datetime | None = None) -> datetime:
        """计算每日定点任务的下次运行时间."""
        base = after or datetime.now()
        candidate = base.replace(hour=t.hour, minute=t.minute, second=0, microsecond=0)
        if candidate <= base:
            candidate += timedelta(days=1)
        return candidate

    def remove(self, name: str) -> None:
        """删除定时任务."""
        self._tasks.pop(name, None)

    def list_tasks(self) -> list[CronTask]:
        """列出所有任务."""
        return list(self._tasks.values())

    def get_task(self, name: str) -> CronTask | None:
        """获取任务."""
        return self._tasks.get(name)

    def enable(self, name: str) -> None:
        if name in self._tasks:
            self._tasks[name].enabled = True

    def disable(self, name: str) -> None:
        if name in self._tasks:
            self._tasks[name].enabled = False

    async def start(self) -> None:
        """启动 Cron 调度器."""
        self._runner_task = asyncio.create_task(self._run_loop())

    async def stop(self) -> None:
        """停止调度器."""
        if self._runner_task:
            self._runner_task.cancel()
            self._runner_task = None

    async def _run_loop(self) -> None:
        """调度循环 — 每秒检查."""
        # 延迟导入避免循环依赖
        from scout.bus.hub import bus

        while True:
            now = datetime.now()
            for task in self._tasks.values():
                if not task.enabled:
                    continue
                if task.next_run and now >= task.next_run:
                    # 触发任务
                    if self._agent_callback:
                        try:
                            await self._agent_callback(task)
                        except Exception:
                            pass
                    
                    # 更新运行时间
                    task.last_run = now
                    interval = task.parse_interval()
                    if interval:
                        task.next_run = now + timedelta(seconds=interval)
                    elif task.parse_daily_time():
                        task.next_run = self._next_daily_run(task.parse_daily_time(), now)
                    
                    # ★ 断裂点修复 1: 广播通知事件到 EventBus
                    # 使用全局 bus 或注入的 bus
                    event_bus = self._bus or bus
                    await event_bus.emit("notification", {
                        "type": "cron_triggered",
                        "title": f"定时任务触发: {task.name}",
                        "message": task.task,
                        "task_name": task.name,
                        "timestamp": now.isoformat(),
                    })
                    
            await asyncio.sleep(1)