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

业务日期端到端：状态库快照 + 下载写入 + 报告读库

Browse Source 
- state_store: site_status 加 expected/actual/undelivered_business_date 列（建表+旧库迁移）；set_data_state 加可选 business_date（None 时保留，心跳不覆盖下载快照）
- runtime: dispatch_task 下载成功后快照业务日期（today−offset）；4 站 undelivered 一并补写 expected/actual/undelivered 三列（_site_undelivered_handler 内部连下绕过 dispatch，否则应到/实到业务日期丢失）
- expected_undelivered: 报告数据日期列改读状态库（_read_business_dates），退役 mtime 反推；KPI 卡 2-3-2-2 等宽对齐 + 柱状图图例 overlay 分行

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
Misaka
2026-07-18 14:09:08 +08:00
parent d44521d953
commit 0be3399165
3 changed files with 139 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

80
expected_undelivered.py
View File

@@ -326,7 +326,24 @@ def write_site_file(name):
return True
def build_full_report(include):
def _read_business_dates(include):
"""从状态库读各站业务日期（dispatch 下载成功时快照写入），供报告「数据日期」列。
4 站取 expected_business_date（报告按应到口径）；百世取 undelivered_business_date。
从未下过的站返回空串（诚实留空，不反推）。"""
import state_store # lazy import：比对模块本身保持纯离线
status = state_store.get_all_status()
dates = {}
for name in include:
s = status.get(name, {})
if name == "百世":
dates[name] = s.get("undelivered_business_date", "")
else:
dates[name] = s.get("expected_business_date", "")
return dates
def build_full_report(include, dates=None):
"""生成全站汇总报表 output/应到未到数据.xlsx。
include: 本次成功的站点集合；未成功站点在汇总里保留行、无数据（不影响他站）。
返回 {站点: 未到件或None} 供日志。"""
@@ -350,7 +367,12 @@ def build_full_report(include):
ws = wb.create_sheet(name)
write_station(ws, columns, rows)
build_summary(summary_ws, summary, datetime.now().strftime("%Y-%m-%d %H:%M"))
build_summary(
summary_ws,
summary,
datetime.now().strftime("%Y-%m-%d %H:%M"),
dates=dates or {},
)
wb.save(OUTFILE)
return {n: (s["未到件"] if s else None) for (n, s) in summary}
@@ -358,7 +380,8 @@ def build_full_report(include):
# ============================ 写汇总报表 ============================
def build_summary(ws, results, generated_at):
def build_summary(ws, results, generated_at, dates=None):
dates = dates or {}
center = Alignment(horizontal="center", vertical="center")
left = Alignment(horizontal="left", vertical="center", indent=1)
@@ -375,31 +398,33 @@ def build_summary(ws, results, generated_at):
ws.sheet_view.showGridLines = False
ws.column_dimensions["A"].width = 2.5
# 列宽按「4 个 KPI 卡等宽」设计：B+C = D+E+F = G+H = I+J = 26
for col, w in {
"B": 11,
"C": 12,
"D": 12,
"E": 12,
"F": 12,
"G": 11,
"B": 12,
"C": 14,
"D": 9,
"E": 9,
"F": 8,
"G": 12,
"H": 14,
"I": 14,
"I": 13,
"J": 13,
}.items():
ws.column_dimensions[col].width = w
ws.row_dimensions[1].height = 6
# —— 标题栏 ——
ws.merge_cells("B2:I2")
ws.merge_cells("B2:J2")
t = ws["B2"]
t.value = "应到未到比对 · 汇总报表"
t.fill = PatternFill("solid", fgColor=NAVY)
t.font = Font(name=FONT, bold=True, size=18, color="FFFFFF")
t.alignment = center
for row in ws["B2:I2"]:
for row in ws["B2:J2"]:
for c in row:
c.fill = PatternFill("solid", fgColor=NAVY)
ws.row_dimensions[2].height = 34
ws.merge_cells("B3:I3")
ws.merge_cells("B3:J3")
sub = ws["B3"]
sub.value = f"数据快照 · 生成于 {generated_at}"
sub.font = Font(name=FONT, size=10, color=GRAY)
@@ -413,11 +438,12 @@ def build_summary(ws, results, generated_at):
("未到总件数", t_miss, RED, "#,##0"),
("总体未到率", rate, RED, "0.0%"),
]
# 2-3-2-2 分布填满 B-J（9 列），配合上方列宽使 4 卡视觉等宽
spans = [
("B5:C5", "B6:C6"),
("D5:E5", "D6:E6"),
("F5:G5", "F6:G6"),
("H5:I5", "H6:I6"),
("D5:F5", "D6:F6"),
("G5:H5", "G6:H6"),
("I5:J5", "I6:J6"),
]
card_bg = PatternFill("solid", fgColor=CARD_BG)
thin = Side(style="thin", color=TILE)
@@ -446,12 +472,12 @@ def build_summary(ws, results, generated_at):
ws.row_dimensions[7].height = 8
# —— 小节标题 ——
ws.merge_cells("B8:I8")
ws.merge_cells("B8:J8")
sec = ws["B8"]
sec.value = "各站点明细统计"
sec.font = Font(name=FONT, bold=True, size=12, color=NAVY)
sec.alignment = Alignment(horizontal="left", vertical="center")
for row in ws["B8:I8"]:
for row in ws["B8:J8"]:
for c in row:
c.border = Border(bottom=Side(style="medium", color=BLUE))
ws.row_dimensions[8].height = 22
@@ -466,16 +492,18 @@ def build_summary(ws, results, generated_at):
"未到率",
"完全未到运单",
"部分未到运单",
"数据日期",
]
head_align = Alignment(horizontal="center", vertical="center", wrap_text=True)
for i, h in enumerate(headers):
col = chr(ord("B") + i)
cell = ws[f"{col}9"]
cell.value = h
cell.fill = HEADER_FILL
cell.font = HEADER_FONT
cell.alignment = center
cell.alignment = head_align
cell.border = BORDER
ws.row_dimensions[9].height = 22
ws.row_dimensions[9].height = 30
# —— 各站数据行（4 站 + 百世）——
r = 10
@@ -498,6 +526,7 @@ def build_summary(ws, results, generated_at):
s["完全未到"],
s["部分未到"],
]
vals.append(dates.get(name, "")) # 末列：该站业务日期
for i, v in enumerate(vals):
col = chr(ord("B") + i)
cell = ws[f"{col}{r}"]
@@ -519,7 +548,7 @@ def build_summary(ws, results, generated_at):
# —— 合计行 ——
tot_fill = PatternFill("solid", fgColor=LIGHTBLUE)
tot_font = Font(name=FONT, bold=True, size=10)
totals = ["合计", t_wb, t_exp, t_arr, t_miss, rate, t_full, t_part]
totals = ["合计", t_wb, t_exp, t_arr, t_miss, rate, t_full, t_part, ""]
for i, v in enumerate(totals):
col = chr(ord("B") + i)
cell = ws[f"{col}{r}"]
@@ -554,6 +583,7 @@ def build_summary(ws, results, generated_at):
chart.x_axis.delete = False
chart.y_axis.delete = False
chart.legend.position = "b"
chart.legend.overlay = False # 不覆盖绘图区：图例独占底部一行，与 X 轴站点名错开
chart.height = 9
chart.width = 20
ws.add_chart(chart, f"B{chart_anchor}")
@@ -565,10 +595,11 @@ def build_summary(ws, results, generated_at):
"合计 / 图表仅含 4 站（顺心/中通/韵达/安能，应到−实到口径）；百世为站点直供未到、无应到基数，单列不计入合计。",
"本次下载失败的站点标注为（无数据）并计 0，不影响其余站点统计。",
"明细见各站点工作表；4 站缺件的子单号 / 扫描单号按各站编号规则生成，并非实到原始记录。",
"数据日期：各站本次纳入数据对应的业务日期（＝应到数据下载日 − 日期偏移；韵达偏移 1 为前一日）；合计为多站混合、不标注。",
]
for k, text in enumerate(notes):
rr = note_row + k
ws.merge_cells(f"B{rr}:I{rr}")
ws.merge_cells(f"B{rr}:J{rr}")
cell = ws[f"B{rr}"]
cell.value = text
cell.font = Font(name=FONT, size=9, color=GRAY)
@@ -579,7 +610,7 @@ def build_summary(ws, results, generated_at):
ws.page_setup.fitToHeight = 0
ws.sheet_properties.pageSetUpPr = PageSetupProperties(fitToPage=True)
ws.page_margins = PageMargins(left=0.4, right=0.4, top=0.5, bottom=0.5)
ws.print_area = f"A1:I{note_row + 1}"
ws.print_area = f"A1:J{note_row + 1}"
# ============================ 主流程 ============================
@@ -605,7 +636,8 @@ def main():
if not include:
print("未处理任何站点：请确认 downloads/ 下存在源数据文件。")
return
undel = build_full_report(include)
dates = _read_business_dates(include)
undel = build_full_report(include, dates=dates)
print("-" * 56)
for name in ALL_REPORT_SITES:
if name in include:

39
runtime.py
View File

@@ -16,7 +16,7 @@ import socket
import subprocess
import time
import urllib.request
from datetime import datetime
from datetime import datetime, timedelta
import yaml
from playwright.sync_api import sync_playwright
@@ -490,6 +490,42 @@ TASK_HANDLERS = {
}
def _record_business_date(site, kind):
"""下载成功后，把本次数据的业务日期快照写进状态库（供前端/报告显示「是哪天的数据」）。
业务日期 = 下载当天 − 该数据对应的日期偏移。__compare__ 无数据概念，跳过。
kind → 写入：
expected/actual：各写自己一列（偏移各取其列）。
undelivered：百世直供（恒 0）写 undelivered；4 站未到由 _site_undelivered_handler
内部连带下了 expected+actual（不经 dispatch，无业务日期写入），故此处一并补写
expected/actual/undelivered 三列——actual 用 actual 偏移、未到跟随 expected 偏移。
顺带置 ready=True，让前端不必等心跳即可反映下载成功；写入失败仅告警、不影响任务判定。"""
if site == "__compare__":
return
today = datetime.now().date()
now = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
def _write(k, off):
biz = (today - timedelta(days=off)).strftime("%Y-%m-%d")
try:
state_store.set_data_state(
site, k, ready=True, generated_at=now, business_date=biz
)
except Exception as e:
print(f">> [状态] 写业务日期失败 {site}/{k}: {e}")
if kind == "expected":
_write("expected", state_store.get_offset(site, "expected"))
elif kind == "actual":
_write("actual", state_store.get_offset(site, "actual"))
elif site == "百世":
_write("undelivered", 0)
else: # 4 站 undelivered：连带补写 expected/actual/undelivered 三列
_write("expected", state_store.get_offset(site, "expected"))
_write("actual", state_store.get_offset(site, "actual"))
_write("undelivered", state_store.get_offset(site, "expected"))
def dispatch_task(ctx, task_spec):
"""执行一条任务。task_spec = {"site", "kind"}。返回 (status, error)。
@@ -513,6 +549,7 @@ def dispatch_task(ctx, task_spec):
ret = handler(ctx)
if ret is False:
return (state_store.TASK_FAILED, "任务执行失败（重试耗尽）")
_record_business_date(site, kind)
return (state_store.TASK_SUCCESS, None)
except Exception as e:
return (state_store.TASK_FAILED, str(e))

65
state_store.py
View File

@@ -44,6 +44,9 @@ def init_db():
actual_generated_at TEXT,
undelivered_ready INTEGER,
undelivered_generated_at TEXT,
expected_business_date TEXT,
actual_business_date TEXT,
undelivered_business_date TEXT,
updated_at TEXT
)
""")
@@ -51,6 +54,9 @@ def init_db():
for _col, _typedef in [
("undelivered_ready", "INTEGER NOT NULL DEFAULT 0"),
("undelivered_generated_at", "TEXT NOT NULL DEFAULT ''"),
("expected_business_date", "TEXT NOT NULL DEFAULT ''"),
("actual_business_date", "TEXT NOT NULL DEFAULT ''"),
("undelivered_business_date", "TEXT NOT NULL DEFAULT ''"),
]:
try:
conn.execute(f"ALTER TABLE site_status ADD COLUMN {_col} {_typedef}")
@@ -111,7 +117,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, undelivered_ready, undelivered_generated_at "
"expected_business_date, actual_ready, actual_generated_at, actual_business_date, "
"undelivered_ready, undelivered_generated_at, undelivered_business_date "
"FROM site_status WHERE site = ?",
(site,),
).fetchone()
@@ -120,10 +127,13 @@ def _upsert(conn, site, **fields):
"login_checked_at": "",
"expected_ready": 0,
"expected_generated_at": "",
"expected_business_date": "",
"actual_ready": 0,
"actual_generated_at": "",
"actual_business_date": "",
"undelivered_ready": 0,
"undelivered_generated_at": "",
"undelivered_business_date": "",
}
if row:
(
@@ -131,10 +141,13 @@ def _upsert(conn, site, **fields):
cur["login_checked_at"],
cur["expected_ready"],
cur["expected_generated_at"],
cur["expected_business_date"],
cur["actual_ready"],
cur["actual_generated_at"],
cur["actual_business_date"],
cur["undelivered_ready"],
cur["undelivered_generated_at"],
cur["undelivered_business_date"],
) = row
cur.update(fields)
cur["updated_at"] = _now()
@@ -145,26 +158,31 @@ def _upsert(conn, site, **fields):
cur["login_checked_at"],
cur["expected_ready"],
cur["expected_generated_at"],
cur["expected_business_date"],
cur["actual_ready"],
cur["actual_generated_at"],
cur["actual_business_date"],
cur["undelivered_ready"],
cur["undelivered_generated_at"],
cur["undelivered_business_date"],
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=?, "
"undelivered_ready=?, undelivered_generated_at=?, updated_at=? "
"expected_generated_at=?, expected_business_date=?, actual_ready=?, "
"actual_generated_at=?, actual_business_date=?, undelivered_ready=?, "
"undelivered_generated_at=?, undelivered_business_date=?, 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, "
"undelivered_ready, undelivered_generated_at, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
"expected_generated_at, expected_business_date, actual_ready, actual_generated_at, "
"actual_business_date, undelivered_ready, undelivered_generated_at, "
"undelivered_business_date, updated_at) "
"VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
values,
)
conn.commit()
@@ -185,14 +203,17 @@ def reset_login_states(sites):
_upsert(conn, site, login_state=LOGIN_UNKNOWN, login_checked_at="")
def set_data_state(site, kind, ready, generated_at):
"""更新单站数据态。kind: 'expected'/'actual'/'undelivered'；ready: bool；generated_at: str。"""
key_ready = f"{kind}_ready"
key_at = f"{kind}_generated_at"
def set_data_state(site, kind, ready, generated_at, business_date=None):
"""更新单站数据态。kind: 'expected'/'actual'/'undelivered'；ready: bool；
generated_at: str。business_date: str 或 None——None 时保留原值（心跳不覆盖下载快照）。"""
fields = {
f"{kind}_ready": 1 if ready else 0,
f"{kind}_generated_at": generated_at or "",
}
if business_date is not None:
fields[f"{kind}_business_date"] = business_date or ""
with sqlite3.connect(STATE_DB_PATH) as conn:
_upsert(
conn, site, **{key_ready: 1 if ready else 0, key_at: generated_at or ""}
)
_upsert(conn, site, **fields)
def get_all_status():
@@ -202,8 +223,9 @@ 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, "
"undelivered_ready, undelivered_generated_at, updated_at "
"expected_generated_at, expected_business_date, actual_ready, "
"actual_generated_at, actual_business_date, undelivered_ready, "
"undelivered_generated_at, undelivered_business_date, updated_at "
"FROM site_status"
).fetchall()
return {
@@ -212,11 +234,14 @@ def get_all_status():
"login_checked_at": r[2],
"expected_ready": bool(r[3]),
"expected_generated_at": r[4],
"actual_ready": bool(r[5]),
"actual_generated_at": r[6],
"undelivered_ready": bool(r[7]),
"undelivered_generated_at": r[8],
"updated_at": r[9],
"expected_business_date": r[5],
"actual_ready": bool(r[6]),
"actual_generated_at": r[7],
"actual_business_date": r[8],
"undelivered_ready": bool(r[9]),
"undelivered_generated_at": r[10],
"undelivered_business_date": r[11],
"updated_at": r[12],
}
for r in rows
}