feat(store): persist 百世 daily basis to PG (baishi_daily_stats)

百世应到/实到基数（应扫/已扫）原仅在 state_store（单值、无历史）。新增百世专用聚合表 baishi_daily_stats（site+business_date UPSERT），baishi 下载时抓到基数直接落库（store.upsert_baishi_daily_stats，一步，不绕 state_store→store）。state_store 双写保留以兼容旧 Excel 汇总（process_baishi），后续统一清理。

真机端到端验证通过：PG (2026-07-31, 194, 186, 8)，state_store 一致。

Co-Authored-By: Claude <noreply@anthropic.com>

inbound_verify/sites/baishi.py

@@ -200,6 +200,11 @@ def baishi_download_undelivered_data_impl(page):
             if _exp_n > 0:
                 state_store.set_setting("百世", "scan_expected_pieces", str(_exp_n))
                 state_store.set_setting("百世", "scan_arrived_pieces", str(_arr_n))
+                from inbound_verify import (
+                    store,
+                )  # 直接落库 PG（一步，不绕 state_store→store）
+
+                store.upsert_baishi_daily_stats(_exp_n, _arr_n)
                 print(f"   ℹ️ 已记录百世应到/实到基数：应扫 {_exp_n} / 已扫 {_arr_n}")
         except Exception as _e:
             # 抓取失败绝不影响未到明细下载主流程

inbound_verify/store.py

@@ -268,6 +268,18 @@ _SQL_UNDELIVERED = """
         ingested_at = now()
 """
 
+_SQL_BAISHI_DAILY_STATS = """
+    INSERT INTO baishi_daily_stats
+        (site, business_date, expected_pieces, arrived_pieces, undelivered_pieces, raw)
+    VALUES (%s,%s,%s,%s,%s,%s)
+    ON CONFLICT (site, business_date) DO UPDATE SET
+        expected_pieces    = COALESCE(EXCLUDED.expected_pieces, baishi_daily_stats.expected_pieces),
+        arrived_pieces     = COALESCE(EXCLUDED.arrived_pieces, baishi_daily_stats.arrived_pieces),
+        undelivered_pieces = COALESCE(EXCLUDED.undelivered_pieces, baishi_daily_stats.undelivered_pieces),
+        raw                = EXCLUDED.raw,
+        ingested_at        = now()
+"""
+
 
 # ============================== 入库 ==============================
 
@@ -369,6 +381,34 @@ def _ingest_undelivered_baishi(cur):
     return len(rows)
 
 
+def upsert_baishi_daily_stats(exp, arr, business_date=None):
+    """直接落库百世当日应到/实到基数（应扫/已扫，站级日聚合）。
+    供 baishi 下载时抓到基数后直接调用（一步落库，不绕 state_store→store）。
+    business_date 默认今天（百世固定当天）。best-effort：失败只告警，不影响下载流程。"""
+    biz = business_date or date.today()
+    if exp is None and arr is None:
+        return
+    undel = (exp - arr) if (exp is not None and arr is not None) else None
+    try:
+        with _connect(_load_pg_config()["dbname"]) as conn:
+            with conn.cursor() as cur:
+                cur.execute(
+                    _SQL_BAISHI_DAILY_STATS,
+                    (
+                        "百世",
+                        biz,
+                        exp,
+                        arr,
+                        undel,
+                        Jsonb({"expected": exp, "arrived": arr, "undelivered": undel}),
+                    ),
+                )
+            conn.commit()
+        print(f"   [基数] 百世 {biz}: 应扫 {exp} / 已扫 {arr} / 未扫 {undel}")
+    except Exception as e:
+        print(f"   [基数] 百世 {biz} 入库失败（不影响下载）: {e}")
+
+
 def ingest(site=None):
     """入库：指定 site 则单站（百世只入未到），否则全站。返回总条数。"""
     dates = _read_business_dates()

schema.sql

@@ -53,3 +53,16 @@ CREATE TABLE IF NOT EXISTS undelivered_record (
     ingested_at  TIMESTAMPTZ  NOT NULL DEFAULT now(),
     UNIQUE (site, piece_no)
 );
+
+-- 百世日聚合（应扫/已扫基数：站级日聚合，区别于运单级/件级/子单级表）
+CREATE TABLE IF NOT EXISTS baishi_daily_stats (
+    id                 BIGSERIAL    PRIMARY KEY,
+    site               TEXT         NOT NULL,   -- 百世
+    business_date      DATE         NOT NULL,   -- 业务日期（百世固定当天）
+    expected_pieces    INTEGER,                  -- 应扫（应到基数）
+    arrived_pieces     INTEGER,                  -- 已扫（实到基数）
+    undelivered_pieces INTEGER,                  -- 未扫（=应扫-已扫，任一缺失则 NULL）
+    raw                JSONB        NOT NULL,    -- 原始抓取值
+    ingested_at        TIMESTAMPTZ  NOT NULL DEFAULT now(),
+    UNIQUE (site, business_date)
+);