82c80fc859
- 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>
620 lines
22 KiB
Python
620 lines
22 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
应到未到数据比对
|
||
|
||
目的：对中通 / 顺心 / 韵达 / 安能四个站点，比对各自的「应到货物数据」与
|
||
「实到货物数据」，逐件找出应到却未到的子单，汇总到 output/应到未到数据.xlsx。
|
||
工作簿结构：
|
||
· 汇总报表 —— 首页，跨站点统计与可视化（不含明细）。
|
||
· 中通 / 顺心 / 韵达 / 安能 —— 各站未到明细，每行一件。
|
||
|
||
核心逻辑（四站点统一）：
|
||
1. 每个运单的应到件数 N = 应到数据中的「录单件数」（输出列名为「总件数」）。
|
||
2. 该运单应到的子单序号集合 = {1, 2, …, N}。
|
||
3. 从实到数据中解析出该运单实际已到的序号集合。
|
||
4. 应到未到 = {1…N} − 已到序号。
|
||
—— 缺件按定义不在实到里，故其子单号 / 扫描单号由程序按各站格式现拼生成。
|
||
|
||
各站差异（已据真实数据核定）：
|
||
中通：实到「运单号」列即复合串 = 运单号 + 总数(4位) + 顺序号(4位)，从右解析。
|
||
顺心：实到有干净「运单号」列 +「子单号」= 运单号 + 顺序号(3位)。
|
||
韵达：实到「主单号」对应运单号，「子单号」= 主单号 + 顺序号(4位)。
|
||
安能：实到「所属单号」对应运单号，「扫描单号」= 所属单号 + 总数(4位) + 顺序号(4位)。
|
||
|
||
目录约定：
|
||
源数据放在脚本同级目录的 downloads/ 下；结果写入 output/（不存在则自动创建）。
|
||
"""
|
||
|
||
import os
|
||
from datetime import datetime
|
||
from collections import defaultdict
|
||
|
||
import pandas as pd
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
|
||
from openpyxl.chart import BarChart, Reference
|
||
from openpyxl.worksheet.page import PageMargins
|
||
from openpyxl.worksheet.properties import PageSetupProperties
|
||
|
||
BASE = os.path.dirname(os.path.abspath(__file__))
|
||
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 = ["类型", "子单号", "运单号", "最新扫描记录"]
|
||
|
||
|
||
# ============================ 比对逻辑 ============================
|
||
|
||
|
||
def arrived_zhongtong(df):
|
||
"""中通：运单号列即复合串，右侧 8 位 = 总数(4)+顺序(4)，其余为运单号。"""
|
||
res = defaultdict(set)
|
||
for v in df["运单号"]:
|
||
v = str(v).strip()
|
||
if len(v) > 8 and v[-4:].isdigit():
|
||
res[v[:-8]].add(int(v[-4:]))
|
||
return res
|
||
|
||
|
||
def arrived_by_prefix(main_col, sub_col, has_total=False):
|
||
"""顺心 / 韵达 / 安能：子单号以主单号为前缀，后缀含顺序号。
|
||
has_total=True 时后缀为 总数(4)+顺序(4)，取末 4 位为顺序号。"""
|
||
|
||
def parse(df):
|
||
res = defaultdict(set)
|
||
for m, s in zip(df[main_col], df[sub_col]):
|
||
m, s = str(m).strip(), str(s).strip()
|
||
if not m or not s:
|
||
continue
|
||
if s == m: # 无后缀的单件，记为第 1 件
|
||
res[m].add(1)
|
||
continue
|
||
if not s.startswith(m):
|
||
continue
|
||
seq = s[len(m) :][-4:] if has_total else s[len(m) :]
|
||
if seq.isdigit():
|
||
res[m].add(int(seq))
|
||
return res
|
||
|
||
return parse
|
||
|
||
|
||
def code_zhongtong(wb, seq, n):
|
||
return f"{seq:04d}" # 中通：仅顺序号
|
||
|
||
|
||
def code_shunxin(wb, seq, n):
|
||
return f"{wb}{seq:03d}" # 顺心：运单号 + 3 位顺序
|
||
|
||
|
||
def code_yunda(wb, seq, n):
|
||
return f"{wb}{seq:04d}" # 韵达：主单号 + 4 位顺序
|
||
|
||
|
||
def code_anneng(wb, seq, n):
|
||
return f"{wb}{n:04d}{seq:04d}" # 安能：运单号 + 总数 + 顺序
|
||
|
||
|
||
STATIONS = [
|
||
{
|
||
"name": "中通",
|
||
"exp": "中通-应到货物数据.xlsx",
|
||
"act": "中通-实到货物数据.xlsx",
|
||
"arrived": arrived_zhongtong,
|
||
"code": code_zhongtong,
|
||
"code_col": "子单号",
|
||
"columns": ["交接单号", "运单号", "子单号", "总件数"],
|
||
},
|
||
{
|
||
"name": "顺心",
|
||
"exp": "顺心-应到货物数据.xlsx",
|
||
"act": "顺心-实到货物数据.xlsx",
|
||
"arrived": arrived_by_prefix("运单号", "子单号"),
|
||
"code": code_shunxin,
|
||
"code_col": "子单号",
|
||
"columns": ["班次号", "交接单号", "运单号", "子单号", "总件数"],
|
||
},
|
||
{
|
||
"name": "韵达",
|
||
"exp": "韵达-应到货物数据.xlsx",
|
||
"act": "韵达-实到货物数据.xlsx",
|
||
"arrived": arrived_by_prefix("主单号", "子单号"),
|
||
"code": code_yunda,
|
||
"code_col": "子单号",
|
||
"columns": ["交接单号", "运单号", "子单号", "总件数"],
|
||
},
|
||
{
|
||
"name": "安能",
|
||
"exp": "安能-应到货物数据.xlsx",
|
||
"act": "安能-实到货物数据.xlsx",
|
||
"arrived": arrived_by_prefix("所属单号", "扫描单号", has_total=True),
|
||
"code": code_anneng,
|
||
"code_col": "扫描单号",
|
||
"columns": ["交接单号", "运单号", "扫描单号", "总件数"],
|
||
},
|
||
]
|
||
|
||
|
||
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):
|
||
print(f"[跳过] {cfg['name']}：downloads 下缺少 {cfg['exp']} 或 {cfg['act']}")
|
||
return None
|
||
|
||
df_exp = pd.read_excel(exp_path, dtype=str).fillna("")
|
||
df_act = pd.read_excel(act_path, dtype=str).fillna("")
|
||
|
||
# 同一运单可能有多条交接记录（录单件数一致），按运单号去重、保留首条
|
||
dup = int(df_exp["运单号"].duplicated().sum())
|
||
df_exp = df_exp.drop_duplicates(subset=["运单号"], keep="first")
|
||
|
||
arrived = cfg["arrived"](df_act)
|
||
|
||
rows = []
|
||
exp_pieces = full_miss = part_miss = 0
|
||
for _, r in df_exp.iterrows():
|
||
wb = str(r["运单号"]).strip()
|
||
if not wb:
|
||
continue
|
||
try:
|
||
n = int(float(r["总件数"] if "总件数" in r else r["录单件数"]))
|
||
except (TypeError, ValueError, KeyError):
|
||
try:
|
||
n = int(float(r["录单件数"]))
|
||
except (TypeError, ValueError, KeyError):
|
||
continue
|
||
if n <= 0:
|
||
continue
|
||
exp_pieces += n
|
||
missing = sorted(set(range(1, n + 1)) - arrived.get(wb, set()))
|
||
if missing:
|
||
if len(missing) == n:
|
||
full_miss += 1
|
||
else:
|
||
part_miss += 1
|
||
for seq in missing:
|
||
row = {}
|
||
for col in cfg["columns"]:
|
||
if col == "运单号":
|
||
row[col] = wb
|
||
elif col == cfg["code_col"]:
|
||
row[col] = cfg["code"](wb, seq, n)
|
||
elif col == "总件数":
|
||
row[col] = n
|
||
else:
|
||
row[col] = str(r.get(col, "")).strip()
|
||
rows.append(row)
|
||
|
||
stats = {
|
||
"运单数": len(df_exp),
|
||
"应到件": exp_pieces,
|
||
"已到件": exp_pieces - len(rows),
|
||
"未到件": len(rows),
|
||
"涉及运单": full_miss + part_miss,
|
||
"完全未到": full_miss,
|
||
"部分未到": part_miss,
|
||
"重复运单": dup,
|
||
}
|
||
return cfg["columns"], rows, stats
|
||
|
||
|
||
# ============================ 样式常量 ============================
|
||
|
||
FONT = "微软雅黑"
|
||
NAVY = "1F3864" # 标题栏
|
||
BLUE = "305496" # 表头
|
||
LIGHTBLUE = "D6DCE5" # 合计行
|
||
CARD_BG = "F2F6FC" # 指标卡底
|
||
RED = "C00000" # 未到
|
||
GREEN = "548235" # 已到
|
||
GRAY = "808080"
|
||
ZEBRA = "F4F7FC"
|
||
LINE = "D9D9D9"
|
||
TILE = "BFBFBF"
|
||
|
||
THIN = Side(style="thin", color=LINE)
|
||
BORDER = Border(left=THIN, right=THIN, top=THIN, bottom=THIN)
|
||
|
||
|
||
def heat(rate):
|
||
"""未到率热力底色：绿(低) / 黄(中) / 红(高)。"""
|
||
if rate >= 0.50:
|
||
return "FFC7CE"
|
||
if rate >= 0.15:
|
||
return "FFEB9C"
|
||
return "C6EFCE"
|
||
|
||
|
||
# ============================ 写明细表 ============================
|
||
|
||
HEADER_FILL = PatternFill("solid", fgColor=BLUE)
|
||
HEADER_FONT = Font(name=FONT, bold=True, color="FFFFFF", size=11)
|
||
BODY_FONT = Font(name=FONT, size=10)
|
||
|
||
|
||
def write_station(ws, columns, rows):
|
||
ws.sheet_view.showGridLines = False
|
||
ws.append(columns)
|
||
for c in range(1, len(columns) + 1):
|
||
cell = ws.cell(row=1, column=c)
|
||
cell.fill = HEADER_FILL
|
||
cell.font = HEADER_FONT
|
||
cell.alignment = Alignment(horizontal="center", vertical="center")
|
||
cell.border = BORDER
|
||
for row in rows:
|
||
ws.append([row[c] for c in columns])
|
||
for r in range(2, ws.max_row + 1):
|
||
for c, col in enumerate(columns, start=1):
|
||
cell = ws.cell(row=r, column=c)
|
||
cell.font = BODY_FONT
|
||
cell.border = BORDER
|
||
if col == "总件数":
|
||
cell.number_format = "#,##0"
|
||
cell.alignment = Alignment(horizontal="right", vertical="center")
|
||
else:
|
||
cell.number_format = "@" # 文本，避免长单号被转科学计数
|
||
for c, col in enumerate(columns, start=1):
|
||
body = [len(str(row[col])) for row in rows] if rows else []
|
||
width = min(max([len(str(col))] + body) + 4, 36)
|
||
ws.column_dimensions[ws.cell(row=1, column=c).column_letter].width = max(
|
||
width, 12
|
||
)
|
||
ws.freeze_panes = "A2"
|
||
ws.page_setup.orientation = "landscape"
|
||
ws.page_setup.fitToWidth = 1
|
||
ws.page_setup.fitToHeight = 0
|
||
ws.sheet_properties.pageSetUpPr = PageSetupProperties(fitToPage=True)
|
||
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}
|
||
|
||
|
||
# ============================ 写汇总报表 ============================
|
||
|
||
|
||
def build_summary(ws, results, generated_at):
|
||
center = Alignment(horizontal="center", vertical="center")
|
||
left = Alignment(horizontal="left", vertical="center", indent=1)
|
||
|
||
# 合计/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
|
||
ws.column_dimensions["A"].width = 2.5
|
||
for col, w in {
|
||
"B": 11,
|
||
"C": 12,
|
||
"D": 12,
|
||
"E": 12,
|
||
"F": 12,
|
||
"G": 11,
|
||
"H": 14,
|
||
"I": 14,
|
||
}.items():
|
||
ws.column_dimensions[col].width = w
|
||
ws.row_dimensions[1].height = 6
|
||
|
||
# —— 标题栏 ——
|
||
ws.merge_cells("B2:I2")
|
||
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 c in row:
|
||
c.fill = PatternFill("solid", fgColor=NAVY)
|
||
ws.row_dimensions[2].height = 34
|
||
ws.merge_cells("B3:I3")
|
||
sub = ws["B3"]
|
||
sub.value = f"数据快照 · 生成于 {generated_at}"
|
||
sub.font = Font(name=FONT, size=10, color=GRAY)
|
||
sub.alignment = Alignment(horizontal="right", vertical="center")
|
||
ws.row_dimensions[3].height = 18
|
||
|
||
# —— KPI 指标卡 ——
|
||
cards = [
|
||
("应到总件数", t_exp, NAVY, "#,##0"),
|
||
("已到总件数", t_arr, GREEN, "#,##0"),
|
||
("未到总件数", t_miss, RED, "#,##0"),
|
||
("总体未到率", rate, RED, "0.0%"),
|
||
]
|
||
spans = [
|
||
("B5:C5", "B6:C6"),
|
||
("D5:E5", "D6:E6"),
|
||
("F5:G5", "F6:G6"),
|
||
("H5:I5", "H6:I6"),
|
||
]
|
||
card_bg = PatternFill("solid", fgColor=CARD_BG)
|
||
thin = Side(style="thin", color=TILE)
|
||
for (lab, val, acc, fmt), (lrng, vrng) in zip(cards, spans):
|
||
ws.merge_cells(lrng)
|
||
ws.merge_cells(vrng)
|
||
acctop = Side(style="medium", color=acc)
|
||
for row in ws[lrng]:
|
||
for c in row:
|
||
c.fill = card_bg
|
||
c.font = Font(name=FONT, size=10, color=GRAY)
|
||
c.alignment = center
|
||
c.border = Border(left=thin, right=thin, top=acctop, bottom=thin)
|
||
for row in ws[vrng]:
|
||
for c in row:
|
||
c.fill = card_bg
|
||
c.font = Font(name=FONT, bold=True, size=20, color=acc)
|
||
c.alignment = center
|
||
c.border = Border(left=thin, right=thin, top=thin, bottom=thin)
|
||
ws[lrng.split(":")[0]].value = lab
|
||
vc = ws[vrng.split(":")[0]]
|
||
vc.value = val
|
||
vc.number_format = fmt
|
||
ws.row_dimensions[5].height = 18
|
||
ws.row_dimensions[6].height = 38
|
||
ws.row_dimensions[7].height = 8
|
||
|
||
# —— 小节标题 ——
|
||
ws.merge_cells("B8:I8")
|
||
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 c in row:
|
||
c.border = Border(bottom=Side(style="medium", color=BLUE))
|
||
ws.row_dimensions[8].height = 22
|
||
|
||
# —— 统计表头 ——
|
||
headers = [
|
||
"站点",
|
||
"应到运单数",
|
||
"应到件数",
|
||
"已到件数",
|
||
"未到件数",
|
||
"未到率",
|
||
"完全未到运单",
|
||
"部分未到运单",
|
||
]
|
||
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.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,
|
||
s["运单数"],
|
||
s["应到件"],
|
||
s["已到件"],
|
||
s["未到件"],
|
||
srate,
|
||
s["完全未到"],
|
||
s["部分未到"],
|
||
]
|
||
for i, v in enumerate(vals):
|
||
col = chr(ord("B") + i)
|
||
cell = ws[f"{col}{r}"]
|
||
cell.value = v
|
||
cell.font = BODY_FONT
|
||
cell.border = BORDER
|
||
cell.alignment = left if i == 0 else center
|
||
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 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
|
||
|
||
# —— 合计行 ——
|
||
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]
|
||
for i, v in enumerate(totals):
|
||
col = chr(ord("B") + i)
|
||
cell = ws[f"{col}{r}"]
|
||
cell.value = v
|
||
cell.fill = tot_fill
|
||
cell.font = tot_font
|
||
cell.border = BORDER
|
||
cell.alignment = left if i == 0 else center
|
||
if i in (1, 2, 3, 4, 6, 7):
|
||
cell.number_format = "#,##0"
|
||
if i == 5:
|
||
cell.number_format = "0.0%"
|
||
ws.row_dimensions[r].height = 20
|
||
last_data_row = 9 + len(four) # 图表只取 4 站（百世无应到/已到基数，不绘图）
|
||
chart_anchor = r + 2
|
||
|
||
# —— 堆叠柱状图：各站已到 / 未到 ——
|
||
chart = BarChart()
|
||
chart.type = "col"
|
||
chart.grouping = "stacked"
|
||
chart.overlap = 100
|
||
chart.title = "各站点到货构成（已到 / 未到 件数）"
|
||
data = Reference(
|
||
ws, min_col=5, max_col=6, min_row=9, max_row=last_data_row
|
||
) # E已到 F未到
|
||
chart.add_data(data, titles_from_data=True)
|
||
cats = Reference(ws, min_col=2, min_row=10, max_row=last_data_row)
|
||
chart.set_categories(cats)
|
||
chart.series[0].graphicalProperties.solidFill = GREEN
|
||
chart.series[1].graphicalProperties.solidFill = RED
|
||
chart.y_axis.title = "件数"
|
||
chart.x_axis.delete = False
|
||
chart.y_axis.delete = False
|
||
chart.legend.position = "b"
|
||
chart.height = 9
|
||
chart.width = 20
|
||
ws.add_chart(chart, f"B{chart_anchor}")
|
||
|
||
# —— 口径说明 ——
|
||
note_row = chart_anchor + 19
|
||
notes = [
|
||
"指标口径：未到率 ＝ 未到件数 ÷ 应到件数；完全未到运单 ＝ 整单零到货；部分未到运单 ＝ 部分到货、部分缺件。",
|
||
"合计 / 图表仅含 4 站（顺心/中通/韵达/安能，应到−实到口径）；百世为站点直供未到、无应到基数，单列不计入合计。",
|
||
"本次下载失败的站点标注为（无数据）并计 0，不影响其余站点统计。",
|
||
"明细见各站点工作表；4 站缺件的子单号 / 扫描单号按各站编号规则生成，并非实到原始记录。",
|
||
]
|
||
for k, text in enumerate(notes):
|
||
rr = note_row + k
|
||
ws.merge_cells(f"B{rr}:I{rr}")
|
||
cell = ws[f"B{rr}"]
|
||
cell.value = text
|
||
cell.font = Font(name=FONT, size=9, color=GRAY)
|
||
cell.alignment = Alignment(horizontal="left", vertical="center", wrap_text=True)
|
||
|
||
ws.page_setup.orientation = "landscape"
|
||
ws.page_setup.fitToWidth = 1
|
||
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}"
|
||
|
||
|
||
# ============================ 主流程 ============================
|
||
|
||
|
||
def main():
|
||
"""菜单 [9] / 离线入口：用 downloads/ 下现有文件生成全站汇总报告（有文件的站即纳入）。"""
|
||
print("应到未到比对（全站汇总）")
|
||
print("-" * 56)
|
||
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
|
||
undel = build_full_report(include)
|
||
print("-" * 56)
|
||
for name in ALL_REPORT_SITES:
|
||
if name in include:
|
||
print(f"{name}：未到 {undel.get(name)} 件")
|
||
else:
|
||
print(f"{name}：无数据，跳过")
|
||
print(f"已输出：{OUTFILE}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|