126 lines
4.1 KiB
Python
126 lines
4.1 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""定时调度器: 支持间隔调度与 5 段 cron 表达式"""
|
|
import threading
|
|
from datetime import datetime, timedelta
|
|
|
|
import store
|
|
|
|
|
|
# ---------------- cron 工具 ----------------
|
|
|
|
def _field(f, lo, hi):
|
|
out = set()
|
|
for part in str(f).split(","):
|
|
part = part.strip()
|
|
if not part:
|
|
continue
|
|
if "/" in part:
|
|
rng, step = part.split("/")
|
|
step = int(step)
|
|
if rng == "*":
|
|
out.update(range(lo, hi + 1, step))
|
|
elif "-" in rng:
|
|
a, b = map(int, rng.split("-"))
|
|
out.update(range(a, b + 1, step))
|
|
else:
|
|
out.update(range(int(rng), hi + 1, step))
|
|
elif "-" in part:
|
|
a, b = map(int, part.split("-"))
|
|
out.update(range(a, b + 1))
|
|
elif part == "*":
|
|
out.update(range(lo, hi + 1))
|
|
else:
|
|
out.add(int(part))
|
|
return out
|
|
|
|
|
|
def cron_next(expr, base=None):
|
|
"""计算 cron 表达式(分 时 日 月 周, 周:0/7=周日)的下一次执行时间"""
|
|
base = base or datetime.now()
|
|
parts = str(expr).strip().split()
|
|
if len(parts) != 5:
|
|
raise ValueError("cron 表达式必须为 5 段: 分 时 日 月 周")
|
|
mins = _field(parts[0], 0, 59)
|
|
hours = _field(parts[1], 0, 23)
|
|
doms = _field(parts[2], 1, 31)
|
|
mons = _field(parts[3], 1, 12)
|
|
dows = _field(parts[4], 0, 7)
|
|
dows = {d % 7 for d in dows} # cron 0/7=周日 -> python 0=周一..6=周日
|
|
t = base.replace(second=0, microsecond=0) + timedelta(minutes=1)
|
|
for _ in range(366 * 24 * 60): # 一年窗口
|
|
if (t.minute in mins and t.hour in hours and t.day in doms
|
|
and t.month in mons and t.weekday() in dows):
|
|
return t
|
|
t += timedelta(minutes=1)
|
|
return None
|
|
|
|
|
|
def interval_delta(sch):
|
|
unit = sch.get("interval_unit", "hours")
|
|
val = int(sch.get("interval_value", 24) or 24)
|
|
if unit == "minutes":
|
|
return timedelta(minutes=val)
|
|
if unit == "days":
|
|
return timedelta(days=val)
|
|
return timedelta(hours=val)
|
|
|
|
|
|
# ---------------- 调度器线程 ----------------
|
|
|
|
class Scheduler(threading.Thread):
|
|
"""每 30 秒检查一次所有定时任务, 到点则触发一次爬取"""
|
|
|
|
def __init__(self, spawn):
|
|
"""
|
|
spawn: callable(task) -> 启动一次运行 (由 app 提供)
|
|
"""
|
|
super().__init__(daemon=True, name="scheduler")
|
|
self._spawn = spawn
|
|
self._stop = threading.Event()
|
|
|
|
def stop(self):
|
|
self._stop.set()
|
|
|
|
def run(self):
|
|
while not self._stop.wait(30):
|
|
try:
|
|
self._tick()
|
|
except Exception:
|
|
pass
|
|
|
|
def _tick(self):
|
|
now = datetime.now()
|
|
for task in store.load_tasks():
|
|
sch = task.get("schedule") or {}
|
|
if task.get("mode") != "scheduled" or not sch.get("enabled"):
|
|
continue
|
|
nxt = sch.get("next_run")
|
|
if not nxt:
|
|
continue
|
|
try:
|
|
due = datetime.strptime(nxt, "%Y-%m-%d %H:%M:%S")
|
|
except Exception:
|
|
continue
|
|
if due > now:
|
|
continue
|
|
runs = store.get_runs(task["id"])
|
|
if runs and runs[-1].get("status") == "running":
|
|
continue # 上次还没跑完, 跳过本次
|
|
self._spawn(task)
|
|
try:
|
|
if sch.get("type") == "cron":
|
|
nn = cron_next(sch.get("cron", "0 * * * *"), now)
|
|
else:
|
|
nn = now + interval_delta(sch)
|
|
sch["next_run"] = nn.strftime("%Y-%m-%d %H:%M:%S") if nn else None
|
|
sch["last_run"] = now.strftime("%Y-%m-%d %H:%M:%S")
|
|
sch["runs_count"] = sch.get("runs_count", 0) + 1
|
|
store.upsert_task(task)
|
|
try:
|
|
import db
|
|
db.upsert_task(task)
|
|
except Exception:
|
|
pass
|
|
except Exception:
|
|
pass
|