定时下载(APScheduler) + 跑比对纯离线 + 偏移拆 expected/actual + 报告下载

定时下载：
- state_store：site_config 加 schedule_enabled/schedule_time（旧库迁移）+ set_schedule/get_all_config
- server：APScheduler BackgroundScheduler，启动按配置注册各站 cron job、改配置重排；job 投 undelivered 任务到队列（错过跳过）；/config 扩展 schedule
- runtime：("__compare__","compare") 改纯离线 expected_undelivered.main()（不下载）
- requirements：加 apscheduler

偏移拆分 + 报告下载：
- state_store：date_offset 拆 expected_offset/actual_offset（迁移）+ get_offset(site,kind)/set_offset(site,kind,n)
- 4 站 actual impl 改读 actual 偏移（expected 走默认 kind）
- server：/config 用 expected_offset/actual_offset；新增 GET /report（服务 output/应到未到数据.xlsx）
- paths：加 OUTPUT_DIR

Co-Authored-By: Claude <noreply@anthropic.com>

paths.py

@@ -10,6 +10,7 @@ BASE_DIR = os.path.dirname(os.path.abspath(__file__))
 
 # 统一的下载 / 输出目录
 DOWNLOAD_DIR = os.path.join(BASE_DIR, "downloads")
+OUTPUT_DIR = os.path.join(BASE_DIR, "output")
 
 # 统一的配置文件路径（注意：config.yaml 需与本项目脚本放在同一目录下）
 CONFIG_PATH = os.path.join(BASE_DIR, "config.yaml")

requirements.txt

@@ -5,3 +5,4 @@ PyYAML>=6.0
 websocket-client>=1.0.0
 fastapi>=0.110.0
 uvicorn>=0.27.0
+apscheduler>=3.10.0

runtime.py

@@ -429,25 +429,7 @@ def _site_undelivered_handler(site):
     return handler
 
 
-def _run_all_compare_handler():
+# 「跑比对」= 纯离线比对（用 downloads/ 现有文件生成全站汇总；下载交由各站定时/手动）。
-    """顶部「跑比对」：顺序跑 5 站未到（4 站 expected+actual→比对；百世直供），
-    记录本次成功站 → build_full_report。失败/未登录站保留行、无数据，不影响他站。"""
-
-    def handler(ctx):
-        include = set()
-        for site in ("顺心", "中通", "韵达", "安能", "百世"):
-            if not probe_site_login(site, ctx.pages_map):
-                print(f">> [跑比对] {site} 未登录，跳过其统计")
-                continue
-            h = TASK_HANDLERS.get((site, "undelivered"))
-            if h and h(ctx):
-                include.add(site)
-            else:
-                print(f">> [跑比对] {site} 未到数据获取失败，跳过其统计")
-        expected_undelivered.build_full_report(include)
-        return True
-
-    return handler
 
 
 TASK_HANDLERS = {
@@ -466,7 +448,7 @@ TASK_HANDLERS = {
     ("安能", "expected"): lambda ctx: site_anneng.anneng_expected_download(),
     ("安能", "actual"): lambda ctx: site_anneng.anneng_actual_download(),
     ("安能", "undelivered"): _site_undelivered_handler("安能"),
-    ("__compare__", "compare"): _run_all_compare_handler(),
+    ("__compare__", "compare"): lambda ctx: (expected_undelivered.main() or True),
 }
 
 

server.py

@@ -17,13 +17,16 @@ import queue
 import threading
 import time
 from contextlib import asynccontextmanager
+from typing import Optional
 
 import uvicorn
+from apscheduler.schedulers.background import BackgroundScheduler
+from apscheduler.triggers.cron import CronTrigger
 from fastapi import FastAPI, HTTPException
 from fastapi.responses import FileResponse
 from pydantic import BaseModel
 
-from paths import DOWNLOAD_DIR
+from paths import DOWNLOAD_DIR, OUTPUT_DIR
 import state_store
 from runtime import (
     HEARTBEAT_INTERVAL,
@@ -48,6 +51,9 @@ worker_state = {
     "error": None,  # worker 启动失败原因
 }
 
+# 每日定时下载调度器（进程内；job 只往 task_queue 投任务，不碰 Playwright）
+scheduler = BackgroundScheduler(daemon=True)
+
 
 def _worker_loop():
     """worker 线程：启动 Playwright + 等就绪 + 任务循环（执行任务 + 心跳）。"""
@@ -86,14 +92,54 @@ def _worker_loop():
     print(">> [worker] 已退出。")
 
 
+def _enqueue_undelivered(site):
+    """定时 job：把该站 undelivered 任务投到队列（worker 串行处理；本线程不碰 Playwright）。"""
+    try:
+        tid = state_store.create_task(site, "undelivered")
+        task_queue.put((tid, {"site": site, "kind": "undelivered"}))
+        print(f">> [定时] 投递 {site}/undelivered 任务 #{tid}")
+    except Exception as e:
+        print(f">> [定时] 投递 {site} 失败: {e}")
+
+
+def _reschedule_site(site):
+    """按持久化配置（重新）注册或取消该站的每日定时 job。"""
+    job_id = f"site_{site}"
+    try:
+        scheduler.remove_job(job_id)
+    except Exception:
+        pass
+    cfg = state_store.get_all_config().get(site)
+    if not cfg or not cfg.get("schedule_enabled") or not cfg.get("schedule_time"):
+        return
+    try:
+        hh, mm = cfg["schedule_time"].split(":")
+        scheduler.add_job(
+            _enqueue_undelivered,
+            CronTrigger(hour=int(hh), minute=int(mm)),
+            args=[site],
+            id=job_id,
+            replace_existing=True,
+        )
+        print(f">> [定时] 已注册 {site} 每日 {cfg['schedule_time']} 下载")
+    except Exception as e:
+        print(f">> [定时] 注册 {site} 失败: {e}")
+
+
 @asynccontextmanager
 async def lifespan(_app):
     """服务启停：起 worker 线程 / 通知 worker 停。"""
     state_store.init_db()  # 先建表/迁移状态库，确保早于 worker 就绪的 /api/status 可用
+    for site in ALL_SITES:  # 按持久化配置注册各站每日定时 job
+        _reschedule_site(site)
+    scheduler.start()
+    print(">> [定时] 调度器已启动")
     t = threading.Thread(target=_worker_loop, daemon=True)
     worker_state["thread"] = t
     t.start()
     yield
+    scheduler.shutdown(wait=False)
+    print(">> [定时] 调度器已停止")
     worker_state["stop"] = True
     t.join(timeout=10)
 
@@ -139,25 +185,70 @@ def get_status():
     }
 
 
-class OffsetRequest(BaseModel):
+class ConfigRequest(BaseModel):
-    date_offset: int
+    expected_offset: Optional[int] = None
+    actual_offset: Optional[int] = None
+    schedule_enabled: Optional[bool] = None
+    schedule_time: Optional[str] = None
+
+
+def _default_cfg():
+    return {
+        "expected_offset": 0,
+        "actual_offset": 0,
+        "schedule_enabled": False,
+        "schedule_time": "",
+    }
 
 
 @app.get("/config")
 def get_config():
-    """各站下载日期偏移（0=今天，1=昨天…）；百世恒 0。"""
+    """各站配置：应到/实到偏移 + 每日定时（百世偏移恒 0）。"""
-    offsets = state_store.get_all_offsets()
+    cfg = state_store.get_all_config()
-    return {site: offsets.get(site, 0) for site in ALL_SITES}
+    return {site: cfg.get(site, _default_cfg()) for site in ALL_SITES}
 
 
 @app.put("/config/{site}")
-def set_config(site: str, req: OffsetRequest):
+def set_config(site: str, req: ConfigRequest):
-    if site == "百世":
+    if site not in ALL_SITES:
-        raise HTTPException(status_code=400, detail="百世固定下载当天，不可配置偏移")
-    if site not in CONFIGURABLE_SITES:
         raise HTTPException(status_code=400, detail=f"未知站点: {site}")
-    stored = state_store.set_offset(site, req.date_offset)
+    cur = state_store.get_all_config().get(site, _default_cfg())
-    return {"site": site, "date_offset": stored}
+    # 应到/实到偏移（百世锁定当天）
+    for kind, val in (
+        ("expected", req.expected_offset),
+        ("actual", req.actual_offset),
+    ):
+        if val is not None:
+            if site == "百世":
+                raise HTTPException(
+                    status_code=400, detail="百世固定下载当天，不可配置偏移"
+                )
+            state_store.set_offset(site, kind, val)
+    # 每日定时（所有站可配，含百世）
+    if req.schedule_enabled is not None or req.schedule_time is not None:
+        enabled = (
+            req.schedule_enabled
+            if req.schedule_enabled is not None
+            else cur["schedule_enabled"]
+        )
+        time_str = (
+            req.schedule_time if req.schedule_time is not None else cur["schedule_time"]
+        )
+        state_store.set_schedule(site, enabled, time_str)
+        _reschedule_site(site)
+    return state_store.get_all_config().get(site, _default_cfg())
+
+
+REPORT_FILE = "应到未到数据.xlsx"
+
+
+@app.get("/report")
+def download_report():
+    """下载 output/应到未到数据.xlsx（跑比对生成；未生成 404）。"""
+    path = os.path.join(OUTPUT_DIR, REPORT_FILE)
+    if not os.path.isfile(path):
+        raise HTTPException(status_code=404, detail="报告尚未生成，请先跑比对")
+    return FileResponse(path, filename=REPORT_FILE)
 
 
 @app.get("/data/{filename}")

site_anneng.py

@@ -1173,7 +1173,7 @@ def anneng_actual_download_impl():
     # 清理上次的最终文件，避免本次无数据/失败时残留旧数据误导比对
     _remove_if_exists(os.path.join(download_dir, ACTUAL_FINAL_FILENAME))
 
-    offset = state_store.get_offset("安能")
+    offset = state_store.get_offset("安能", "actual")
     today = datetime.now()
     target = today - timedelta(days=offset)
     start_str = f"{target.year}/{target.month:02d}/{target.day:02d} 00:00:00"

site_shunxin.py

@@ -540,7 +540,7 @@ def shunxin_actual_download_impl(page, out_tag=""):
         print("✅ 卸车扫描记录界面加载完毕")
 
         # 2. 读取服务端日期偏移（0=今天，1=昨天…），单日范围：起止同日
-        offset = state_store.get_offset("顺心")
+        offset = state_store.get_offset("顺心", "actual")
         today = datetime.now()
         target = today - timedelta(days=offset)
         target_str = target.strftime("%Y-%m-%d")

site_yunda.py

@@ -414,7 +414,7 @@ def yunda_actual_download_impl(page):
         ).first.wait_for(state="visible", timeout=15000)
         print("✅ 扫描记录查询页面已初始化")
 
-        offset = state_store.get_offset("韵达")
+        offset = state_store.get_offset("韵达", "actual")
         today = datetime.now()
         target = today - timedelta(days=offset)
         start_date = target  # 单日范围：起止同日

site_zto.py

@@ -364,7 +364,7 @@ def zto_actual_download_impl(page):
         arr_frame.locator("#daterange").wait_for(state="attached", timeout=15000)
 
         # 2. 读取服务端日期偏移（0=今天，1=昨天…），单日：起止同日
-        offset = state_store.get_offset("中通")
+        offset = state_store.get_offset("中通", "actual")
 
         print(f">> 正在设定查询日期: 偏移 {offset}（0=今天）...")
         arr_frame.locator("#daterange").click()

state_store.py

@@ -69,11 +69,33 @@ def init_db():
             """)
         conn.execute("""
             CREATE TABLE IF NOT EXISTS site_config (
-                site        TEXT PRIMARY KEY,
+                site             TEXT PRIMARY KEY,
-                date_offset INTEGER NOT NULL DEFAULT 0,
+                expected_offset  INTEGER NOT NULL DEFAULT 0,
-                updated_at  TEXT
+                actual_offset    INTEGER NOT NULL DEFAULT 0,
+                schedule_enabled INTEGER NOT NULL DEFAULT 0,
+                schedule_time    TEXT NOT NULL DEFAULT '',
+                updated_at       TEXT
             )
             """)
+        # 旧库迁移：补 schedule 两列 + expected/actual 偏移（新库已含；重复添加抛错忽略）
+        for _col, _typedef in [
+            ("schedule_enabled", "INTEGER NOT NULL DEFAULT 0"),
+            ("schedule_time", "TEXT NOT NULL DEFAULT ''"),
+            ("expected_offset", "INTEGER NOT NULL DEFAULT 0"),
+            ("actual_offset", "INTEGER NOT NULL DEFAULT 0"),
+        ]:
+            try:
+                conn.execute(f"ALTER TABLE site_config ADD COLUMN {_col} {_typedef}")
+            except sqlite3.OperationalError:
+                pass
+        # 旧库若有 date_offset 列，把值搬到 expected/actual（一次性；新库无此列则跳过）
+        try:
+            conn.execute(
+                "UPDATE site_config SET expected_offset=date_offset, actual_offset=date_offset "
+                "WHERE expected_offset=0 AND date_offset IS NOT NULL AND date_offset>0"
+            )
+        except sqlite3.OperationalError:
+            pass
         conn.commit()
 
 
@@ -197,38 +219,67 @@ def get_all_status():
 MAX_DATE_OFFSET = 30  # 0=今天，最大回溯 30 天
 
 
-def get_offset(site):
+def get_offset(site, kind="expected"):
-    """读取单站下载日期偏移（0=今天，1=昨天…）；未配置返回 0。"""
+    """读取单站下载日期偏移（kind: 'expected'/'actual'；0=今天，1=昨天…）；未配置返回 0。"""
+    col = "expected_offset" if kind == "expected" else "actual_offset"
     if not os.path.exists(STATE_DB_PATH):
         return 0
     with sqlite3.connect(STATE_DB_PATH) as conn:
         row = conn.execute(
-            "SELECT date_offset FROM site_config WHERE site=?", (site,)
+            f"SELECT {col} FROM site_config WHERE site=?", (site,)
         ).fetchone()
     return int(row[0]) if row else 0
 
 
-def set_offset(site, offset):
+def set_offset(site, kind, offset):
-    """设置单站下载日期偏移，钳制到 [0, MAX_DATE_OFFSET]，返回实际写入值。"""
+    """设置单站下载日期偏移（kind: 'expected'/'actual'），钳制到 [0, MAX_DATE_OFFSET]。"""
+    col = "expected_offset" if kind == "expected" else "actual_offset"
     offset = max(0, min(MAX_DATE_OFFSET, int(offset)))
     with sqlite3.connect(STATE_DB_PATH) as conn:
         conn.execute(
-            "INSERT INTO site_config (site, date_offset, updated_at) VALUES (?, ?, ?) "
+            f"INSERT INTO site_config (site, {col}, updated_at) VALUES (?, ?, ?) "
-            "ON CONFLICT(site) DO UPDATE SET "
+            f"ON CONFLICT(site) DO UPDATE SET {col}=excluded.{col}, "
-            "date_offset=excluded.date_offset, updated_at=excluded.updated_at",
+            f"updated_at=excluded.updated_at",
             (site, offset, _now()),
         )
         conn.commit()
     return offset
 
 
-def get_all_offsets():
+def set_schedule(site, enabled, time_str):
-    """返回 {site: date_offset}；未配置站点不在结果中（调用方按 0 兜底）。"""
+    """设置单站每日定时下载（enabled: bool；time_str: 'HH:MM' 或 ''）。"""
+    enabled_int = 1 if enabled else 0
+    with sqlite3.connect(STATE_DB_PATH) as conn:
+        conn.execute(
+            "INSERT INTO site_config (site, schedule_enabled, schedule_time, updated_at) "
+            "VALUES (?, ?, ?, ?) "
+            "ON CONFLICT(site) DO UPDATE SET "
+            "schedule_enabled=excluded.schedule_enabled, "
+            "schedule_time=excluded.schedule_time, updated_at=excluded.updated_at",
+            (site, enabled_int, time_str or "", _now()),
+        )
+        conn.commit()
+    return bool(enabled_int), (time_str or "")
+
+
+def get_all_config():
+    """返回 {site: {expected_offset, actual_offset, schedule_enabled, schedule_time}}。"""
     if not os.path.exists(STATE_DB_PATH):
         return {}
     with sqlite3.connect(STATE_DB_PATH) as conn:
-        rows = conn.execute("SELECT site, date_offset FROM site_config").fetchall()
+        rows = conn.execute(
-    return {r[0]: int(r[1]) for r in rows}
+            "SELECT site, expected_offset, actual_offset, schedule_enabled, schedule_time "
+            "FROM site_config"
+        ).fetchall()
+    return {
+        r[0]: {
+            "expected_offset": int(r[1]),
+            "actual_offset": int(r[2]),
+            "schedule_enabled": bool(r[3]),
+            "schedule_time": r[4] or "",
+        }
+        for r in rows
+    }
 
 
 # ============================ 任务历史 ============================