d55c460c2d
- 应到未到比对.py -> expected_undelivered.py（英文名）；main() 的 sys.exit 改为 return，使其可被 main_router 安全调用而不杀进程。 - main_router.py：菜单 [9] 改为调用 expected_undelivered.main()——全站点自动比对， 输出 output/应到未到数据.xlsx（汇总报表 + 中通/顺心/韵达/安能 各站明细）；移除被 取代的旧 task_process_undelivered_data 及其专属 import pandas。 - .gitignore：忽略 output/（比对输出目录）。 Co-Authored-By: Claude <noreply@anthropic.com>
525 lines
18 KiB
Python
525 lines
18 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")
|
||
|
||
|
||
# ============================ 比对逻辑 ============================
|
||
|
||
|
||
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 process(cfg):
|
||
"""返回 (列名list, 明细行list[dict], 统计dict)；源文件缺失时返回 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 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)
|
||
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
|
||
|
||
# —— 各站数据行 ——
|
||
r = 10
|
||
for idx, (name, s) in enumerate(results):
|
||
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 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%"
|
||
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(results)
|
||
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 = [
|
||
"指标口径：未到率 ＝ 未到件数 ÷ 应到件数；完全未到运单 ＝ 整单零到货；部分未到运单 ＝ 部分到货、部分缺件。",
|
||
"明细见各站点工作表；缺件的子单号 / 扫描单号按各站编号规则生成，并非实到原始记录。",
|
||
]
|
||
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():
|
||
os.makedirs(OUTPUT, exist_ok=True)
|
||
wb = Workbook()
|
||
wb.remove(wb.active)
|
||
summary_ws = wb.create_sheet("汇总报表") # 首页占位
|
||
|
||
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:
|
||
print("未处理任何站点：请确认 downloads/ 下存在源数据文件。")
|
||
return
|
||
|
||
build_summary(summary_ws, results, datetime.now().strftime("%Y-%m-%d %H:%M"))
|
||
|
||
print("-" * 56)
|
||
wb.save(OUTFILE)
|
||
print(f"已输出：{OUTFILE}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|