diff --git a/inbound_verify/db_compare.py b/inbound_verify/db_compare.py index 001f93e..f147e5a 100644 --- a/inbound_verify/db_compare.py +++ b/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( diff --git a/inbound_verify/runtime.py b/inbound_verify/runtime.py index b0bd9cd..e2e0882 100644 --- a/inbound_verify/runtime.py +++ b/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)) diff --git a/inbound_verify/store.py b/inbound_verify/store.py index 14a22e4..7b19580 100644 --- a/inbound_verify/store.py +++ b/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 - handover_no = EXCLUDED.handover_no, + ON CONFLICT (site, handover_no, waybill_no) DO UPDATE SET 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": diff --git a/schema.sql b/schema.sql index fba8f53..02d86bd 100644 --- a/schema.sql +++ b/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);