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 <noreply@anthropic.com>

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)

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 / STATIONS / _site_cfg / BAISHI_FILE 等）见 domain.py。
-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):

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(
+        stale = os.path.join(DOWNLOAD_DIR, SITE_UNDELIVERED_FILE.format(name=site))
-            DOWNLOAD_DIR, expected_undelivered.SITE_UNDELIVERED_FILE.format(name=site)
-        )
         if os.path.exists(stale):
             os.remove(stale)
         return False

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 = []