#!/usr/bin/env python3 # -*- coding: utf-8 -*- # scheduler.py —— 定时任务：存储、下次触发时间计算、异步调度循环 """ 定时任务只做一件事：到点把任务的 prompt 通过 WebSocket 推给网页， 网页那边（Tampermonkey）把它发进输入框，等于替用户发了一条消息。 支持四种周期： once 只跑一次，参数 at="2026-11-06 09:00" interval 每 N 分钟，参数 every_minutes=30（最少 1 分钟） daily 每天固定时刻，参数 time="09:00" cron 五段 cron，参数 cron="30 9 * * 1-5"（分 时 日 月 周） 存储用 JSON（默认 <仓库>/user/schedules.json），写入走临时文件 + os.replace， 避免容器中途被杀留下半个文件。本模块只依赖标准库，不引入调度框架。 """ import asyncio import copy import json import logging import os import re import threading import uuid from datetime import datetime, timedelta, timezone from pathlib import Path try: from zoneinfo import ZoneInfo except ImportError: ZoneInfo = None logger = logging.getLogger("AgentScheduler") DEFAULT_TZ_NAME = os.getenv("AGENT_SCHEDULE_TZ", "Asia/Shanghai") KINDS = ("once", "interval", "daily", "cron") MIN_INTERVAL_SECONDS = 60 MAX_INTERVAL_SECONDS = 30 * 24 * 3600 MAX_CRON_DAYS = 366 * 5 # --------------------------------------------------------- # 时区与时间解析 # --------------------------------------------------------- _TZ_CACHE = {} def resolve_tz(name=None): """取时区；容器里没有 tzdata 时退回固定 +08:00，不让整个调度器挂掉。""" name = (name or DEFAULT_TZ_NAME or "Asia/Shanghai").strip() if name in _TZ_CACHE: return _TZ_CACHE[name] tz = None if ZoneInfo is not None: try: tz = ZoneInfo(name) except Exception: tz = None if tz is None: m = re.fullmatch(r"(?:UTC|GMT)?\s*([+-])(\d{1,2})(?::?(\d{2}))?", name) if m: hours = int(m.group(2)) + int(m.group(3) or 0) / 60.0 tz = timezone(timedelta(hours=-hours if m.group(1) == "-" else hours)) else: tz = timezone(timedelta(hours=8)) _TZ_CACHE[name] = tz return tz def iso(dt): """统一存成带偏移的 ISO 字符串，秒级精度。""" if dt is None: return None return dt.isoformat(timespec="seconds") def parse_stored_dt(text, tz=None): if not text: return None if isinstance(text, datetime): dt = text else: try: dt = datetime.fromisoformat(str(text).strip()) except ValueError: return None if dt.tzinfo is None: dt = dt.replace(tzinfo=tz or resolve_tz()) return dt def parse_local_datetime(text, tz=None): """把 "2026-11-06 09:00" 之类的本地时间解析成带时区的 datetime。""" tz = tz or resolve_tz() text = str(text or "").strip() if not text: raise ValueError("时间不能为空（示例：2026-11-06 09:00）") dt = None try: dt = datetime.fromisoformat(text) except ValueError: for fmt in ("%Y-%m-%d %H:%M", "%Y-%m-%d %H:%M:%S", "%Y-%m-%d", "%Y/%m/%d %H:%M", "%Y/%m/%d"): try: dt = datetime.strptime(text.replace("T", " "), fmt) break except ValueError: continue if dt is None: raise ValueError(f"无法识别的时间格式：{text}（用 YYYY-MM-DD HH:MM）") if dt.tzinfo is None: dt = dt.replace(tzinfo=tz) return dt def parse_clock(text, tz=None): """"09:00" → 今天/明天最近的一次。""" text = str(text or "").strip() m = re.fullmatch(r"(\d{1,2}):(\d{2})(?::(\d{2}))?", text) if not m: raise ValueError(f"每日时间格式不对：{text}（用 HH:MM）") hour, minute = int(m.group(1)), int(m.group(2)) if not (0 <= hour <= 23 and 0 <= minute <= 59): raise ValueError(f"每日时间超出范围：{text}") return hour, minute # --------------------------------------------------------- # cron 解析 # --------------------------------------------------------- def _parse_cron_field(field, lo, hi): values = set() for part in str(field).split(","): part = part.strip() if not part: raise ValueError("cron 字段里有空项") step = 1 if "/" in part: base, step_text = part.split("/", 1) try: step = int(step_text) except ValueError: raise ValueError(f"cron 步长不是数字：{part}") if step <= 0: raise ValueError(f"cron 步长必须大于 0：{part}") else: base = part if base == "*": start, end = lo, hi elif "-" in base: try: a, b = base.split("-", 1) start, end = int(a), int(b) except ValueError: raise ValueError(f"cron 区间不是数字：{part}") else: try: start = end = int(base) except ValueError: raise ValueError(f"cron 字段不是数字：{part}") if start > end: raise ValueError(f"cron 区间起点大于终点：{part}") if start < lo or end > hi: raise ValueError(f"cron 字段超出范围 {lo}-{hi}：{part}") values.update(range(start, end + 1, step)) return values def parse_cron(expr): parts = str(expr or "").split() if len(parts) != 5: raise ValueError('cron 需要 5 段（分 时 日 月 周），例如 "30 9 * * 1-5"') minutes = _parse_cron_field(parts[0], 0, 59) hours = _parse_cron_field(parts[1], 0, 23) doms = _parse_cron_field(parts[2], 1, 31) months = _parse_cron_field(parts[3], 1, 12) dows = _parse_cron_field(parts[4], 0, 7) if 7 in dows: # 周日两种写法都收 dows.discard(7) dows.add(0) return { "minute": minutes, "hour": hours, "day": doms, "month": months, "weekday": dows, "day_any": parts[2].strip() == "*", "weekday_any": parts[4].strip() == "*", } def cron_next(expr, after, tz=None): """返回 after 之后（不含）最近一次 cron 命中时间；找不到返回 None。""" tz = tz or resolve_tz() spec = parse_cron(expr) cur = after.astimezone(tz).replace(second=0, microsecond=0) + timedelta(minutes=1) limit = cur + timedelta(days=MAX_CRON_DAYS) while cur < limit: if cur.month not in spec["month"]: if cur.month == 12: cur = cur.replace(year=cur.year + 1, month=1, day=1, hour=0, minute=0) else: cur = cur.replace(month=cur.month + 1, day=1, hour=0, minute=0) continue day_ok = cur.day in spec["day"] dow_ok = (cur.isoweekday() % 7) in spec["weekday"] if spec["day_any"] or spec["weekday_any"]: match_day = day_ok and dow_ok # 未限制的一边视为全集 else: match_day = day_ok or dow_ok # 标准 cron：两边都限定时取并集 if not match_day: cur = (cur + timedelta(days=1)).replace(hour=0, minute=0) continue if cur.hour not in spec["hour"]: cur = (cur + timedelta(hours=1)).replace(minute=0) continue if cur.minute not in spec["minute"]: cur = cur + timedelta(minutes=1) continue return cur return None # --------------------------------------------------------- # 任务的规格与下次触发时间 # --------------------------------------------------------- def normalize_spec(kind, at=None, every_minutes=None, every_seconds=None, time=None, cron=None): kind = str(kind or "").strip().lower() if kind not in KINDS: raise ValueError(f"未知的周期类型：{kind}（可选 {'/'.join(KINDS)}）") if kind == "once": if not at: raise ValueError('kind=once 需要 at，例如 "2026-11-06 09:00"') return {"at": str(at).strip()} if kind == "interval": if every_seconds is None: if every_minutes is None: raise ValueError("kind=interval 需要 every_minutes") try: every_seconds = int(every_minutes) * 60 except (TypeError, ValueError): raise ValueError(f"every_minutes 不是数字：{every_minutes}") else: try: every_seconds = int(every_seconds) except (TypeError, ValueError): raise ValueError(f"every_seconds 不是数字：{every_seconds}") if every_seconds < MIN_INTERVAL_SECONDS: raise ValueError(f"间隔不能小于 {MIN_INTERVAL_SECONDS} 秒") if every_seconds > MAX_INTERVAL_SECONDS: raise ValueError("间隔不能超过 30 天") return {"every_seconds": every_seconds} if kind == "daily": if not time: raise ValueError('kind=daily 需要 time，例如 "09:00"') hour, minute = parse_clock(time) return {"time": f"{hour:02d}:{minute:02d}"} if not cron: raise ValueError('kind=cron 需要 cron，例如 "30 9 * * 1-5"') parse_cron(cron) # 提前校验 return {"cron": str(cron).strip()} def compute_next(kind, spec, tz_name, after): """根据任务规格算 after 之后的下一次触发时间。""" tz = resolve_tz(tz_name) if kind == "once": dt = parse_local_datetime(spec.get("at"), tz) return dt if dt > after else None if kind == "interval": return after + timedelta(seconds=int(spec.get("every_seconds") or 0)) if kind == "daily": hour, minute = parse_clock(spec.get("time"), tz) candidate = after.astimezone(tz).replace( hour=hour, minute=minute, second=0, microsecond=0) if candidate <= after: candidate += timedelta(days=1) return candidate if kind == "cron": return cron_next(spec.get("cron"), after, tz) raise ValueError(f"未知的周期类型：{kind}") def human_task(task): """给日志/工具输出用的一行摘要。""" if not task: return "" spec = task.get("spec") or {} if task.get("kind") == "once": when = spec.get("at", "") elif task.get("kind") == "interval": when = f"每 {int(spec.get('every_seconds') or 0) // 60} 分钟" elif task.get("kind") == "daily": when = f"每天 {spec.get('time', '')}" else: when = f"cron {spec.get('cron', '')}" state = "启用" if task.get("enabled") else "停用" return f"{task.get('id')} | {task.get('name')} | {when} | 下次 {task.get('next_run') or '无'} | {state}" # --------------------------------------------------------- # 存储 # --------------------------------------------------------- class ScheduleStore: def __init__(self, path, missed_grace=None): self.path = Path(path) self.missed_grace = max( 0, int(os.getenv("AGENT_SCHEDULE_MISSED_GRACE", "300"))) \ if missed_grace is None else max(0, int(missed_grace)) self._lock = threading.RLock() self._tasks = [] self.load() # ---------- 持久化 ---------- def load(self): with self._lock: self._tasks = [] if not self.path.exists(): return try: raw = json.loads(self.path.read_text(encoding="utf-8")) except (OSError, ValueError) as e: logger.error(f"读取定时任务文件失败（{self.path}）：{e}") # 坏文件先备份，别让下一次保存把原内容盖掉 try: backup = self.path.with_name(self.path.name + ".broken") os.replace(self.path, backup) logger.error(f"坏文件已备份到：{backup}") except OSError: pass return items = raw.get("tasks") if isinstance(raw, dict) else raw for item in items or []: if isinstance(item, dict) and item.get("id") and item.get("kind") in KINDS: item.setdefault("enabled", True) item.setdefault("timezone", DEFAULT_TZ_NAME) item.setdefault("spec", {}) item.setdefault("run_count", 0) self._tasks.append(item) def save(self): with self._lock: self.path.parent.mkdir(parents=True, exist_ok=True) payload = {"version": 1, "tasks": self._tasks} tmp = self.path.with_name(self.path.name + ".tmp") tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") os.replace(tmp, self.path) # ---------- 查询 ---------- def list_tasks(self, include_disabled=True): with self._lock: items = [copy.deepcopy(t) for t in self._tasks if include_disabled or t.get("enabled")] items.sort(key=lambda t: (t.get("next_run") or "9999", t.get("id"))) return items def get(self, task_id): with self._lock: for task in self._tasks: if task.get("id") == task_id: return copy.deepcopy(task) return None def due(self, now=None): now = now or datetime.now(resolve_tz()) result = [] with self._lock: for task in self._tasks: if not task.get("enabled"): continue nxt = parse_stored_dt(task.get("next_run")) if nxt is not None and nxt <= now: result.append(copy.deepcopy(task)) result.sort(key=lambda t: t.get("next_run") or "") return result # ---------- 变更 ---------- def add(self, name, prompt, kind, spec, timezone_name=None, now=None): tz_name = (timezone_name or DEFAULT_TZ_NAME).strip() or DEFAULT_TZ_NAME tz = resolve_tz(tz_name) now = now or datetime.now(tz) name = str(name or "").strip() or "未命名任务" prompt = str(prompt or "").strip() if not prompt: raise ValueError("prompt 不能为空（说明到点要网页 AI 做什么）") task = { "id": "sch_" + uuid.uuid4().hex[:8], "name": name[:80], "prompt": prompt, "kind": kind, "spec": spec, "timezone": tz_name, "enabled": True, "created_at": iso(now), "updated_at": iso(now), "next_run": None, "last_run": None, "last_status": None, "last_error": None, "run_count": 0, } nxt = compute_next(kind, spec, tz_name, now) if nxt is None: raise ValueError("这个时间已经过去了，请给一个将来的时间") task["next_run"] = iso(nxt) with self._lock: self._tasks.append(task) self.save() return copy.deepcopy(task) def delete(self, task_id): with self._lock: before = len(self._tasks) self._tasks = [t for t in self._tasks if t.get("id") != task_id] if len(self._tasks) == before: return False self.save() return True def set_enabled(self, task_id, enabled, now=None): now = now or datetime.now(resolve_tz()) with self._lock: for task in self._tasks: if task.get("id") != task_id: continue if enabled: nxt = compute_next(task["kind"], task.get("spec") or {}, task.get("timezone"), now) if nxt is None: raise ValueError("这个一次性任务的时间已经过去，请删除后重建") task["next_run"] = iso(nxt) task["enabled"] = bool(enabled) task["updated_at"] = iso(now) self.save() return copy.deepcopy(task) return None def mark_run(self, task_id, status="ok", error=None, now=None): """任务触发后调用：记录结果，一次性任务停用，周期任务算下一次。""" now = now or datetime.now(resolve_tz()) with self._lock: for task in self._tasks: if task.get("id") != task_id: continue task["last_run"] = iso(now) task["last_status"] = status task["last_error"] = (str(error)[:500] if error else None) task["run_count"] = int(task.get("run_count") or 0) + 1 if task.get("kind") == "once": task["enabled"] = False task["next_run"] = None else: nxt = compute_next(task["kind"], task.get("spec") or {}, task.get("timezone"), now) task["next_run"] = iso(nxt) if nxt is None: task["enabled"] = False task["updated_at"] = iso(now) self.save() return copy.deepcopy(task) return None def normalize(self, now=None): """启动时校正：过期的一次性任务标记错过，周期任务跳过积压，只保留最近一次。""" now = now or datetime.now(resolve_tz()) changed = False with self._lock: for task in self._tasks: if not task.get("enabled"): continue nxt = parse_stored_dt(task.get("next_run")) try: if nxt is None: nxt = compute_next(task["kind"], task.get("spec") or {}, task.get("timezone"), now) task["next_run"] = iso(nxt) changed = True continue if nxt > now: continue late = (now - nxt).total_seconds() if late <= self.missed_grace: continue # 还在宽限期内，这一轮直接补跑 if task.get("kind") == "once": task["enabled"] = False task["next_run"] = None task["last_status"] = "missed" task["last_error"] = "服务重启，错过触发时间" else: task["next_run"] = iso( compute_next(task["kind"], task.get("spec") or {}, task.get("timezone"), now)) changed = True except (ValueError, TypeError) as e: task["enabled"] = False task["last_status"] = "error" task["last_error"] = f"计算下次时间失败：{e}" changed = True if changed: self.save() return changed # --------------------------------------------------------- # 执行历史（看板展示用） # --------------------------------------------------------- HISTORY_STATUSES = ("pushed", "sent", "deferred", "failed", "ignored", "skipped") class RunHistory: """ 每次定时任务到点记一条：先记 pushed，页面回执后再改成 sent/deferred/failed/ignored； 没有页面接手、超过等待上限时记 skipped。 文件是 JSON Lines，内存里最多保留 max_records 条，改动后整体重写（几百条很轻）。 """ def __init__(self, path, max_records=None): self.path = Path(path) if max_records is None: try: max_records = int(os.getenv("AGENT_SCHEDULE_HISTORY_MAX", "500")) except ValueError: max_records = 500 self.max_records = max(1, int(max_records)) self._lock = threading.RLock() self._records = [] self.load() def load(self): with self._lock: self._records = [] if not self.path.exists(): return try: lines = self.path.read_text(encoding="utf-8").splitlines() except OSError as e: logger.error(f"读取执行历史失败（{self.path}）：{e}") return for line in lines: line = line.strip() if not line: continue try: rec = json.loads(line) except ValueError: continue if isinstance(rec, dict) and rec.get("delivery_id") and rec.get("status"): self._records.append(rec) self._records = self._records[-self.max_records:] def _save(self): self.path.parent.mkdir(parents=True, exist_ok=True) tmp = self.path.with_name(self.path.name + ".tmp") text = "\n".join(json.dumps(r, ensure_ascii=False) for r in self._records) tmp.write_text(text + ("\n" if text else ""), encoding="utf-8") os.replace(tmp, self.path) def append(self, task, status, note="", delivery_id=None, now=None): now = now or datetime.now(resolve_tz()) rec = { "delivery_id": delivery_id or ("dlv_" + uuid.uuid4().hex[:10]), "ts": iso(now), "task_id": task.get("id"), "name": task.get("name") or "定时任务", "kind": task.get("kind"), "status": status if status in HISTORY_STATUSES else "pushed", "note": str(note or "")[:300], } with self._lock: self._records.append(rec) if len(self._records) > self.max_records: self._records = self._records[-self.max_records:] try: self._save() except OSError as e: logger.error(f"写入执行历史失败：{e}") return copy.deepcopy(rec) def update(self, delivery_id, status, note=None, now=None): """页面回执更新对应记录；找不到（已滚动掉）返回 False。""" if not delivery_id: return False with self._lock: for rec in reversed(self._records): if rec.get("delivery_id") != delivery_id: continue if status in HISTORY_STATUSES: rec["status"] = status if note is not None: rec["note"] = str(note)[:300] rec["updated_at"] = iso(now or datetime.now(resolve_tz())) try: self._save() except OSError as e: logger.error(f"写入执行历史失败：{e}") return True return False def list(self, limit=100): try: limit = max(1, min(int(limit), self.max_records)) except (TypeError, ValueError): limit = 100 with self._lock: items = self._records[-limit:] return [copy.deepcopy(r) for r in reversed(items)] def count(self): with self._lock: return len(self._records) # --------------------------------------------------------- # 异步调度循环 # --------------------------------------------------------- class Scheduler: """ tick 时取 due 任务交给 deliver(task)； deliver 返回 True 表示已经推给某个页面，返回 False 表示暂时没有页面可推， 任务保持待发，最多等 max_delay 秒，超时就记 skipped 并推进到下一次。 """ def __init__(self, store, deliver, tick_seconds=None, max_delay=None, history=None): self.store = store self.deliver = deliver self.history = history self.tick_seconds = max(1, int( tick_seconds if tick_seconds is not None else os.getenv("AGENT_SCHEDULE_TICK", "5"))) self.max_delay = max(0, int( max_delay if max_delay is not None else os.getenv("AGENT_SCHEDULE_MAX_DELAY", "600"))) self._pending_since = {} def _record(self, task, status, note, now, delivery_id): if self.history is None: return try: self.history.append(task, status, note=note, delivery_id=delivery_id, now=now) except Exception as e: # 看板记录失败不能影响调度本身 logger.error(f"写执行历史失败（{task.get('id')}）：{e}") async def tick(self, now=None): now = now or datetime.now(resolve_tz()) fired = [] for original in self.store.due(now): # 每次推送带一个投递号，页面回执时按它回写执行记录 task = dict(original) delivery_id = "dlv_" + uuid.uuid4().hex[:10] task["delivery_id"] = delivery_id ok = False try: ok = bool(await self.deliver(task)) except Exception as e: logger.error(f"推送定时任务 {task.get('id')} 失败：{e}", exc_info=True) if ok: self._pending_since.pop(task["id"], None) self.store.mark_run(task["id"], "ok", now=now) self._record(task, "pushed", "", now, delivery_id) logger.info(f"⏰ 已推送定时任务：{human_task(task)}") fired.append(task["id"]) continue since = self._pending_since.setdefault(task["id"], now) waited = (now - since).total_seconds() if waited >= self.max_delay: self._pending_since.pop(task["id"], None) note = "页面未连接或未就绪，超过等待上限" self.store.mark_run(task["id"], "skipped", error=note, now=now) self._record(task, "skipped", note, now, delivery_id) logger.warning(f"⏰ 定时任务无人接手，已跳过：{human_task(task)}") else: logger.info(f"⏰ 定时任务等待页面连接（{int(waited)}s/{self.max_delay}s）：{task['id']}") return fired async def run(self): logger.info(f"⏰ 定时调度启动：每 {self.tick_seconds}s 检查一次，任务文件 {self.store.path}") while True: try: await self.tick() except asyncio.CancelledError: raise except Exception as e: logger.error(f"定时调度循环异常：{e}", exc_info=True) await asyncio.sleep(self.tick_seconds) # --------------------------------------------------------- # 进程内单例（工具和调度器共用同一份内存状态） # --------------------------------------------------------- _STORE = None _HISTORY = None _STORE_LOCK = threading.Lock() def default_store_path(): env_path = os.getenv("AGENT_SCHEDULE_FILE", "").strip() if env_path: return Path(env_path) return Path(__file__).resolve().parent / "user" / "schedules.json" def schedule_enabled(): return os.getenv("AGENT_SCHEDULE_ENABLED", "1").strip().lower() not in ( "0", "false", "no", "off") def get_store(): global _STORE with _STORE_LOCK: if _STORE is None: _STORE = ScheduleStore(default_store_path()) return _STORE def set_store(store): """测试或热替换用；传 None 则下次 get_store() 重新按环境变量建。""" global _STORE with _STORE_LOCK: _STORE = store return _STORE def default_history_path(): env_path = os.getenv("AGENT_SCHEDULE_RUNS_FILE", "").strip() if env_path: return Path(env_path) return Path(__file__).resolve().parent / "user" / "schedule_runs.jsonl" def get_history(): global _HISTORY with _STORE_LOCK: if _HISTORY is None: _HISTORY = RunHistory(default_history_path()) return _HISTORY def set_history(history): global _HISTORY with _STORE_LOCK: _HISTORY = history return _HISTORY