From eaf56c1c9e5241315493d1b0a91b050cbb4b89a5 Mon Sep 17 00:00:00 2001 From: Misaka Date: Fri, 31 Jul 2026 23:47:23 +0800 Subject: [PATCH] refactor(status): derive readiness from ingest_state; drop Excel probe & /data MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 状态盘就绪态 {kind}_ready 改为从持久的 ingest_state 派生（DB 真相、重启不丢）， 不再由心跳读 downloads/ Excel、也不在启动时重置： - 心跳 _fresh_ingest/_ready_flags/_apply_ready：expected/actual_ready=该类今天入库成功； undelivered_ready=百世原生(今天入库) / 4 站派生(expected ∧ actual)。 - _persist_to_db 入库后调 _refresh_ready 立即派生（省 30s 心跳等待，与心跳同源）； 4 站 undelivered 连入 expected+actual，故 ingest_state 补记 expected/actual/undelivered 三行。 - _record_business_date 只写 business_date，不再碰 ready。 - 删 run_heartbeat 的 Excel 探测循环、DATA_FILENAMES、probe_data_file、遗物清理。 - state_store 增 set_business_date / set_ready（仅写单字段，不碰彼此）。 - 删 GET /data/{filename}（前端不再下载原始中转 Excel）；/report 保留。 原则：Excel 只作「站点下载→入库」中转，状态/比对一律走 DB。 真机+单元验证：重启后心跳派生（百世今日入库→立即绿、重启不掉灰）；韵达 expected-only 时 undelivered 不亮、补 actual 后亮（派生）；百世原生；_ready_flags 7 例边界全过。 Co-Authored-By: Claude --- inbound_verify/cli/server.py | 16 +---- inbound_verify/runtime.py | 112 ++++++++++++++++++---------------- inbound_verify/state_store.py | 19 ++++++ 3 files changed, 79 insertions(+), 68 deletions(-) diff --git a/inbound_verify/cli/server.py b/inbound_verify/cli/server.py index c927949..821de4e 100644 --- a/inbound_verify/cli/server.py +++ b/inbound_verify/cli/server.py @@ -27,7 +27,7 @@ from fastapi import FastAPI, HTTPException from fastapi.responses import FileResponse from pydantic import BaseModel -from inbound_verify.paths import DOWNLOAD_DIR, OUTPUT_DIR +from inbound_verify.paths import OUTPUT_DIR from inbound_verify import state_store from inbound_verify.runtime import ( HEARTBEAT_INTERVAL, @@ -417,20 +417,6 @@ def download_report(): return FileResponse(path, filename=REPORT_FILE) -@app.get("/data/{filename}") -def download_data(filename: str): - """下载 downloads/ 下的数据文件（防路径穿越）。""" - if not filename or "/" in filename or "\\" in filename or ".." in filename: - raise HTTPException(status_code=400, detail="非法文件名") - path = os.path.join(DOWNLOAD_DIR, filename) - # 双重校验：解析后绝对路径仍在 DOWNLOAD_DIR 内 - if not os.path.abspath(path).startswith(os.path.abspath(DOWNLOAD_DIR) + os.sep): - raise HTTPException(status_code=400, detail="非法路径") - if not os.path.isfile(path): - raise HTTPException(status_code=404, detail="文件不存在") - return FileResponse(path, filename=filename) - - def main(): """服务模式入口。传字符串导入路径（规范写法；不开 reload/workers 时进程内 import，行为等价）。""" uvicorn.run("inbound_verify.cli.server:app", host="0.0.0.0", port=8000) diff --git a/inbound_verify/runtime.py b/inbound_verify/runtime.py index 867ca4b..ace1f5d 100644 --- a/inbound_verify/runtime.py +++ b/inbound_verify/runtime.py @@ -21,8 +21,7 @@ from datetime import datetime, timedelta import yaml from playwright.sync_api import sync_playwright -from inbound_verify.paths import DOWNLOAD_DIR, CONFIG_PATH -from inbound_verify.domain import SITE_UNDELIVERED_FILE +from inbound_verify.paths import CONFIG_PATH from inbound_verify import state_store from inbound_verify.sites import shunxin, baishi, zto, yunda, anneng @@ -50,31 +49,6 @@ APP_SITES = {"安能"} # 心跳间隔（秒） HEARTBEAT_INTERVAL = 30 -# 各站最终数据文件名（探测"数据是否已跑出来"）；百世为单流程 -DATA_FILENAMES = { - "顺心": { - "expected": "顺心-应到货物数据.xlsx", - "actual": "顺心-实到货物数据.xlsx", - "undelivered": "顺心-未到数据.xlsx", - }, - "中通": { - "expected": "中通-应到货物数据.xlsx", - "actual": "中通-实到货物数据.xlsx", - "undelivered": "中通-未到数据.xlsx", - }, - "韵达": { - "expected": "韵达-应到货物数据.xlsx", - "actual": "韵达-实到货物数据.xlsx", - "undelivered": "韵达-未到数据.xlsx", - }, - "安能": { - "expected": "安能-应到货物数据.xlsx", - "actual": "安能-实到货物数据.xlsx", - "undelivered": "安能-未到数据.xlsx", - }, - "百世": {"expected": "", "actual": "", "undelivered": "百世-应到未到货物数据.xlsx"}, -} - # ============================ 安能启动（CDP）============================ @@ -150,19 +124,6 @@ def probe_site_login(site_name, pages_map): return False -def probe_data_file(site_name, kind): - """探测单站应到/实到数据文件是否存在且为今天。返回 (is_today, mtime_str)。""" - fname = DATA_FILENAMES.get(site_name, {}).get(kind, "") - if not fname: - return (False, "") - path = os.path.join(DOWNLOAD_DIR, fname) - if not os.path.exists(path): - return (False, "") - dt = datetime.fromtimestamp(os.path.getmtime(path)) - is_today = dt.date() == datetime.now().date() - return (is_today, dt.strftime("%Y-%m-%d %H:%M:%S")) - - # ============================ 运行上下文 ============================ @@ -488,9 +449,6 @@ def _site_undelivered_handler(site): else False ) if not exp_ok or not act_ok: - stale = os.path.join(DOWNLOAD_DIR, SITE_UNDELIVERED_FILE.format(name=site)) - if os.path.exists(stale): - os.remove(stale) return False # ── 先入库再比对（修复时序：比对须读到本次下载的数据， @@ -591,11 +549,10 @@ def _record_business_date(site, kind, date=None): undelivered：百世直供（恒当天）写 undelivered；4 站未到由 _site_undelivered_handler 内部连带下了 expected+actual（不经 dispatch，无业务日期写入），故此处一并补写 expected/actual/undelivered 三列——actual 用 actual 偏移、未到跟随 expected 偏移。 - 顺带置 ready=True，让前端不必等心跳即可反映下载成功；写入失败仅告警、不影响任务判定。""" + 只写业务日期；ready 语义已移交「入库成功」（_persist_to_db 置位），此处不再碰 ready。""" if site == "__compare__": return today = datetime.now().date() - now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") def _write(k, biz_or_off): # biz_or_off: int=偏移（today−off）；str=已确定业务日期（date） @@ -605,9 +562,7 @@ def _record_business_date(site, kind, date=None): else biz_or_off ) try: - state_store.set_data_state( - site, k, ready=True, generated_at=now, business_date=biz - ) + state_store.set_business_date(site, k, biz) except Exception as e: print(f">> [状态] 写业务日期失败 {site}/{k}: {e}") @@ -626,6 +581,45 @@ def _record_business_date(site, kind, date=None): _write("undelivered", date if date else off("expected")) +def _fresh_ingest(ingest, site, kind): + """ingest_state[site][kind] 是否「今天入库成功」——ready 派生的 DB 真相来源。""" + v = ingest.get(site, {}).get(kind) + if not v or not v.get("ok"): + return False + return (v.get("ingested_at") or "")[:10] == datetime.now().strftime("%Y-%m-%d") + + +def _ready_flags(ingest, site): + """从 ingest_state 派生单站三就绪态（ready = DB 入库真相）。 + + expected/actual = 该类今天入库成功；undelivered = 百世（原生未到）今天入库成功 + / 4 站（派生未到）= expected ∧ actual。 + """ + if site == "百世": + return { + "expected": False, + "actual": False, + "undelivered": _fresh_ingest(ingest, site, "undelivered"), + } + exp = _fresh_ingest(ingest, site, "expected") + act = _fresh_ingest(ingest, site, "actual") + return {"expected": exp, "actual": act, "undelivered": exp and act} + + +def _apply_ready(site, flags): + """写入单站就绪态（仅 ready，不碰 business_date/generated_at）。失败仅告警。""" + for k, rdy in flags.items(): + try: + state_store.set_ready(site, k, rdy) + except Exception as e: + print(f">> [状态] 置就绪态失败 {site}/{k}: {e}") + + +def _refresh_ready(site): + """入库后立即据最新 ingest_state 派生并写入该站就绪态（省 30s 心跳等待，与心跳同源）。""" + _apply_ready(site, _ready_flags(state_store.get_all_ingest_state(), site)) + + def _persist_to_db(site, kind): """下载成功后把本次数据入库 PostgreSQL（尽力而为，绝不外抛，不影响任务判定）。 - __compare__ 无源数据，跳过。 @@ -645,7 +639,16 @@ def _persist_to_db(site, kind): print(">> [入库] 已关闭 (auto_ingest=false)，跳过") return count = store.ingest_task(site, kind) - state_store.set_ingest_state(site, kind, ok=True, count=count) + # 4 站 undelivered 连带入了 expected+actual：按实际入库的类补记 ingest_state， + # 否则心跳派生 ready（expected ∧ actual → undelivered）会读到陈旧值。 + logged = ( + ["expected", "actual", "undelivered"] + if kind == "undelivered" and site != "百世" + else [kind] + ) + for k in logged: + state_store.set_ingest_state(site, k, ok=True, count=count) + _refresh_ready(site) # 入库成功 → 立即据 ingest_state 派生就绪态（与心跳同源） print(f">> [入库] {site}/{kind} 成功，{count} 条") except Exception as e: print(f">> [warn] 入库失败 {site}/{kind}: {e}") @@ -689,11 +692,16 @@ def dispatch_task(ctx, task_spec): def run_heartbeat(ctx): - """一轮心跳：探测各站登录态 + 数据文件，写状态库；登录态变化时提示。 + """一轮心跳：探测各站登录态 + 据 ingest_state 派生数据就绪态；登录态变化时提示。 + ready 不再读 Excel、也不在启动时重置，而是每轮从持久的 ingest_state 派生（重启不丢）： + expected/actual_ready = 该类今天入库成功； + undelivered_ready = 百世（原生未到）今天入库成功 / 4 站（派生未到）= expected ∧ actual。 + _refresh_ready 在入库瞬间即据 ingest_state 派生（省 30s 等待），心跳同源复核。 只在 Playwright 所属线程调用。 """ prev = state_store.get_all_status() + ingest = state_store.get_all_ingest_state() for site_name in ctx.sites_to_watch: logged_in = probe_site_login(site_name, ctx.pages_map) prev_login = prev.get(site_name, {}).get("login_state") @@ -701,6 +709,4 @@ def run_heartbeat(ctx): now_login = state_store.LOGIN_IN if logged_in else state_store.LOGIN_OUT if prev_login and prev_login not in (now_login, state_store.LOGIN_UNKNOWN): print(f"\n ⚠️【{site_name}】登录态变化: {prev_login} → {now_login}") - for kind in ("expected", "actual", "undelivered"): - ready, gen_at = probe_data_file(site_name, kind) - state_store.set_data_state(site_name, kind, ready, gen_at) + _apply_ready(site_name, _ready_flags(ingest, site_name)) diff --git a/inbound_verify/state_store.py b/inbound_verify/state_store.py index 874b8d9..2cfc669 100644 --- a/inbound_verify/state_store.py +++ b/inbound_verify/state_store.py @@ -242,6 +242,25 @@ def set_data_state(site, kind, ready, generated_at, business_date=None): _upsert(conn, site, **fields) +def set_business_date(site, kind, business_date): + """仅写业务日期快照（不碰 ready/generated_at）。 + + 下载成功钩子用：ready 语义已移交「入库成功」（见 reset_data_ready / _persist_to_db）， + 下载阶段只记业务日期，供前端状态盘显示「是哪天的数据」。 + """ + with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: + _upsert(conn, site, **{f"{kind}_business_date": business_date or ""}) + + +def set_ready(site, kind, ready): + """仅写就绪态（不碰 business_date/generated_at）。 + + 供心跳从 ingest_state 派生 ready 用——ready 现为 DB 入库真相的派生视图， + 非启动重置、不读 Excel。""" + with sqlite3.connect(STATE_DB_PATH, timeout=3.0) as conn: + _upsert(conn, site, **{f"{kind}_ready": 1 if ready else 0}) + + def get_all_status(): """返回 {site: {各字段}}；库不存在则返回 {}。""" if not os.path.exists(STATE_DB_PATH):