refactor(ingest): unique key (site,handover_no,waybill_no) + business_date purge + shortfall-sum undelivered

- schema: expected_record unique key (site,waybill_no) -> (site,handover_no,waybill_no),
  so the same waybill can belong to multiple handover batches without being overwritten
- store: dedup by (handover_no,waybill_no) instead of waybill_no alone; UPSERT conflict
  key matches the new unique key; handover_no no longer force-overwritten on conflict
- store: add purge_expected, called on force re-download; deletes by business_date
  (download day, same source as task date) rather than batch_out_date, which is offset
  from the download day for sites like yunda where download day != outbound day
- store/runtime: thread force + target_date through ingest_task / _persist_to_db /
  _site_undelivered_handler so force re-download purges then re-ingests in one tx
- db_compare: undelivered_pieces now sums per-waybill shortfall (total - arrived)
  instead of expected_total - arrived_total, unaffected by unrelated/extra scans

inbound_verify/db_compare.py

@@ -427,6 +427,9 @@ def _do_compare(
             stats.sf_undelivered += 1
 
         max_arrived = max(max_arrived, arrived_cnt)
+        # 差缺累加：未到件数 = Σ(每个差缺运单的 应到件 − 实到件)，逐运单精确，
+        # 不受实到里无关运单/重复扫描干扰（替代旧的「应到总件 − 实到总行」全局减法）。
+        stats.undelivered_pieces += handover_pcs - arrived_cnt
         rows.append(
             UndeliveredRow(
                 handover_no=handover_no,
@@ -438,7 +441,6 @@ def _do_compare(
             )
         )
 
-    stats.undelivered_pieces = max(0, stats.expected_pieces - stats.arrived_pieces)
     stats.undelivered_wb = stats.full_miss + stats.part_miss
 
     result = CompareResult(

inbound_verify/runtime.py

@@ -506,7 +506,7 @@ def _site_undelivered_handler(site):
 
             if store.ingest_enabled():
                 store.ingest_task(
-                    site, "undelivered"
+                    site, "undelivered", force=force, target_date=date
                 )  # 4 站 = ingest expected + actual
                 print(f">> [入库] {site} 前置入库完成")
         except Exception as e:
@@ -693,10 +693,11 @@ def _refresh_ready(site):
     _apply_ready(site, flags, dates)
 
 
-def _persist_to_db(site, kind):
+def _persist_to_db(site, kind, force=False, target_date=None):
     """下载成功后把本次数据入库 PostgreSQL（尽力而为，绝不外抛，不影响任务判定）。
     - __compare__ 无源数据，跳过。
     - auto_ingest=false 时跳过（无 PG/cpolar 的开发机）。
+    - force + target_date：应到类先按 business_date 清空本次下载日再整批重入（防反复入库归属漂移）。
     - 懒导入 store 以回避 import 顺序（store↔compare 与 runtime↔compare 共存）。
     - 结果写 state_store.ingest_state，供 /api/status 反映入库健康。
     所有写库/写状态都包 try/except：失败仅告警，绝不改变 dispatch_task 的 SUCCESS 判定。"""
@@ -711,7 +712,7 @@ def _persist_to_db(site, kind):
         if not store.ingest_enabled():  # 移入 try：config.yaml 缺失/损坏时也不外抛
             print(">> [入库] 已关闭 (auto_ingest=false)，跳过")
             return
-        count = store.ingest_task(site, kind)
+        count = store.ingest_task(site, kind, force=force, target_date=target_date)
         # 4 站 undelivered 连带入了 expected+actual：按实际入库的类补记 ingest_state，
         # 否则心跳派生 ready（expected ∧ actual → undelivered）会读到陈旧值。
         logged = (
@@ -755,7 +756,12 @@ def dispatch_task(ctx, task_spec):
         if ret is False:
             return (state_store.TASK_FAILED, "任务执行失败（重试耗尽）")
         _record_business_date(site, kind, task_spec.get("date"))
-        _persist_to_db(site, kind)
+        _persist_to_db(
+            site,
+            kind,
+            force=bool(task_spec.get("force", False)),
+            target_date=task_spec.get("date"),
+        )
         return (state_store.TASK_SUCCESS, None)
     except Exception as e:
         return (state_store.TASK_FAILED, str(e))

inbound_verify/store.py

@@ -263,8 +263,7 @@ _SQL_EXPECTED = """
         (site, waybill_no, handover_no, handover_pieces, order_pieces,
          business_date, out_date, batch_out_date, raw)
     VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)
-    ON CONFLICT (site, waybill_no) DO UPDATE SET
+    ON CONFLICT (site, handover_no, waybill_no) DO UPDATE SET
-        handover_no     = EXCLUDED.handover_no,
         handover_pieces = EXCLUDED.handover_pieces,
         order_pieces    = EXCLUDED.order_pieces,
         business_date   = COALESCE(EXCLUDED.business_date, expected_record.business_date),
@@ -314,15 +313,31 @@ _SQL_BAISHI_DAILY_STATS = """
 # ============================== 入库 ==============================
 
 
+def purge_expected(cur, site, target_date):
+    """删除某站某次下载（business_date = 下载目标日）的全部应到记录。
+
+    force 重下/整批重入前调用，清掉旧数据避免反复入库时运单归属漂移。
+    用 business_date（下载日，与任务 date 同源、单链路可靠）而非 batch_out_date
+    （数据归属日，对韵达等"下载日≠出库日"站点与下载日错开，会删错天）。
+    与 UPSERT 在同一事务/同一 cursor 内执行，失败一并回滚。返回删除行数。
+    """
+    cur.execute(
+        "DELETE FROM expected_record WHERE site=%s AND business_date=%s",
+        (site, target_date),
+    )
+    return cur.rowcount
+
+
 def _ingest_expected(cur, site, business_date):
-    """入库单站应到（运单级，按 waybill_no 去重 keep-first 后 UPSERT）。"""
+    """入库单站应到（运单级，按 交接单号+运单号 去重 keep-first 后 UPSERT）。"""
     cfg = _site_cfg(site)
     path = os.path.join(DOWNLOAD_DIR, cfg["exp"])
     if not os.path.exists(path):
         print(f"   [跳过] {site} 应到：文件不存在 {cfg['exp']}")
         return 0
     df = pd.read_excel(path, dtype=str).fillna("")
-    df = df.drop_duplicates(subset=[cfg["exp_wb"]], keep="first")
+    # 按 交接单号+运单号 去重：同运单可跨批次（不同交接单号），但同批次内运单唯一。
+    df = df.drop_duplicates(subset=[cfg["exp_jd"], cfg["exp_wb"]], keep="first")
     biz = _parse_date(business_date)
     out_col = "出库时间"
     has_out_col = out_col in df.columns
@@ -471,7 +486,7 @@ def ingest(site=None):
     return total
 
 
-def ingest_task(site, kind):
+def ingest_task(site, kind, force=False, target_date=None):
     """按 (site, kind) 入库本次刚下载的文件（幂等 UPSERT），返回总条数。
     与 ingest(site) 的区别：只入本次刷新的那一类，避免重读写另一类文件（同步钩子里减少阻塞）。
     kind 路由：
@@ -479,6 +494,9 @@ def ingest_task(site, kind):
       undelivered 百世        入未到；
       undelivered 4 站        _site_undelivered_handler 内部连带下了 expected+actual，故入两者；
       __compare__ / 其它组合  返回 0。
+
+    force + target_date（YYYY-MM-DD）：应到类先按 batch_out_date 清空目标日，再整批重入，
+    避免反复入库时同运单跨批次导致归属漂移。删除与 UPSERT 在同一事务内，失败一并回滚。
     """
     if site == "__compare__":
         return 0
@@ -490,6 +508,15 @@ def ingest_task(site, kind):
     total = 0
     with _connect(_load_pg_config()["dbname"]) as conn:
         with conn.cursor() as cur:
+            # force 重下：应到类先清空本次下载日（business_date 口径），再整批重入
+            if (
+                force
+                and target_date
+                and kind in ("expected", "undelivered")
+                and site != "百世"
+            ):
+                purged = purge_expected(cur, site, target_date)
+                print(f">> [入库] force 清空 {site} {target_date} 应到: {purged} 行")
             if kind == "expected":
                 total += _ingest_expected(cur, site, dates.get(site))
             elif kind == "actual":

schema.sql

@@ -10,11 +10,11 @@
 CREATE SCHEMA IF NOT EXISTS inbound_verify;
 SET search_path TO inbound_verify;
 
--- 应到货物（运单级：一运单一行；按运单号去重 keep-first 后入库）
+-- 应到货物（运单级：一运单一批次一行；同运单可跨批次，按 交接单号+运单号 去重后入库）
 CREATE TABLE IF NOT EXISTS expected_record (
     id              BIGSERIAL    PRIMARY KEY,
     site            TEXT         NOT NULL,   -- 站点：顺心 / 中通 / 韵达 / 安能
-    waybill_no      TEXT         NOT NULL,   -- 运单基号（业务唯一键，去重键）
+    waybill_no      TEXT         NOT NULL,   -- 运单基号
     handover_no     TEXT,                     -- 交接单号
     handover_pieces INTEGER,                  -- 交接件数（应到件数口径）
     order_pieces    INTEGER,                  -- 录单件数
@@ -23,8 +23,22 @@ CREATE TABLE IF NOT EXISTS expected_record (
     batch_out_date  DATE,                     -- 批次归属日（同交接单号出库日众数，并列取较早）
     raw             JSONB        NOT NULL,    -- 站点原始全列（key=原列名）
     ingested_at     TIMESTAMPTZ  NOT NULL DEFAULT now(),
-    UNIQUE (site, waybill_no)
+    UNIQUE (site, handover_no, waybill_no)
 );
+-- 迁移：旧唯一键 (site, waybill_no) → 新 (site, handover_no, waybill_no)。
+-- CREATE TABLE IF NOT EXISTS 不改已存在的表，故用幂等 ALTER 切换约束。
+ALTER TABLE expected_record DROP CONSTRAINT IF EXISTS expected_record_site_waybill_no_key;
+DO $$
+BEGIN
+    IF NOT EXISTS (
+        SELECT 1 FROM pg_constraint
+        WHERE conname = 'expected_record_site_handover_waybill_key'
+    ) THEN
+        ALTER TABLE expected_record
+            ADD CONSTRAINT expected_record_site_handover_waybill_key
+            UNIQUE (site, handover_no, waybill_no);
+    END IF;
+END $$;
 CREATE INDEX IF NOT EXISTS idx_expected_site_date ON expected_record (site, business_date);
 CREATE INDEX IF NOT EXISTS idx_expected_handover ON expected_record (site, handover_no);
 CREATE INDEX IF NOT EXISTS idx_expected_out_date ON expected_record (site, out_date);