feat: global harvest queue + configurable schedule planner

- harvest.sh: global queue lock (runlocks/.queue, mkdir-atomic, stale-PID
  steal, 180min wait cap) — a source whose time arrives while another is
  harvesting now waits (phase 排队中) and auto-starts when the previous
  run exits; runlock is created before queueing so the panel shows queued
  sources; manual panel trigger no longer refused (it queues)
- bin/plan_schedules.py + schedule-plan.json: frequency-driven planner —
  deadline/safety/min-gap/day-window/avoid all configurable; morning slots
  laid backward from deadline-safety with exact gaps (wrapping to previous
  evening when the roster exceeds the night), freq>1 sources spread evenly
  across the day window, global gap resolution; --dry-run supported
- 调度中心: ⚙ 按计划重排全部排程 button (POST /schedule-plan-apply)
- applied: 12 sources — 3×4/day (wsj/ft/digitimes-feed latest morning
  slots) + 9×1/day, 40-min spacing, last start 05:30 vs 06:00 deadline

bin/harvest.sh

@@ -60,18 +60,20 @@ cd "$WORK_DIR"
 source "$SCRIPT_DIR/result.sh"
 prepare_result_contract || exit 1
 
-# 登录预检:配置了凭据的源自动检查/登录其浏览器实例(失败不中断)
-nh_login_preflight "$SOURCE"
-
-# 任务结束(任何退出路径)回收浏览器实例:登录态在 profile 目录,关闭不丢;
-# 实例按站点共享 → 按端口引用计数,同站(同端口)最后一个任务退出才回收
+# 运行锁先建(排队中也算"运行中",面板可见);全局排队:同一时刻仅一个任务使用浏览器,
+# 到点时若忙则等待(排队中),前一任务结束自动开始。陈旧队列锁(PID 已死)自动窃取。
 _nh_lock_dir="$WORK_DIR/runlocks"
 mkdir -p "$_nh_lock_dir" 2>/dev/null || true
 _nh_port="${CDP_PORT_OVERRIDE:-$(nh_cdp_port "$(nh_login_site "$SOURCE")" 2>/dev/null)}"
 [ -n "$_nh_port" ] && touch "$_nh_lock_dir/port-${_nh_port}.$$.lock"
+_nh_queue_dir="$_nh_lock_dir/.queue"
+QUEUE_WAIT_MAX_MIN="${QUEUE_WAIT_MAX_MIN:-180}"
 _nh_quit_instance() {
   local skill_dir f base pid
   rm -f "$_nh_lock_dir/port-${_nh_port}.$$.lock" 2>/dev/null
+  # 释放全局排队锁
+  rm -f "$_nh_queue_dir/pid" 2>/dev/null
+  rmdir "$_nh_queue_dir" 2>/dev/null || true
   # 清理已被 kill -9 的任务留下的陈旧锁(锁名 port-<port>.<pid>.lock;
   # 文件名含两个点,须先去 .lock 后缀再取最后一段,否则会误把活锁当陈旧锁删掉)
   for f in "$_nh_lock_dir"/port-"${_nh_port}".*.lock; do
@@ -93,6 +95,27 @@ _nh_quit_instance() {
 }
 trap _nh_quit_instance EXIT
 
+echo "排队中" > "$LOG_DIR/${SOURCE}.phase"
+_queue_t0=$SECONDS
+until mkdir "$_nh_queue_dir" 2>/dev/null; do
+  # 陈旧队列锁自愈:持锁进程已死则窃取
+  _qp="$(cat "$_nh_queue_dir/pid" 2>/dev/null || true)"
+  if [ -n "$_qp" ] && ! kill -0 "$_qp" 2>/dev/null; then
+    rm -f "$_nh_queue_dir/pid" 2>/dev/null
+    rmdir "$_nh_queue_dir" 2>/dev/null || true
+    continue
+  fi
+  if [ $((SECONDS - _queue_t0)) -gt $((QUEUE_WAIT_MAX_MIN * 60)) ]; then
+    echo "[queue] 排队超过 ${QUEUE_WAIT_MAX_MIN} 分钟,放弃等待直接启动(可能并发)" >&2
+    break
+  fi
+  sleep 20
+done
+echo "$$" > "$_nh_queue_dir/pid" 2>/dev/null || true
+
+# 登录预检:配置了凭据的源自动检查/登录其浏览器实例(失败不中断)
+nh_login_preflight "$SOURCE"
+
 START_EPOCH=$(date +%s)
 
 # 阶段文件:面板实时读取的真实阶段(初始由本脚本写入,后续由采集会话按阶段更新)

bin/plan_schedules.py

+#!/usr/bin/env python3
+"""plan_schedules.py — 按频率计划自动生成并应用各源运行时间。
+
+配置:schedule-plan.json(与 default.conf 同目录),字段:
+  deadline      "HH:MM"  所有源的首次运行必须在此前完成(留 safety_min 缓冲)
+  safety_min    int      完成缓冲(默认 30 分钟:deadline - safety = 最后一个起始槽)
+  min_gap_min   int      相邻起始槽最小间隔(默认 40)
+  day_window    ["HH:MM","HH:MM"]  多次源白天附加运行的可分配区间(默认 06:30-21:30)
+  avoid         {"HH:MM": 分钟数}  避让窗口(如 digest 22:00 ±30),默认 {"22:00": 30}
+  frequencies   {"<source>": 次数}  每源每日次数;1 = 仅晨间;>1 = 晨间 + 白天均布
+
+算法:晨间槽从 (deadline - safety) 起,按 min_gap_min 逆序铺满全部源;
+多次源剩余次数在 day_window 内均布;最后全局按 min_gap_min 去冲突(后到顺延)。
+应用:逐源 nh_schedule_write + enable(幂等,可重跑);--dry-run 只打印不应用。
+
+用法:
+  python3 bin/plan_schedules.py [--dry-run] [--conf /path/to/schedule-plan.json]
+"""
+import json
+import subprocess
+import sys
+from datetime import datetime, timedelta
+from pathlib import Path
+
+BIN_DIR = Path(__file__).resolve().parent
+sys.path.insert(0, str(BIN_DIR))
+import platform_layer as pl  # noqa: E402
+
+
+def hm(s):
+    h, m = s.split(":")
+    return timedelta(hours=int(h), minutes=int(m))
+
+
+def fmt(td):
+    total = int(td.total_seconds()) // 60
+    total %= 24 * 60  # 跨午夜回绕(前晚槽位合法:22:10 早于次日 06:00 交付线)
+    return f"{total // 60:02d}:{total % 60:02d}"
+
+
+def load_conf(path):
+    default = {
+        "deadline": "06:00", "safety_min": 30, "min_gap_min": 40,
+        "day_window": ["06:30", "21:30"], "avoid": {"22:00": 30},
+        "frequencies": {},
+    }
+    if path.exists():
+        default.update(json.loads(path.read_text(encoding="utf-8")))
+    return default
+
+
+def plan(conf):
+    gap = timedelta(minutes=conf["min_gap_min"])
+    day = timedelta(days=1)
+    sources = sorted(conf["frequencies"], key=lambda s: -conf["frequencies"][s])
+    # 晨间槽:deadline-safety 为最后起始槽,逆序铺(多次源拿到最新鲜的槽);
+    # 负时间归一化到前晚(22:10 等合法:早于次日交付线)
+    last = hm(conf["deadline"]) - timedelta(minutes=conf["safety_min"])
+    morning = {s: (last - i * gap) % day for i, s in enumerate(sources)}
+    # 白天附加槽:day_window 内均布剩余次数
+    dw0, dw1 = hm(conf["day_window"][0]), hm(conf["day_window"][1])
+    result = {}
+    for s in sources:
+        freq = conf["frequencies"][s]
+        times = [morning[s]]
+        if freq > 1:
+            span = (dw1 - dw0) / freq
+            times += [dw0 + span * (k + 1) for k in range(freq - 1)]
+        result[s] = times
+    avoid = [(hm(k) - timedelta(minutes=v), hm(k) + timedelta(minutes=v))
+             for k, v in conf.get("avoid", {}).items()]
+    # 全局去冲突+避让(按时刻排序,冲突/避让者顺延 gap;迭代收敛)
+    for _ in range(4):
+        slots = sorted((t, s) for s, ts in result.items() for t in ts)
+        new = {s: [] for s in result}
+        prev = None
+        changed = False
+        for t, s in slots:
+            orig = t
+            if prev is not None and t - prev < gap:
+                t = prev + gap
+            for a0, a1 in avoid:
+                if a0 <= t <= a1:
+                    t = a1
+                    if prev is not None and t - prev < gap:
+                        t = prev + gap
+            if t != orig:
+                changed = True
+            new[s].append(t)
+            prev = t
+        result = {s: sorted(ts) for s, ts in new.items()}
+        if not changed:
+            break
+    return {s: sorted(fmt(t) for t in ts) for s, ts in result.items()}
+
+
+def apply(plan_map):
+    lines = []
+    for s, times in plan_map.items():
+        joined = ",".join(times)
+        rc, out, err = pl.call("nh_schedule_write", s, joined, timeout=60)
+        if rc == 0:
+            rc2, _, err2 = pl.call("nh_schedule_enable", s, timeout=60)
+            if not pl.skill_sources().get(s, {}).get("enabled"):
+                pl.skill_set_enabled(s, True)
+            lines.append(f"✅ {s:24} {joined}")
+        else:
+            lines.append(f"❌ {s:24} {joined} 写入失败: {err or out}")
+    return lines
+
+
+def main():
+    dry = "--dry-run" in sys.argv
+    from pathlib import Path as _P
+    conf_path = _P(pl.default_conf_dir(pl.default_work_dir())) / "schedule-plan.json"
+    if "--conf" in sys.argv:
+        conf_path = Path(sys.argv[sys.argv.index("--conf") + 1])
+    conf = load_conf(Path(conf_path))
+    if not conf["frequencies"]:
+        print("配置中无 frequencies,退出"); return 1
+    p = plan(conf)
+    print(f"排程计划(deadline={conf['deadline']} 间隔≥{conf['min_gap_min']}min "
+          f"safety={conf['safety_min']}min){' [DRY-RUN]' if dry else ''}")
+    order = sorted(p, key=lambda s: p[s][0])
+    for s in order:
+        n = len(p[s])
+        print(f"  {s:24} {'/'.join(p[s]):32} 每日{n}次")
+    if not dry:
+        print("\n应用:")
+        for line in apply(p):
+            print(" ", line)
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

web/app.py

@@ -912,16 +912,10 @@ def edit_source(name):
 def run_source(name):
     if not re.match(r"^[a-zA-Z0-9_-]+$", name):
         abort(404)
-    active = _active_harvests()
-    if active:
-        flash(
-            f"已有采集任务在运行({('、'.join(active))}),为避免并发冲突已拒绝触发,请等其完成后再试。",
-            "error",
-        )
-        return redirect(url_for("source_detail", name=name))
+    # 并发已由 harvest.sh 全局排队锁串行化:忙时触发会自动排队,前者结束即开始
     rc, stdout, stderr = pl.service_trigger(name)
     if rc == 0:
-        flash(f"已触发 {name} 抓取", "success")
+        flash(f"已触发 {name} 抓取(若正忙将自动排队)", "success")
     else:
         flash(f"触发失败: {stderr}", "error")
     return redirect(url_for("source_detail", name=name))
@@ -1374,6 +1368,20 @@ def schedule_time_set():
     return redirect(url_for("schedule_board"))
 
 
+@app.route("/schedule-plan-apply", methods=["POST"])
+@require_auth
+def schedule_plan_apply():
+    """按 schedule-plan.json 重排全部源时间(频率/交付线/间隔可配置)。"""
+    r = subprocess.run(
+        ["python3", str(Path(__file__).resolve().parent.parent / "bin" / "plan_schedules.py")],
+        capture_output=True, text=True, timeout=300)
+    out = (r.stdout or "")[-3000:]
+    flash(f"排程计划已应用\n{out}" if r.returncode == 0
+          else f"计划失败 (exit {r.returncode})\n{out}\n{r.stderr[-800:]}",
+          "success" if r.returncode == 0 else "error")
+    return redirect(url_for("schedule_board"))
+
+
 @app.route("/schedule-delete/<name>", methods=["POST"])
 @require_auth
 def schedule_delete(name):

web/templates/schedules.html

@@ -28,6 +28,11 @@
   每日汇总 digest:{{ digest_times|join(', ') if digest_times else '未排' }}{% if digest_next %},下次 {{ digest_next.strftime('%m-%d %H:%M') }}{% endif %}。
   时间格式 HH:MM,多个用逗号分隔(如 05:00,17:00 = 每日两次);删除仅移除调度,技能源与历史保留。
 </p>
+<form method="POST" action="{{ url_for('schedule_plan_apply') }}" style="margin:6px 0 12px"
+      onsubmit="return confirm('按 schedule-plan.json(频率/交付线/间隔)重排全部源时间?将覆盖现有排程。')">
+  <button type="submit" class="btn btn-primary">⚙ 按计划重排全部排程</button>
+  <span style="color:#888;font-size:12px;margin-left:8px">配置:~/NewsHarvester/etc/news-harvester/schedule-plan.json(频率/交付线/间隔可改,改完点此按钮生效)</span>
+</form>
 
 {% with messages = get_flashed_messages() %}
   {% if messages %}{% for m in messages %}<div class="flash">{{ m }}</div>{% endfor %}{% endif %}