diff --git a/paths.py b/paths.py index 07ba717..8d8c255 100644 --- a/paths.py +++ b/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") diff --git a/requirements.txt b/requirements.txt index 3a67faf..1652680 100644 --- a/requirements.txt +++ b/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 diff --git a/runtime.py b/runtime.py index c6c34fb..a15c9d8 100644 --- a/runtime.py +++ b/runtime.py @@ -429,25 +429,7 @@ def _site_undelivered_handler(site): return handler -def _run_all_compare_handler(): - """顶部「跑比对」：顺序跑 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 +# 「跑比对」= 纯离线比对（用 downloads/ 现有文件生成全站汇总；下载交由各站定时/手动）。 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), } diff --git a/server.py b/server.py index 9f298c9..7ff2556 100644 --- a/server.py +++ b/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): - date_offset: int +class ConfigRequest(BaseModel): + 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。""" - offsets = state_store.get_all_offsets() - return {site: offsets.get(site, 0) for site in ALL_SITES} + """各站配置：应到/实到偏移 + 每日定时（百世偏移恒 0）。""" + cfg = state_store.get_all_config() + return {site: cfg.get(site, _default_cfg()) for site in ALL_SITES} @app.put("/config/{site}") -def set_config(site: str, req: OffsetRequest): - if site == "百世": - raise HTTPException(status_code=400, detail="百世固定下载当天，不可配置偏移") - if site not in CONFIGURABLE_SITES: +def set_config(site: str, req: ConfigRequest): + if site not in ALL_SITES: raise HTTPException(status_code=400, detail=f"未知站点: {site}") - stored = state_store.set_offset(site, req.date_offset) - return {"site": site, "date_offset": stored} + cur = state_store.get_all_config().get(site, _default_cfg()) + # 应到/实到偏移（百世锁定当天） + 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}") diff --git a/site_anneng.py b/site_anneng.py index 486e10b..99e4d14 100644 --- a/site_anneng.py +++ b/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" diff --git a/site_shunxin.py b/site_shunxin.py index 43e25d7..53cca49 100644 --- a/site_shunxin.py +++ b/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") diff --git a/site_yunda.py b/site_yunda.py index 2ab75fc..f2478ab 100644 --- a/site_yunda.py +++ b/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 # 单日范围：起止同日 diff --git a/site_zto.py b/site_zto.py index 11dfd87..fa27cf5 100644 --- a/site_zto.py +++ b/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() diff --git a/state_store.py b/state_store.py index 90625b8..9906144 100644 --- a/state_store.py +++ b/state_store.py @@ -69,11 +69,33 @@ def init_db(): """) conn.execute(""" CREATE TABLE IF NOT EXISTS site_config ( - site TEXT PRIMARY KEY, - date_offset INTEGER NOT NULL DEFAULT 0, - updated_at TEXT + site TEXT PRIMARY KEY, + expected_offset INTEGER NOT NULL DEFAULT 0, + 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): - """读取单站下载日期偏移（0=今天，1=昨天…）；未配置返回 0。""" +def get_offset(site, kind="expected"): + """读取单站下载日期偏移（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): - """设置单站下载日期偏移，钳制到 [0, MAX_DATE_OFFSET]，返回实际写入值。""" +def set_offset(site, kind, 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 (?, ?, ?) " - "ON CONFLICT(site) DO UPDATE SET " - "date_offset=excluded.date_offset, updated_at=excluded.updated_at", + f"INSERT INTO site_config (site, {col}, updated_at) VALUES (?, ?, ?) " + f"ON CONFLICT(site) DO UPDATE SET {col}=excluded.{col}, " + f"updated_at=excluded.updated_at", (site, offset, _now()), ) conn.commit() return offset -def get_all_offsets(): - """返回 {site: date_offset}；未配置站点不在结果中（调用方按 0 兜底）。""" +def set_schedule(site, enabled, time_str): + """设置单站每日定时下载（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() - return {r[0]: int(r[1]) for r in rows} + rows = conn.execute( + "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 + } # ============================ 任务历史 ============================