17293bee79
Tier 1 regression: expected_undelivered.py carries its own BASE=dirname(__file__) anchor (a duplicate of paths.py). The package move dropped the file one level deeper, so BASE resolved to inbound_verify/ and DOWNLOADS/OUTPUT pointed at non-existent inbound_verify/downloads|output — while site downloads write to the project-root dirs via paths.py. process()/write_site_file() thus skipped the compare with '[跳过] downloads 下缺少 ...', returned None/False, and every undelivered task for the 4 web/app sites reported failed (重试耗尽) despite the files downloading fine. Fix: anchor BASE two levels up (mirrors paths.py). Verified: write_site_file returns True; all 4 undelivered tasks now success with 未到 files generated. Co-Authored-By: Claude <noreply@anthropic.com>
695 lines
27 KiB
Python
695 lines
27 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
应到未到数据比对（重构版）
|
||
|
||
目的：对中通 / 顺心 / 韵达 / 安能 四个站点，比对各自的「应到货物数据」与
|
||
「实到货物数据」，找出应到却未到的运单，汇总到 output/应到未到数据.xlsx。
|
||
（百世为站点直供未到明细，不参与 4 站比对；其应到/实到基数取自「扫描综合查询」应扫/已扫，见 process_baishi。）
|
||
|
||
核心口径（四站点统一，重构后）：
|
||
1. 应到件数 = 应到表「交接件数」之和（按运单号去重 keep-first）。
|
||
—— 录单件数 只是该单号的总录单量，实际只有“交接件数”会真正到站，
|
||
故应到必须按交接件数统计，不能用录单件数。
|
||
2. 实到件数 = 实到表「单号」的去重数量（直接数，不再由“应到−未到”倒推）。
|
||
—— 每扫描一件，系统生成该件的单号（一个单号=一件）；后缀含总数/顺序号，
|
||
但计数时无视后缀，仅对单号去重即得实到件数。
|
||
3. 未到件数 = max(0, 应到件数 − 实到件数)。
|
||
4. 未到明细（downloads/<站>-未到数据.xlsx）仅列“短少”运单（实到 < 应到），
|
||
每行：交接单号 | 运单号 | 总件数(=应到/交接件数) | 已到单号1 | 已到单号2 | …。
|
||
—— 实到扫描的顺序号是乱序的，缺件的“顺序号”无法反推，故不再编造子单号；
|
||
改为把该运单“实际扫到的单号”依次填到后续单元格，便于核对到了哪几件。
|
||
|
||
各站实到单号列 / 运单基号：
|
||
中通：单号列=运单号(复合串 H+运单号+总数+顺序)，基号=v[:-8]
|
||
顺心：单号列=子单号，基号=运单号
|
||
韵达：单号列=子单号，基号=主单号
|
||
安能：单号列=扫描单号，基号=所属单号
|
||
|
||
目录约定：
|
||
源数据放在脚本同级目录的 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
|
||
|
||
# 包内文件：上两级 = 项目根（与 paths.BASE_DIR 一致；站点下载落在 <root>/downloads）。
|
||
# 注：本模块自带锚点是 paths.py 的重复，Tier 2 计划改为直接引用 paths.DOWNLOAD_DIR/OUTPUT_DIR。
|
||
BASE = os.path.dirname(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_pieces_zhongtong(df):
|
||
"""中通：实到「运单号」为复合串（H + 运单号(12) + 总数(4) + 顺序(4)）。
|
||
基号 = v[:-8]（与应到表运单号对齐），单件 = 整串（每串即一件）。"""
|
||
res = defaultdict(set)
|
||
for v in df["运单号"]:
|
||
v = str(v).strip()
|
||
if len(v) > 8 and v[-4:].isdigit():
|
||
res[v[:-8]].add(v) # 以完整复合串作为“已到单号”存入
|
||
return res
|
||
|
||
|
||
def arrived_pieces_by_cols(wb_col, piece_col):
|
||
"""顺心 / 韵达 / 安能：按干净运单列分组，单件 = 子单号 / 扫描单号。
|
||
wb_col：实到表中与应到运单号对齐的干净列
|
||
（顺心=运单号 / 韵达=主单号 / 安能=所属单号）
|
||
piece_col：实到表中每件货物的单号列（子单号 / 扫描单号）"""
|
||
|
||
def parse(df):
|
||
res = defaultdict(set)
|
||
for m, s in zip(df[wb_col], df[piece_col]):
|
||
m, s = str(m).strip(), str(s).strip()
|
||
if m and s:
|
||
res[m].add(s)
|
||
return res
|
||
|
||
return parse
|
||
|
||
|
||
STATIONS = [
|
||
{
|
||
"name": "中通",
|
||
"exp": "中通-应到货物数据.xlsx",
|
||
"act": "中通-实到货物数据.xlsx",
|
||
"exp_qty": "交接件数", # 应到件数口径：交接件数（非录单件数）
|
||
"exp_wb": "运单号", # 应到表运单号列（兼作去重键）
|
||
"exp_jd": "交接单号", # 未到数据需展示的交接单号
|
||
"arrived_pieces": arrived_pieces_zhongtong,
|
||
"columns": ["交接单号", "运单号", "总件数"],
|
||
},
|
||
{
|
||
"name": "顺心",
|
||
"exp": "顺心-应到货物数据.xlsx",
|
||
"act": "顺心-实到货物数据.xlsx",
|
||
"exp_qty": "交接件数",
|
||
"exp_wb": "运单号",
|
||
"exp_jd": "交接单号",
|
||
"arrived_pieces": arrived_pieces_by_cols("运单号", "子单号"),
|
||
"columns": ["交接单号", "运单号", "总件数"],
|
||
},
|
||
{
|
||
"name": "韵达",
|
||
"exp": "韵达-应到货物数据.xlsx",
|
||
"act": "韵达-实到货物数据.xlsx",
|
||
"exp_qty": "交接件数",
|
||
"exp_wb": "运单号",
|
||
"exp_jd": "交接单号",
|
||
"arrived_pieces": arrived_pieces_by_cols("主单号", "子单号"),
|
||
"columns": ["交接单号", "运单号", "总件数"],
|
||
},
|
||
{
|
||
"name": "安能",
|
||
"exp": "安能-应到货物数据.xlsx",
|
||
"act": "安能-实到货物数据.xlsx",
|
||
"exp_qty": "交接件数",
|
||
"exp_wb": "运单号",
|
||
"exp_jd": "交接单号",
|
||
"arrived_pieces": arrived_pieces_by_cols("所属单号", "扫描单号"),
|
||
"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[cfg["exp_wb"]].duplicated().sum())
|
||
df_exp = df_exp.drop_duplicates(subset=[cfg["exp_wb"]], keep="first")
|
||
|
||
# 应到件数（新口径）= 交接件数 之和；记录 运单 -> (交接单号, 应到件数)
|
||
exp_by_wb = {}
|
||
exp_pieces = 0
|
||
for _, r in df_exp.iterrows():
|
||
wb = str(r[cfg["exp_wb"]]).strip()
|
||
if not wb:
|
||
continue
|
||
try:
|
||
n = int(float(r[cfg["exp_qty"]]))
|
||
except (TypeError, ValueError, KeyError):
|
||
n = 0
|
||
if n <= 0:
|
||
continue
|
||
exp_pieces += n
|
||
if wb not in exp_by_wb:
|
||
exp_by_wb[wb] = {
|
||
"jd": str(r.get(cfg["exp_jd"], "")).strip(),
|
||
"n": n,
|
||
}
|
||
|
||
# 实到件数（新口径）= 实到表单号去重数量（分组 运单->已到单号集合）
|
||
arrived = cfg["arrived_pieces"](df_act)
|
||
act_pieces = sum(len(s) for s in arrived.values()) # 全局去重单号数
|
||
|
||
# 未到：逐运单比较，列出实际已到的单号（顺序号乱序，无法反推缺件序号）
|
||
rows = []
|
||
full_miss = part_miss = 0
|
||
max_arrived = 0
|
||
for wb, info in exp_by_wb.items():
|
||
n = info["n"]
|
||
arrived_set = arrived.get(wb, set())
|
||
arrived_cnt = len(arrived_set)
|
||
if arrived_cnt >= n:
|
||
continue # 足额或溢到，不进未到表
|
||
if arrived_cnt == 0:
|
||
full_miss += 1
|
||
else:
|
||
part_miss += 1
|
||
max_arrived = max(max_arrived, arrived_cnt)
|
||
row = {
|
||
cfg["exp_jd"]: info["jd"],
|
||
cfg["exp_wb"]: wb,
|
||
"总件数": n,
|
||
}
|
||
for i, piece in enumerate(sorted(arrived_set, key=lambda x: str(x))):
|
||
row[f"已到单号{i+1}"] = piece
|
||
rows.append(row)
|
||
|
||
# 动态列：基础 3 列 + 已到单号1..max_arrived
|
||
columns = list(cfg["columns"]) + [f"已到单号{i+1}" for i in range(max_arrived)]
|
||
|
||
stats = {
|
||
"运单数": len(exp_by_wb),
|
||
"应到件": exp_pieces,
|
||
"已到件": act_pieces,
|
||
"未到件": max(0, exp_pieces - act_pieces),
|
||
"涉及运单": full_miss + part_miss,
|
||
"完全未到": full_miss,
|
||
"部分未到": part_miss,
|
||
"重复运单": dup,
|
||
}
|
||
return 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.get(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.get(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)
|
||
|
||
# 应到/实到基数取自「扫描综合查询」应扫/已扫（到/接件扫描→当日），
|
||
# 由 baishi_download_undelivered_data_impl 在同次导航里抓取并落 site_settings。
|
||
# 未抓取过则 get_setting 返回 "" → 视为无基数（报表显示「—」）。
|
||
from inbound_verify import (
|
||
state_store,
|
||
) # 与 _read_business_dates 一致：比对模块纯离线，懒加载
|
||
|
||
def _to_int(v):
|
||
v = (v or "").strip().replace(",", "")
|
||
try:
|
||
return int(float(v)) if v not in ("", "-") else None
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
exp_n = _to_int(state_store.get_setting("百世", "scan_expected_pieces"))
|
||
arr_n = _to_int(state_store.get_setting("百世", "scan_arrived_pieces"))
|
||
|
||
stats = {
|
||
"运单数": wb_count,
|
||
"应到件": exp_n,
|
||
"已到件": arr_n,
|
||
"未到件": 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 _read_business_dates(include):
|
||
"""从状态库读各站业务日期（dispatch 下载成功时快照写入），供报告「数据日期」列。
|
||
4 站取 expected_business_date（报告按应到口径）；百世取 undelivered_business_date。
|
||
从未下过的站返回空串（诚实留空，不反推）。"""
|
||
from inbound_verify 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} 供日志。"""
|
||
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"),
|
||
dates=dates or {},
|
||
)
|
||
wb.save(OUTFILE)
|
||
return {n: (s["未到件"] if s else None) for (n, s) in summary}
|
||
|
||
|
||
# ============================ 写汇总报表 ============================
|
||
|
||
|
||
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)
|
||
|
||
# 合计/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
|
||
# 列宽按「4 个 KPI 卡等宽」设计：B+C = D+E+F = G+H = I+J = 26
|
||
for col, w in {
|
||
"B": 12,
|
||
"C": 14,
|
||
"D": 9,
|
||
"E": 9,
|
||
"F": 8,
|
||
"G": 12,
|
||
"H": 14,
|
||
"I": 13,
|
||
"J": 13,
|
||
}.items():
|
||
ws.column_dimensions[col].width = w
|
||
ws.row_dimensions[1].height = 6
|
||
|
||
# —— 标题栏 ——
|
||
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:J2"]:
|
||
for c in row:
|
||
c.fill = PatternFill("solid", fgColor=NAVY)
|
||
ws.row_dimensions[2].height = 34
|
||
ws.merge_cells("B3:J3")
|
||
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%"),
|
||
]
|
||
# 2-3-2-2 分布填满 B-J（9 列），配合上方列宽使 4 卡视觉等宽
|
||
spans = [
|
||
("B5:C5", "B6:C6"),
|
||
("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)
|
||
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: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:J8"]:
|
||
for c in row:
|
||
c.border = Border(bottom=Side(style="medium", color=BLUE))
|
||
ws.row_dimensions[8].height = 22
|
||
|
||
# —— 统计表头 ——
|
||
headers = [
|
||
"站点",
|
||
"应到运单数",
|
||
"应到件数",
|
||
"已到件数",
|
||
"未到件数",
|
||
"未到率",
|
||
"完全未到运单",
|
||
"部分未到运单",
|
||
"数据日期",
|
||
]
|
||
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 = head_align
|
||
cell.border = BORDER
|
||
ws.row_dimensions[9].height = 30
|
||
|
||
# —— 各站数据行（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:
|
||
# 百世：未到明细已知；若已抓取应到/实到基数（扫描综合查询应扫/已扫）则填真实值
|
||
if s["应到件"] is not None and s["已到件"] is not None:
|
||
srate = (s["未到件"] / s["应到件"]) if s["应到件"] else 0
|
||
vals = [
|
||
name,
|
||
s["运单数"],
|
||
s["应到件"],
|
||
s["已到件"],
|
||
s["未到件"],
|
||
srate,
|
||
"—",
|
||
"—",
|
||
]
|
||
else:
|
||
vals = [name, s["运单数"], "—", "—", s["未到件"], "—", "—", "—"]
|
||
else:
|
||
srate = (s["未到件"] / s["应到件"]) if s["应到件"] else 0
|
||
vals = [
|
||
name,
|
||
s["运单数"],
|
||
s["应到件"],
|
||
s["已到件"],
|
||
s["未到件"],
|
||
srate,
|
||
s["完全未到"],
|
||
s["部分未到"],
|
||
]
|
||
vals.append(dates.get(name, "")) # 末列：该站业务日期
|
||
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 isinstance(v, (int, float))
|
||
and not isinstance(v, bool)
|
||
):
|
||
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.legend.overlay = False # 不覆盖绘图区：图例独占底部一行，与 X 轴站点名错开
|
||
chart.height = 9
|
||
chart.width = 20
|
||
ws.add_chart(chart, f"B{chart_anchor}")
|
||
|
||
# —— 口径说明 ——
|
||
note_row = chart_anchor + 19
|
||
notes = [
|
||
"指标口径：未到率 ＝ 未到件数 ÷ 应到件数；完全未到运单 ＝ 整单零到货；部分未到运单 ＝ 部分到货、部分缺件。",
|
||
"合计 / 图表仅含 4 站（顺心/中通/韵达/安能，应到−实到口径）；百世应到/实到取自「扫描综合查询」应扫/已扫（到/接件扫描→当日），已填入百世行，但为保持 4 站口径一致、不计入合计与图表。",
|
||
"本次下载失败的站点标注为（无数据）并计 0，不影响其余站点统计。",
|
||
"明细见各站点工作表；未到明细仅列短少运单，并列出该运单实际扫到的单号（已到单号1…），缺件不再编造子单号。",
|
||
"数据日期：各站本次纳入数据对应的业务日期（＝应到数据下载日 − 日期偏移；韵达偏移 1 为前一日）；合计为多站混合、不标注。",
|
||
]
|
||
for k, text in enumerate(notes):
|
||
rr = note_row + k
|
||
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)
|
||
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:J{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
|
||
dates = _read_business_dates(include)
|
||
undel = build_full_report(include, dates=dates)
|
||
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()
|