通用爬虫系统 v1.0.0: 批量/定时/自动爬取 + Web管理界面 + 图片/通知/热更新

scheduler.py

@@ -0,0 +1,120 @@
+# -*- 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)
+            except Exception:
+                pass