From 91ef8970b835baa5a9b628ab498f2a9ec545b252 Mon Sep 17 00:00:00 2001 From: Misaka_Company Date: Fri, 24 Jul 2026 08:38:18 +0800 Subject: [PATCH] refactor: extract domain.py (shared site/file/colmap config) Move ALL_REPORT_SITES / SITE_UNDELIVERED_FILE / BAISHI_FILE / BAISHI_COLUMNS / arrived_pieces_* / STATIONS / _site_cfg out of expected_undelivered into a new leaf module inbound_verify/domain.py. store.py and runtime.py now read site/file config from domain directly instead of through the compare engine (store keeps eu only for _read_business_dates). Behavior identical. Co-Authored-By: Claude --- inbound_verify/domain.py | 92 ++++++++++++++++++++++++ inbound_verify/expected_undelivered.py | 97 +++----------------------- inbound_verify/runtime.py | 5 +- inbound_verify/store.py | 14 ++-- 4 files changed, 114 insertions(+), 94 deletions(-) create mode 100644 inbound_verify/domain.py diff --git a/inbound_verify/domain.py b/inbound_verify/domain.py new file mode 100644 index 0000000..bf6dc3b --- /dev/null +++ b/inbound_verify/domain.py @@ -0,0 +1,92 @@ +# -*- coding: utf-8 -*- +"""domain.py — 站点 / 文件名 / 列映射的共享配置（单一来源）。 + +比对（compare）与入库（store）都依赖这套配置；抽出独立 leaf 模块， +让 store 不必为读配置而依赖整个比对引擎。纯数据，无 state_store / 文件 IO 依赖。 +""" + +from collections import defaultdict + +# 汇总报表覆盖的全部站点（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) diff --git a/inbound_verify/expected_undelivered.py b/inbound_verify/expected_undelivered.py index 38e0012..9629826 100644 --- a/inbound_verify/expected_undelivered.py +++ b/inbound_verify/expected_undelivered.py @@ -41,96 +41,21 @@ from openpyxl.worksheet.page import PageMargins from openpyxl.worksheet.properties import PageSetupProperties from inbound_verify.paths import DOWNLOAD_DIR, OUTPUT_DIR +from inbound_verify.domain import ( + ALL_REPORT_SITES, + BAISHI_COLUMNS, + BAISHI_FILE, + SITE_UNDELIVERED_FILE, + STATIONS, + _site_cfg, + arrived_pieces_by_cols, + arrived_pieces_zhongtong, +) # 比对报表输出文件（路径锚定统一走 paths.py）。 OUTFILE = os.path.join(OUTPUT_DIR, "应到未到数据.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) +# 站点 / 文件名 / 列映射配置（ALL_REPORT_SITES / STATIONS / _site_cfg / BAISHI_FILE 等）见 domain.py。 def process(name): diff --git a/inbound_verify/runtime.py b/inbound_verify/runtime.py index 7fb24ef..5700ece 100644 --- a/inbound_verify/runtime.py +++ b/inbound_verify/runtime.py @@ -22,6 +22,7 @@ import yaml from playwright.sync_api import sync_playwright from inbound_verify.paths import DOWNLOAD_DIR, CONFIG_PATH +from inbound_verify.domain import SITE_UNDELIVERED_FILE from inbound_verify import state_store from inbound_verify.sites import shunxin, baishi, zto, yunda, anneng @@ -444,9 +445,7 @@ def _site_undelivered_handler(site): ) 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) - ) + stale = os.path.join(DOWNLOAD_DIR, SITE_UNDELIVERED_FILE.format(name=site)) if os.path.exists(stale): os.remove(stale) return False diff --git a/inbound_verify/store.py b/inbound_verify/store.py index da18680..6ccead3 100644 --- a/inbound_verify/store.py +++ b/inbound_verify/store.py @@ -29,10 +29,14 @@ import yaml from psycopg.types.json import Jsonb from inbound_verify.paths import BASE_DIR, CONFIG_PATH, DOWNLOAD_DIR +from inbound_verify.domain import ( + BAISHI_FILE, + _site_cfg, +) # 站点 / 文件名配置（单一来源） from inbound_verify import ( expected_undelivered as eu, -) # 复用站点 / 文件名 / 列映射 / 基号口径（单一来源） +) # _read_business_dates（比对侧业务日期读取） SCHEMA_PATH = os.path.join(BASE_DIR, "schema.sql") @@ -261,7 +265,7 @@ _SQL_UNDELIVERED = """ def _ingest_expected(cur, site, business_date): """入库单站应到（运单级，按 waybill_no 去重 keep-first 后 UPSERT）。""" - cfg = eu._site_cfg(site) + cfg = _site_cfg(site) path = os.path.join(DOWNLOAD_DIR, cfg["exp"]) if not os.path.exists(path): print(f" [跳过] {site} 应到：文件不存在 {cfg['exp']}") @@ -293,7 +297,7 @@ def _ingest_expected(cur, site, business_date): def _ingest_actual(cur, site): """入库单站实到（扫描件级，按 piece_no UPSERT）。""" - cfg = eu._site_cfg(site) + cfg = _site_cfg(site) path = os.path.join(DOWNLOAD_DIR, cfg["act"]) if not os.path.exists(path): print(f" [跳过] {site} 实到：文件不存在 {cfg['act']}") @@ -332,9 +336,9 @@ def _ingest_actual(cur, site): def _ingest_undelivered_baishi(cur): """入库百世应到未到明细（子单级，按 (site, piece_no) UPSERT）。""" - path = os.path.join(DOWNLOAD_DIR, eu.BAISHI_FILE) + path = os.path.join(DOWNLOAD_DIR, BAISHI_FILE) if not os.path.exists(path): - print(f" [跳过] 百世 未到：文件不存在 {eu.BAISHI_FILE}") + print(f" [跳过] 百世 未到：文件不存在 {BAISHI_FILE}") return 0 df = pd.read_excel(path, dtype=str).fillna("") rows = []