test(scheduler): 新增「定时出单/通知书」能力检查脚本

沙箱 + 触发日注入，验证：
  电费       到点 -> 取已录入批次 -> 发送
  房租车位费 到点 -> 自动出单（批次 + PDF 通知书）-> 出单后推进 AC/AJ -> 发送
  反例：未到点不触发
发送用桩替代（不发真消息），真实 data/ 零改动。

Co-authored-by: CodeBuddy

tools/check_scheduler.py

+# -*- coding: utf-8 -*-
+"""
+定时出单 / 通知书 能力检查（2026-10-13）
+======================================
+回答：「是否可以定时产生账单和通知书？」
+
+做法：**全沙箱** —— 把 data/ 复制到临时目录并改写路径常量，真实 data/ 零改动；
+把「今天」注入到触发日，走真实的调度判定与出单渲染，仅**发送**用桩代替（不发真消息）。
+
+覆盖：
+  A. 排期判定：两张账单的 next_at 计算
+  B. 到点闸门：due_plans(触发日) 是否正确包含该账单
+  C. 房租车位费：到点 -> 自动出单（批次 + PDF 通知书）-> （桩）发送
+  D. 电费：到点 -> 取已录入批次 -> （桩）发送
+  E. 反例：未到点时不触发
+
+运行：python tools/check_scheduler.py
+"""
+import datetime as dt
+import io
+import os
+import shutil
+import sys
+import tempfile
+
+sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8")
+ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+sys.path.insert(0, ROOT)
+
+from rentos import config as C                       # noqa: E402
+from rentos import scheduler as SCH                  # noqa: E402
+from rentos import service, sender, store            # noqa: E402
+
+OK, FAIL = [], []
+
+
+def ck(name, cond, extra=""):
+    (OK if cond else FAIL).append(name)
+    print("  [%s] %s%s" % ("ok" if cond else "FAIL", name,
+                           ("  → %s" % extra) if extra != "" else ""))
+
+
+def sandbox():
+    tmp = tempfile.mkdtemp(prefix="rentos-sched-")
+    shutil.copytree(os.path.join(ROOT, "data"), os.path.join(tmp, "data"))
+    data = os.path.join(tmp, "data")
+    for name, val in {
+        "DATA_DIR": data,
+        "BATCH_DIR": os.path.join(data, "batches"),
+        "TABLES_DIR": os.path.join(data, "tables"),
+        "ATTACH_DIR": os.path.join(data, "attachments"),
+        "CONFIG_FILE": os.path.join(data, "config.json"),
+        "SEND_LOG_FILE": os.path.join(data, "send_log.json"),
+        "SESSION_FILE": os.path.join(data, "sessions.json"),
+        "UNIT_MAP_FILE": os.path.join(data, "unit_map.json"),
+        "UPLOAD_DIR": os.path.join(tmp, "uploads"),
+        "OUTPUT_DIR": os.path.join(tmp, "output", "elec"),
+        "OUTPUT_RENT_DIR": os.path.join(tmp, "output", "rent"),
+        "TABLE_EXPORT_DIR": os.path.join(tmp, "output", "tables"),
+        "BATCH_EXPORT_DIR": os.path.join(tmp, "output", "exports"),
+        "LOG_DIR": os.path.join(tmp, "logs"),
+    }.items():
+        setattr(C, name, val)
+    C.ensure_dirs()
+    # 模块级缓存常量，需手动重定向
+    SCH.STATE_FILE = os.path.join(data, "scheduler_state.json")
+    from rentos import directory
+    directory.SEED_FILE = os.path.join(data, "seed", "directory.json")
+    return tmp
+
+
+def main():
+    tmp = sandbox()
+    print("沙箱数据目录：%s（真实 data/ 未被触碰）\n" % tmp)
+
+    from rentos import plans, tables, rent_parser as RP
+
+    cfg = store.load_config()
+    pc = plans.all_of(cfg)
+    print("总开关 notify.enabled = %s | 电费计划 enabled=%s | 房租计划 enabled=%s\n"
+          % (cfg["notify"]["enabled"], pc["elec"]["enabled"], pc["rent_parking"]["enabled"]))
+
+    # ---------------------------------------------------------- A 排期
+    print("A. 排期判定（读真实台账 + 配置）")
+    st = SCH.status()
+    for it in st["plans"]:
+        ck("账单「%s」已启用并有 next_at 或状态说明" % it["name"],
+           bool(it["enabled"]), "state=%s next_at=%s | %s"
+           % (it["state"], it["next_at"] or "-", it["note"]))
+    rent_next = None
+    for it in st["plans"]:
+        if it["key"] == "rent_parking" and it["next_at"]:
+            rent_next = it["next_at"][:10]
+    elec_next = None
+    for it in st["plans"]:
+        if it["key"] == "elec" and it["next_at"]:
+            elec_next = it["next_at"][:10]
+    print()
+
+    # ---------------------------------------------------------- E 反例（基线，尚未出过单）
+    print("E. 反例：未到点不应触发（在出单前验证，避免「未发完批次重试」干扰）")
+    far = dt.datetime(2025, 1, 5, 9, 1)
+    ck("提前约两年不应触发任何账单", SCH.due_plans(far) == [],
+       "due_plans(%s) = %s" % (far.date(), SCH.due_plans(far) or "[]"))
+    ck("该日期下房租无可出单行", SCH.rent_state(pc["rent_parking"], far.date())["due_count"] == 0,
+       "due_count = %d" % SCH.rent_state(pc["rent_parking"], far.date())["due_count"])
+    print()
+
+    # ---------------------------------------------------------- C 房租：到点自动出单
+    print("C. 房租车位费：到点 -> 自动出单（账单 + PDF 通知书）-> 发送")
+    rs = SCH.rent_state(pc["rent_parking"])
+    trigger = rs["next_day"]
+    ck("台账能算出「下次推送日」（= 下一账单日 − 提前 5 天）", trigger is not None,
+       "下次推送日 = %s（今天 %s）" % (trigger, dt.date.today()))
+    if not trigger:
+        return finish(tmp)
+
+    # 把「今天」注入到触发日（只改日期来源，其余全走真实逻辑）
+    real_rent_state = SCH.rent_state
+    SCH.rent_state = lambda p, on_date=None: real_rent_state(p, on_date or trigger)
+    real_auto = service.auto_issue_rent
+    service.auto_issue_rent = lambda **kw: real_auto(
+        **{**kw, "on_date": kw.get("on_date") or trigger})
+
+    # 桩：不发真消息
+    sent = {}
+    sender.send_batch = lambda batch_id, indexes=None, cfg=None, progress=None, force=False: (
+        sent.update(batch_id=batch_id, n=len(indexes or [])) or
+        {"ok": len(indexes or []), "fail": 0, "skipped": 0,
+         "results": [], "chat_name": "（桩）"})
+
+    # B 到点闸门
+    print("B. 到点闸门 due_plans(%s)" % trigger)
+    fire_now = dt.datetime(trigger.year, trigger.month, trigger.day, 9, 1)
+    due = SCH.due_plans(fire_now)
+    ck("触发日 09:01 判定为「应触发」，含 rent_parking", "rent_parking" in due,
+       "due_plans = %s" % (due or "[]"))
+    before = {r["_id"]: (r.get("AC"), r.get("AJ"))
+              for r in tables.load_table(tables.RENT_KEY)["rows"]}
+    n_batches_before = len(store.list_batches("rent_parking"))
+
+    res = SCH.fire("定时", plan_key="rent_parking")
+
+    rows_after = {r["_id"]: (r.get("AC"), r.get("AJ"))
+                  for r in tables.load_table(tables.RENT_KEY)["rows"]}
+    new_ids = [k for k in rows_after if rows_after[k] != before.get(k)]
+    ck("自动出单产生了新批次", bool(res.get("plans")),
+       "批次 %s" % (res["plans"][0]["batch_id"] if res.get("plans") else "-"))
+
+    bid = (res["plans"][0]["batch_id"] if res.get("plans") else None)
+    batch = store.load_batch(bid) if bid else None
+    if batch:
+        docs = batch.get("documents") or []
+        ck("批次标记 auto_issued（防重复出单）", batch.get("auto_issued") is True,
+           batch.get("auto_issue_rule", ""))
+        ck("批次含 PDF 通知书单据", len(docs) > 0, "共 %d 份，例：%s"
+           % (len(docs), docs[0].get("file") if docs else "-"))
+        bad = []
+        for d in docs:
+            fp = d.get("abs_path") or ""
+            if not os.path.isfile(fp):
+                bad.append(d.get("file") + "(缺失)")
+                continue
+            with open(fp, "rb") as f:
+                if f.read(5) != b"%PDF-":
+                    bad.append(d.get("file") + "(非 PDF)")
+        ck("每份通知书都是真实 PDF 文件（%PDF- 头）", not bad,
+           ("问题件：%s" % bad) if bad else "%d 份全部有效" % len(docs))
+        ck("台账账期已推进 AC/AJ（同一行不会被重复自动出单）", len(new_ids) > 0,
+           "推进 %d 行，例 %s: %s -> %s"
+           % (len(new_ids), new_ids[0] if new_ids else "-",
+              before.get(new_ids[0]) if new_ids else "-",
+              rows_after.get(new_ids[0]) if new_ids else "-"))
+        ck("（桩）发送被调用，目标批次一致", sent.get("batch_id") == bid,
+           "已推送 %s 份" % sent.get("n"))
+    ck("出单前批次总数 -> 后：+1", len(store.list_batches("rent_parking"))
+       == n_batches_before + 1)
+    print()
+
+    # ---------------------------------------------------------- D 电费：到点发送已录入批次
+    print("D. 电费：到点 -> 取已录入批次 -> 发送")
+    b = SCH.pending_batch("elec")
+    if not b:
+        ck("存在待发送的电费批次", False, "当前无待发电费批次（需先录入抄表数）")
+    else:
+        target, next_at, due_s = SCH._trigger_at("elec", pc["elec"], b)
+        ck("电费批次可按缴费期限排期", target is not None,
+           "批次 %s 缴费期限 %s，提前 %s 天 -> 触发日 %s"
+           % (b["batch_id"], due_s, pc["elec"]["advance_days"], target))
+        d_now = dt.datetime(target.year, target.month, target.day, 9, 1)
+        ck("触发日 09:01 判定为「应触发」，含 elec", "elec" in SCH.due_plans(d_now),
+           "due_plans = %s" % (SCH.due_plans(d_now) or "[]"))
+        sent.clear()
+        res2 = SCH.fire("定时", plan_key="elec")
+        ck("（桩）电费批次被推送", sent.get("batch_id") == b["batch_id"],
+           "批次 %s 推送 %s 份" % (sent.get("batch_id"), sent.get("n")))
+    print()
+
+    return finish(tmp)
+
+
+def finish(tmp):
+    print("=" * 74)
+    print("通过 %d 项，失败 %d 项" % (len(OK), len(FAIL)))
+    if FAIL:
+        for f in FAIL:
+            print("  FAIL:", f)
+    print("沙箱目录（可删）：%s" % tmp)
+    return 0 if not FAIL else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())