LogisticsHubIPA/InboundVerify
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

未到数据按站独立 + 百世并入汇总 + 全量跑比对（失败容错）

Browse Source 
- state_store：site_status 加 undelivered_ready 字段（旧库 ALTER 迁移）；init_db 提早到启动最前（server lifespan + launch_and_prepare 第0步），避免 /api/status 早于迁移报错
- expected_undelivered：重构为 process(name)/process_baishi/write_site_file/build_full_report；build_summary 支持百世（仅未到件、无基数，不计入合计/图表）与失败容错（未成功站保留行无数据）
- runtime：4 站 ("站","undelivered") = 下应到+实到 → 比对写 <站>-未到数据.xlsx；("__compare__","compare") 改 run_all（顺序跑5站、记成功清单 → build_full_report，未登录/失败跳过）；DATA_FILENAMES 加 undelivered、心跳探测之；_site_undelivered_handler 用 is not False 与 dispatch 一致
- site_shunxin：shunxin_expected/actual_download 改为 return with_retry 结果（修复返回 None 致调用方误判失败、以及 with_retry 失败被当成功的潜在 bug）
- server：lifespan 启动时 init_db

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Misaka
2026-07-17 23:19:21 +08:00
parent 26a94999dc
commit 82c80fc859
5 changed files with 272 additions and 82 deletions
Download Patch File Download Diff File Expand all files Collapse all files

181
expected_undelivered.py
View File

@@ -41,6 +41,13 @@ DOWNLOADS = os.path.join(BASE, "downloads")
OUTPUT = os.path.join(BASE, "output")
OUTFILE = os.path.join(OUTPUT, "应到未到数据.xlsx")
# 汇总报表覆盖的全部站点（4 站在前、百世在末；汇总页图表只取 4 站）
ALL_REPORT_SITES = ["顺心", "中通", "韵达", "安能", "百世"]
# 4 站单站未到明细文件名（百世未到文件由站点直接产出，名为 BAISHI_FILE）
SITE_UNDELIVERED_FILE = "{name}-未到数据.xlsx"
BAISHI_FILE = "百世-应到未到货物数据.xlsx"
BAISHI_COLUMNS = ["类型", "子单号", "运单号", "最新扫描记录"]
# ============================ 比对逻辑 ============================
@@ -134,8 +141,16 @@ STATIONS = [
]
def process(cfg):
"""返回 (列名list, 明细行list[dict], 统计dict)；源文件缺失时返回 None。"""
def _site_cfg(name):
"""按名称取 4 站配置（百世不在 STATIONS，返回 None）。"""
return next((c for c in STATIONS if c["name"] == name), None)
def process(name):
"""4 站单站比对：返回 (列名list, 明细行list[dict], 统计dict)；源文件缺失或非 4 站返回 None。"""
cfg = _site_cfg(name)
if cfg is None:
return None
exp_path = os.path.join(DOWNLOADS, cfg["exp"])
act_path = os.path.join(DOWNLOADS, cfg["act"])
if not os.path.exists(exp_path) or not os.path.exists(act_path):
@@ -268,6 +283,78 @@ def write_station(ws, columns, rows):
ws.print_title_rows = "1:1"
# ============================ 单站 / 全量产出 ============================
def process_baishi():
"""百世：读站点直供的未到明细，返回 (columns, rows, stats)；文件缺失返回 None。
百世文件本身即未到结果（无应到/已到基数），统计只能给出未到件数。"""
path = os.path.join(DOWNLOADS, BAISHI_FILE)
if not os.path.exists(path):
return None
df = pd.read_excel(path, dtype=str).fillna("")
rows = df.to_dict("records")
wb_count = df["运单号"].nunique() if "运单号" in df.columns else len(rows)
stats = {
"运单数": wb_count,
"应到件": None,
"已到件": None,
"未到件": len(rows),
"涉及运单": wb_count,
"完全未到": None,
"部分未到": None,
"重复运单": 0,
}
return (BAISHI_COLUMNS, rows, stats)
def write_site_file(name):
"""4 站：把该站未到明细写到 downloads/<站>-未到数据.xlsx。
应到/实到缺（process 返回 None）→ 删旧文件、返回 False；成功返回 True。"""
path = os.path.join(DOWNLOADS, SITE_UNDELIVERED_FILE.format(name=name))
out = process(name)
if out is None:
if os.path.exists(path):
os.remove(path)
return False
columns, rows, _stats = out
wb = Workbook()
wb.remove(wb.active)
ws = wb.create_sheet(name)
write_station(ws, columns, rows)
wb.save(path)
return True
def build_full_report(include):
"""生成全站汇总报表 output/应到未到数据.xlsx。
include: 本次成功的站点集合；未成功站点在汇总里保留行、无数据（不影响他站）。
返回 {站点: 未到件或None} 供日志。"""
os.makedirs(OUTPUT, exist_ok=True)
wb = Workbook()
wb.remove(wb.active)
summary_ws = wb.create_sheet("汇总报表") # 首页占位
summary = [] # (name, stats_or_None)，顺序：4 站 + 百世
for name in ALL_REPORT_SITES:
if name == "百世":
out = process_baishi() if "百世" in include else None
columns = BAISHI_COLUMNS
else:
out = process(name) if name in include else None
cfg = _site_cfg(name)
columns = cfg["columns"] if cfg else []
stats = out[2] if out is not None else None
rows = out[1] if out is not None else []
summary.append((name, stats))
ws = wb.create_sheet(name)
write_station(ws, columns, rows)
build_summary(summary_ws, summary, datetime.now().strftime("%Y-%m-%d %H:%M"))
wb.save(OUTFILE)
return {n: (s["未到件"] if s else None) for (n, s) in summary}
# ============================ 写汇总报表 ============================
@@ -275,12 +362,15 @@ def build_summary(ws, results, generated_at):
center = Alignment(horizontal="center", vertical="center")
left = Alignment(horizontal="left", vertical="center", indent=1)
t_wb = sum(s["运单数"] for _, s in results)
t_exp = sum(s["应到件"] for _, s in results)
t_arr = sum(s["已到件"] for _, s in results)
t_miss = sum(s["未到件"] for _, s in results)
t_full = sum(s["完全未到"] for _, s in results)
t_part = sum(s["部分未到"] for _, s in results)
# 合计/KPI 只算 4 站中本次成功的（百世无应到基数、失败站无数据，均不计入）
four = [(n, s) for (n, s) in results if n != "百世"]
ok = [s for _, s in four if s]
t_wb = sum(s["运单数"] for s in ok)
t_exp = sum(s["应到件"] for s in ok)
t_arr = sum(s["已到件"] for s in ok)
t_miss = sum(s["未到件"] for s in ok)
t_full = sum(s["完全未到"] for s in ok)
t_part = sum(s["部分未到"] for s in ok)
rate = (t_miss / t_exp) if t_exp else 0
ws.sheet_view.showGridLines = False
@@ -387,9 +477,16 @@ def build_summary(ws, results, generated_at):
cell.border = BORDER
ws.row_dimensions[9].height = 22
# —— 各站数据行 ——
# —— 各站数据行（4 站 + 百世）——
r = 10
for idx, (name, s) in enumerate(results):
is_baishi = name == "百世"
srate = 0
if s is None:
vals = [f"{name}（无数据）", 0, 0, 0, 0, 0, 0, 0]
elif is_baishi:
vals = [name, s["运单数"], "—", "—", s["未到件"], "—", "—", "—"]
else:
srate = (s["未到件"] / s["应到件"]) if s["应到件"] else 0
vals = [
name,
@@ -408,12 +505,13 @@ def build_summary(ws, results, generated_at):
cell.font = BODY_FONT
cell.border = BORDER
cell.alignment = left if i == 0 else center
if idx % 2 == 1 and i != 5:
if s is None:
cell.fill = PatternFill("solid", fgColor="EFEFEF")
elif not is_baishi and idx % 2 == 1 and i != 5:
cell.fill = PatternFill("solid", fgColor=ZEBRA)
if i in (1, 2, 3, 4, 6, 7):
cell.number_format = "#,##0"
if i == 5:
cell.number_format = "0.0%"
if isinstance(v, (int, float)):
cell.number_format = "0.0%" if i == 5 else "#,##0"
if i == 5 and s is not None and not is_baishi:
cell.fill = PatternFill("solid", fgColor=heat(srate))
ws.row_dimensions[r].height = 19
r += 1
@@ -435,7 +533,7 @@ def build_summary(ws, results, generated_at):
if i == 5:
cell.number_format = "0.0%"
ws.row_dimensions[r].height = 20
last_data_row = 9 + len(results)
last_data_row = 9 + len(four) # 图表只取 4 站（百世无应到/已到基数，不绘图）
chart_anchor = r + 2
# —— 堆叠柱状图：各站已到 / 未到 ——
@@ -464,7 +562,9 @@ def build_summary(ws, results, generated_at):
note_row = chart_anchor + 19
notes = [
"指标口径：未到率 ＝ 未到件数 ÷ 应到件数；完全未到运单 ＝ 整单零到货；部分未到运单 ＝ 部分到货、部分缺件。",
"明细见各站点工作表；缺件的子单号 / 扫描单号按各站编号规则生成，并非实到原始记录。",
"合计 / 图表仅含 4 站（顺心/中通/韵达/安能，应到−实到口径）；百世为站点直供未到、无应到基数，单列不计入合计。",
"本次下载失败的站点标注为（无数据）并计 0，不影响其余站点统计。",
"明细见各站点工作表；4 站缺件的子单号 / 扫描单号按各站编号规则生成，并非实到原始记录。",
]
for k, text in enumerate(notes):
rr = note_row + k
@@ -486,37 +586,32 @@ def build_summary(ws, results, generated_at):
def main():
os.makedirs(OUTPUT, exist_ok=True)
wb = Workbook()
wb.remove(wb.active)
summary_ws = wb.create_sheet("汇总报表") # 首页占位
print("应到未到比对结果")
"""菜单 [9] / 离线入口：用 downloads/ 下现有文件生成全站汇总报告（有文件的站即纳入）。"""
print("应到未到比对（全站汇总）")
print("-" * 56)
results = []
for cfg in STATIONS:
out = process(cfg)
if out is None:
continue
columns, rows, stats = out
ws = wb.create_sheet(cfg["name"])
write_station(ws, columns, rows)
results.append((cfg["name"], stats))
extra = f"，应到重复运单 {stats['重复运单']}" if stats["重复运单"] else ""
print(
f"{cfg['name']}：应到运单 {stats['运单数']}，应到件 {stats['应到件']}，"
f"已到 {stats['已到件']}，未到 {stats['未到件']} 件"
f"（涉及运单 {stats['涉及运单']}：完全未到 {stats['完全未到']} / 部分未到 {stats['部分未到']}）{extra}"
)
if not results:
include = set()
for name in ALL_REPORT_SITES:
if name == "百世":
if os.path.exists(os.path.join(DOWNLOADS, BAISHI_FILE)):
include.add(name)
else:
cfg = _site_cfg(name)
if (
cfg
and os.path.exists(os.path.join(DOWNLOADS, cfg["exp"]))
and os.path.exists(os.path.join(DOWNLOADS, cfg["act"]))
):
include.add(name)
if not include:
print("未处理任何站点：请确认 downloads/ 下存在源数据文件。")
return
build_summary(summary_ws, results, datetime.now().strftime("%Y-%m-%d %H:%M"))
undel = build_full_report(include)
print("-" * 56)
wb.save(OUTFILE)
for name in ALL_REPORT_SITES:
if name in include:
print(f"{name}：未到 {undel.get(name)} 件")
else:
print(f"{name}：无数据，跳过")
print(f"已输出：{OUTFILE}")

83
runtime.py
View File

@@ -55,11 +55,27 @@ HEARTBEAT_INTERVAL = 30
# 各站最终数据文件名（探测"数据是否已跑出来"）；百世为单流程
DATA_FILENAMES = {
"顺心": {"expected": "顺心-应到货物数据.xlsx", "actual": "顺心-实到货物数据.xlsx"},
"中通": {"expected": "中通-应到货物数据.xlsx", "actual": "中通-实到货物数据.xlsx"},
"韵达": {"expected": "韵达-应到货物数据.xlsx", "actual": "韵达-实到货物数据.xlsx"},
"安能": {"expected": "安能-应到货物数据.xlsx", "actual": "安能-实到货物数据.xlsx"},
"百世": {"expected": "百世-应到未到货物数据.xlsx", "actual": ""},
"顺心": {
"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"},
}
@@ -196,6 +212,9 @@ def launch_and_prepare(debug_mode=False, debug_target=""):
必须在"持有 Playwright 的线程"调用（交互模式主线程 / 服务模式 worker 线程）。
阻塞至所有站点登录就绪才返回。
"""
# 0. 状态库建表/迁移（须在 reset_login_states 等读写新字段的调用之前）
state_store.init_db()
# 1. 读 config（anneng.app_path）
anneng_app_path = ""
try:
@@ -309,8 +328,7 @@ def launch_and_prepare(debug_mode=False, debug_target=""):
if s in pages_map:
print(f" ✅ 【{s}】已就绪。")
# 8. 状态库 + 心跳初值（就绪轮询刚通过 → 各站视为已登录）
state_store.init_db()
# 8. 心跳初值（就绪轮询刚通过 → 各站视为已登录；init_db 已在启动时完成）
sites_to_watch = list(ready_status.keys())
for _site in sites_to_watch:
state_store.set_login_state(_site, True)
@@ -389,19 +407,66 @@ def _web_handler(site, download_func):
return handler
def _site_undelivered_handler(site):
"""4 站未到：下应到+实到 → 比对写 downloads/<站>-未到数据.xlsx。
任一下载失败 → 清掉旧未到文件、返回 False（前端不展示陈旧未到）。"""
def handler(ctx):
# 各站下载入口约定返回 True/False；顺心历史返回 None（视为成功，与 dispatch 一致）
exp_ok = TASK_HANDLERS[(site, "expected")](ctx) is not False
act_ok = (
(TASK_HANDLERS[(site, "actual")](ctx) is not False) if exp_ok else False
)
if exp_ok and act_ok:
return expected_undelivered.write_site_file(site)
stale = os.path.join(
DOWNLOAD_DIR, expected_undelivered.SITE_UNDELIVERED_FILE.format(name=site)
)
if os.path.exists(stale):
os.remove(stale)
return False
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
TASK_HANDLERS = {
("顺心", "expected"): _web_handler("顺心", site_shunxin.shunxin_expected_download),
("顺心", "actual"): _web_handler("顺心", site_shunxin.shunxin_actual_download),
("顺心", "undelivered"): _site_undelivered_handler("顺心"),
("百世", "undelivered"): _web_handler(
"百世", site_baishi.baishi_download_undelivered_data
),
("中通", "expected"): _web_handler("中通", site_zto.zto_expected_download),
("中通", "actual"): _web_handler("中通", site_zto.zto_actual_download),
("中通", "undelivered"): _site_undelivered_handler("中通"),
("韵达", "expected"): _web_handler("韵达", site_yunda.yunda_expected_download),
("韵达", "actual"): _web_handler("韵达", site_yunda.yunda_actual_download),
("韵达", "undelivered"): _site_undelivered_handler("韵达"),
("安能", "expected"): lambda ctx: site_anneng.anneng_expected_download(),
("安能", "actual"): lambda ctx: site_anneng.anneng_actual_download(),
("__compare__", "compare"): lambda ctx: (expected_undelivered.main() or True),
("安能", "undelivered"): _site_undelivered_handler("安能"),
("__compare__", "compare"): _run_all_compare_handler(),
}
@@ -449,6 +514,6 @@ 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"):
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)

1
server.py
View File

@@ -89,6 +89,7 @@ def _worker_loop():
@asynccontextmanager
async def lifespan(_app):
"""服务启停：起 worker 线程 / 通知 worker 停。"""
state_store.init_db() # 先建表/迁移状态库，确保早于 worker 就绪的 /api/status 可用
t = threading.Thread(target=_worker_loop, daemon=True)
worker_state["thread"] = t
t.start()

10
site_shunxin.py
View File

@@ -171,14 +171,17 @@ def shunxin_expected_download(pages):
for idx, (pg, tag) in enumerate(zip(pages, tags), start=1):
pg.bring_to_front()
print(f"\n========== 顺心 · 账号{idx}（{tag}）应到数据下载 ==========")
with_retry(
ok = with_retry(
f"顺心-{tag}",
"应到",
lambda p=pg, t=tag: shunxin_expected_download_impl(p, out_tag=t),
lambda p=pg: shunxin_reset(p),
)
if not ok:
return False # 某账号重试耗尽 → 整体失败，不融合（避免部分数据）
shunxin_merge_final("应到", tags)
return True
def shunxin_expected_download_impl(page, out_tag=""):
@@ -493,14 +496,17 @@ def shunxin_actual_download(pages):
for idx, (pg, tag) in enumerate(zip(pages, tags), start=1):
pg.bring_to_front()
print(f"\n========== 顺心 · 账号{idx}（{tag}）实到数据下载 ==========")
with_retry(
ok = with_retry(
f"顺心-{tag}",
"实到",
lambda p=pg, t=tag: shunxin_actual_download_impl(p, out_tag=t),
lambda p=pg: shunxin_reset(p),
)
if not ok:
return False
shunxin_merge_final("实到", tags)
return True
def shunxin_actual_download_impl(page, out_tag=""):

41
state_store.py
View File

@@ -42,9 +42,20 @@ def init_db():
expected_generated_at TEXT,
actual_ready INTEGER,
actual_generated_at TEXT,
undelivered_ready INTEGER,
undelivered_generated_at TEXT,
updated_at TEXT
)
""")
# 旧库迁移：补 undelivered 两列（新库已含；重复添加抛 OperationalError，忽略）
for _col, _typedef in [
("undelivered_ready", "INTEGER NOT NULL DEFAULT 0"),
("undelivered_generated_at", "TEXT NOT NULL DEFAULT ''"),
]:
try:
conn.execute(f"ALTER TABLE site_status ADD COLUMN {_col} {_typedef}")
except sqlite3.OperationalError:
pass
conn.execute("""
CREATE TABLE IF NOT EXISTS task_history (
id INTEGER PRIMARY KEY AUTOINCREMENT,
@@ -70,7 +81,8 @@ def _upsert(conn, site, **fields):
"""更新（或插入）单站：保留未传字段，刷新 updated_at。"""
row = conn.execute(
"SELECT login_state, login_checked_at, expected_ready, expected_generated_at, "
"actual_ready, actual_generated_at FROM site_status WHERE site = ?",
"actual_ready, actual_generated_at, undelivered_ready, undelivered_generated_at "
"FROM site_status WHERE site = ?",
(site,),
).fetchone()
cur = {
@@ -80,6 +92,8 @@ def _upsert(conn, site, **fields):
"expected_generated_at": "",
"actual_ready": 0,
"actual_generated_at": "",
"undelivered_ready": 0,
"undelivered_generated_at": "",
}
if row:
(
@@ -89,6 +103,8 @@ def _upsert(conn, site, **fields):
cur["expected_generated_at"],
cur["actual_ready"],
cur["actual_generated_at"],
cur["undelivered_ready"],
cur["undelivered_generated_at"],
) = row
cur.update(fields)
cur["updated_at"] = _now()
@@ -101,20 +117,24 @@ def _upsert(conn, site, **fields):
cur["expected_generated_at"],
cur["actual_ready"],
cur["actual_generated_at"],
cur["undelivered_ready"],
cur["undelivered_generated_at"],
cur["updated_at"],
)
if row:
conn.execute(
"UPDATE site_status SET login_state=?, login_checked_at=?, expected_ready=?, "
"expected_generated_at=?, actual_ready=?, actual_generated_at=?, updated_at=? "
"expected_generated_at=?, actual_ready=?, actual_generated_at=?, "
"undelivered_ready=?, undelivered_generated_at=?, updated_at=? "
"WHERE site=?",
values[1:] + (site,),
)
else:
conn.execute(
"INSERT INTO site_status (site, login_state, login_checked_at, expected_ready, "
"expected_generated_at, actual_ready, actual_generated_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?)",
"expected_generated_at, actual_ready, actual_generated_at, "
"undelivered_ready, undelivered_generated_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
values,
)
conn.commit()
@@ -136,9 +156,9 @@ def reset_login_states(sites):
def set_data_state(site, kind, ready, generated_at):
"""更新单站数据态。kind: 'expected'/'actual'；ready: bool；generated_at: str 或 ''。"""
key_ready = "expected_ready" if kind == "expected" else "actual_ready"
key_at = "expected_generated_at" if kind == "expected" else "actual_generated_at"
"""更新单站数据态。kind: 'expected'/'actual'/'undelivered'；ready: bool；generated_at: str。"""
key_ready = f"{kind}_ready"
key_at = f"{kind}_generated_at"
with sqlite3.connect(STATE_DB_PATH) as conn:
_upsert(
conn, site, **{key_ready: 1 if ready else 0, key_at: generated_at or ""}
@@ -152,7 +172,8 @@ def get_all_status():
with sqlite3.connect(STATE_DB_PATH) as conn:
rows = conn.execute(
"SELECT site, login_state, login_checked_at, expected_ready, "
"expected_generated_at, actual_ready, actual_generated_at, updated_at "
"expected_generated_at, actual_ready, actual_generated_at, "
"undelivered_ready, undelivered_generated_at, updated_at "
"FROM site_status"
).fetchall()
return {
@@ -163,7 +184,9 @@ def get_all_status():
"expected_generated_at": r[4],
"actual_ready": bool(r[5]),
"actual_generated_at": r[6],
"updated_at": r[7],
"undelivered_ready": bool(r[7]),
"undelivered_generated_at": r[8],
"updated_at": r[9],
}
for r in rows
}