引入 PostgreSQL 持久化：到货核销数据入库管道

- 新增 db_store.py：解析 downloads/ 下各站应到/实到/未到 Excel，幂等 UPSERT 入库（统一核心列 + raw JSONB 兜底，单号按文本存）
- 新增 schema.sql：三表（expected/actual/undelivered_record），隔离到专用 schema inbound_verify（CREATE SCHEMA + search_path）
- 韵达实到业务清洗：抛弃「交接单号」为空行 + 按子单号去重
- 同步 config.example.yaml（postgres 配置）与 requirements.txt（psycopg[binary]）

本次仅手动入库管道；自动挂载到下载流程留待后续。

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

config.example.yaml

@@ -68,3 +68,22 @@ anneng:
   # 安能 Electron 应用的可执行文件路径。
   # 路径含反斜杠/空格/@，用单引号包裹即可（YAML 单引号串按字面解析）。
   app_path: 'D:\SoftWare\SoftWare Installation\@ane-electron-uiapp\安能全网门户.exe'
+
+# ----------------------------------------------------------------------------
+# PostgreSQL 数据持久化（到货核销数据入库，详见 db_store.py）
+# ----------------------------------------------------------------------------
+# createdb 会连接名为 postgres 的维护库来创建下方 dbname 指定的数据库。
+# 命令行：
+#   python db_store.py createdb        创建数据库（幂等）
+#   python db_store.py init            建表（幂等）
+#   python db_store.py ingest [site]   入库全站或单站（幂等 UPSERT）
+#   python db_store.py all             createdb → init → 全站 ingest
+postgres:
+  host: 127.0.0.1
+  port: 5432
+  user: postgres
+  # 连接密码；config.yaml 已 gitignore，真实凭据只写在那里，切勿提交示例值。
+  password: "YOUR_PASSWORD_HERE"
+  dbname: CQHXDB
+  # 承载到货核销表的专用 schema（隔离 public）；表建在此 schema 下。
+  schema: inbound_verify

db_store.py

@@ -0,0 +1,398 @@
+# -*- coding: utf-8 -*-
+"""
+db_store.py — 到货核销数据持久化（PostgreSQL）
+
+职责：把 downloads/ 下各站点下载的应到 / 实到 / 未到 Excel 解析后，幂等写入 PostgreSQL。
+与下载流程解耦：本模块只读 downloads/ 现有文件入库，不关心谁触发下载、下载了几次。
+
+设计要点：
+  - 三张表：expected_record（运单级）/ actual_record（扫描件级）
+            / undelivered_record（百世站点直供未到明细，子单级）
+  - 每行 = 统一核心列 + raw JSONB（站点原始全列，key=原列名，一字段不丢）
+  - 幂等：业务唯一键 UPSERT，重复下载天然合并、零冗余
+  - 单号一律按文本读写（dtype=str），防长数字被科学计数 / 精度丢失
+
+命令行：
+  python db_store.py createdb        创建数据库（幂等）
+  python db_store.py init            建表（幂等 CREATE TABLE IF NOT EXISTS）
+  python db_store.py ingest [site]   入库全站或单站（幂等 UPSERT）
+  python db_store.py all             createdb → init → 全站 ingest 一条龙
+"""
+
+import os
+import sys
+from datetime import date, datetime
+
+import pandas as pd
+import psycopg
+import yaml
+from psycopg.types.json import Jsonb
+
+from paths import BASE_DIR, CONFIG_PATH, DOWNLOAD_DIR
+
+import expected_undelivered as eu  # 复用站点 / 文件名 / 列映射 / 基号口径（单一来源）
+
+SCHEMA_PATH = os.path.join(BASE_DIR, "schema.sql")
+
+# 有应到 / 实到的 4 站（百世只有站点直供的未到明细，单独处理）
+ALL_SITES = ["顺心", "中通", "韵达", "安能"]
+
+
+# ============================== 配置 / 连接 ==============================
+
+
+def _load_pg_config():
+    """从 config.yaml 读 postgres 段；缺失项给默认。"""
+    if not os.path.exists(CONFIG_PATH):
+        raise FileNotFoundError(
+            f"未找到配置文件 {CONFIG_PATH}（请参考 config.example.yaml 创建 config.yaml）"
+        )
+    with open(CONFIG_PATH, "r", encoding="utf-8") as f:
+        cfg = yaml.safe_load(f) or {}
+    pg = cfg.get("postgres") or {}
+    return {
+        "host": pg.get("host", "127.0.0.1"),
+        "port": int(pg.get("port", 5432)),
+        "user": pg.get("user", "postgres"),
+        "password": pg.get("password", ""),
+        "dbname": pg.get("dbname", "CQHXDB"),
+        "schema": pg.get("schema", "inbound_verify"),
+    }
+
+
+def _connect(dbname):
+    """用关键字参数连接（避开 conninfo 对密码特殊字符的解析）。
+    options 设 search_path 到专用 schema，使无 schema 限定的表名解析到该 schema。"""
+    c = _load_pg_config()
+    return psycopg.connect(
+        host=c["host"],
+        port=c["port"],
+        dbname=dbname,
+        user=c["user"],
+        password=c["password"],
+        options=f"-c search_path={c['schema']}",
+    )
+
+
+# ============================== 建库 / 建表 ==============================
+
+
+def create_database():
+    """连接维护库 postgres，创建目标数据库（幂等）。"""
+    c = _load_pg_config()
+    target = c["dbname"]
+    with _connect("postgres") as conn:  # autocommit：CREATE DATABASE 不能在事务里
+        conn.autocommit = True
+        with conn.cursor() as cur:
+            cur.execute("SELECT 1 FROM pg_database WHERE datname = %s", (target,))
+            if cur.fetchone():
+                print(f">> [db] 数据库 {target} 已存在，跳过创建")
+                return
+            cur.execute(f'CREATE DATABASE "{target}"')
+            print(f">> [db] 已创建数据库 {target}")
+
+
+def init_schema():
+    """在目标库执行 schema.sql（幂等）。"""
+    with open(SCHEMA_PATH, "r", encoding="utf-8") as f:
+        sql = f.read()
+    with _connect(_load_pg_config()["dbname"]) as conn:
+        with conn.cursor() as cur:
+            cur.execute(sql)
+        conn.commit()
+    print(">> [db] 表结构已就绪")
+
+
+# ============================== 解析辅助 ==============================
+
+# 实到单号列 / 基号列映射（口径取自 expected_undelivered）：
+#   piece     = 实到表里「每件」的单号列（扫描单号 / 子单号 / 复合串）
+#   waybill   = 与应到运单号对齐的干净列（中通无干净列，由 piece 复合串 v[:-8] 推导）
+#   scan_time = 扫描时间列（缺失则不填，原始值仍在 raw）
+#   scan_site = 扫描网点列
+ACTUAL_COLMAP = {
+    "中通": {
+        "piece": "运单号",
+        "waybill": None,
+        "scan_time": "扫描时间",
+        "scan_site": "扫描网点",
+    },
+    "顺心": {
+        "piece": "子单号",
+        "waybill": "运单号",
+        "scan_time": "操作时间",
+        "scan_site": "操作网点",
+    },
+    "韵达": {
+        "piece": "子单号",
+        "waybill": "主单号",
+        "scan_time": "扫描时间",
+        "scan_site": "扫描站点",
+    },
+    "安能": {
+        "piece": "扫描单号",
+        "waybill": "所属单号",
+        "scan_time": "扫描时间",
+        "scan_site": "扫描网点",
+    },
+}
+
+_TIME_FMTS = (
+    "%Y-%m-%d %H:%M:%S",
+    "%Y-%m-%d %H:%M",
+    "%Y/%m/%d %H:%M:%S",
+    "%Y/%m/%d %H:%M",
+    "%Y-%m-%d",
+    "%Y/%m/%d",
+)
+
+
+def _to_int(v):
+    """尽力把单元格转 int（件数）；空 / 非数返回 None。"""
+    s = str(v).strip().replace(",", "")
+    if s in ("", "-", "nan", "None"):
+        return None
+    try:
+        return int(float(s))
+    except (TypeError, ValueError):
+        return None
+
+
+def _parse_time(v):
+    """尽力解析多种时间格式为 datetime；失败返回 None（原始值在 raw 里）。"""
+    if v is None:
+        return None
+    if isinstance(v, datetime):
+        return v
+    s = str(v).strip()
+    if not s or s in ("nan", "NaT"):
+        return None
+    for fmt in _TIME_FMTS:
+        try:
+            return datetime.strptime(s, fmt)
+        except ValueError:
+            continue
+    try:  # 兜底：交给 pandas 推断
+        return pd.to_datetime(s).to_pydatetime()
+    except Exception:
+        return None
+
+
+def _parse_date(v):
+    if not v:
+        return None
+    try:
+        return datetime.strptime(str(v).strip(), "%Y-%m-%d").date()
+    except (ValueError, TypeError):
+        return None
+
+
+def _raw_row(row):
+    """把一行原始记录转成 JSONB 兼容 dict（key=原列名；丢空值，保留全部有值字段）。"""
+    out = {}
+    for k, v in row.items():
+        if v is None:
+            continue
+        if isinstance(v, float) and pd.isna(v):
+            continue
+        s = str(v).strip()
+        if s == "":
+            continue
+        if isinstance(v, (datetime, date)):
+            out[str(k)] = v.isoformat()
+        else:
+            out[str(k)] = v
+    return out
+
+
+def _read_business_dates():
+    """从状态库读各站本次业务日期（与报告口径一致；读不到返回空 dict）。"""
+    try:
+        return eu._read_business_dates(ALL_SITES + ["百世"]) or {}
+    except Exception as e:
+        print(f">> [warn] 读取业务日期失败（不影响入库）: {e}")
+        return {}
+
+
+# ============================== UPSERT SQL ==============================
+
+_SQL_EXPECTED = """
+    INSERT INTO expected_record
+        (site, waybill_no, handover_no, handover_pieces, order_pieces, business_date, raw)
+    VALUES (%s,%s,%s,%s,%s,%s,%s)
+    ON CONFLICT (site, 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),
+        raw             = EXCLUDED.raw,
+        ingested_at     = now()
+"""
+
+_SQL_ACTUAL = """
+    INSERT INTO actual_record
+        (site, waybill_no, piece_no, scan_time, scan_site, raw)
+    VALUES (%s,%s,%s,%s,%s,%s)
+    ON CONFLICT (site, piece_no) DO UPDATE SET
+        waybill_no  = COALESCE(EXCLUDED.waybill_no, actual_record.waybill_no),
+        scan_time   = EXCLUDED.scan_time,
+        scan_site   = EXCLUDED.scan_site,
+        raw         = EXCLUDED.raw,
+        ingested_at = now()
+"""
+
+_SQL_UNDELIVERED = """
+    INSERT INTO undelivered_record
+        (site, waybill_no, piece_no, biz_type, last_scan, raw)
+    VALUES (%s,%s,%s,%s,%s,%s)
+    ON CONFLICT (site, piece_no) DO UPDATE SET
+        waybill_no  = COALESCE(EXCLUDED.waybill_no, undelivered_record.waybill_no),
+        biz_type    = EXCLUDED.biz_type,
+        last_scan   = EXCLUDED.last_scan,
+        raw         = EXCLUDED.raw,
+        ingested_at = now()
+"""
+
+
+# ============================== 入库 ==============================
+
+
+def _ingest_expected(cur, site, business_date):
+    """入库单站应到（运单级，按 waybill_no 去重 keep-first 后 UPSERT）。"""
+    cfg = eu._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")
+    biz = _parse_date(business_date)
+    rows = []
+    for r in df.to_dict("records"):
+        wb = str(r.get(cfg["exp_wb"], "")).strip()
+        if not wb:
+            continue
+        rows.append(
+            (
+                site,
+                wb,
+                str(r.get(cfg["exp_jd"], "")).strip() or None,
+                _to_int(r.get(cfg["exp_qty"])),
+                _to_int(r.get("录单件数")),
+                biz,
+                Jsonb(_raw_row(r)),
+            )
+        )
+    if rows:
+        cur.executemany(_SQL_EXPECTED, rows)
+    print(f"   [应到] {site}：{len(rows)} 条运单")
+    return len(rows)
+
+
+def _ingest_actual(cur, site):
+    """入库单站实到（扫描件级，按 piece_no UPSERT）。"""
+    cfg = eu._site_cfg(site)
+    path = os.path.join(DOWNLOAD_DIR, cfg["act"])
+    if not os.path.exists(path):
+        print(f"   [跳过] {site} 实到：文件不存在 {cfg['act']}")
+        return 0
+    cm = ACTUAL_COLMAP[site]
+    df = pd.read_excel(path, dtype=str).fillna("")
+    if site == "韵达":
+        # 韵达业务清洗：抛弃「交接单号」为空的行（派件/签收等其他扫描无交接单号），
+        # 再按子单号去重（一件多扫只留一条；清洗后子单号已天然唯一，drop 为保险）。
+        df = df[df["交接单号"].astype(str).str.strip() != ""]
+        df = df.drop_duplicates(subset=[cm["piece"]], keep="last")
+    rows = []
+    for r in df.to_dict("records"):
+        piece = str(r.get(cm["piece"], "")).strip()
+        if not piece:
+            continue
+        if site == "中通":  # 复合串 H+运单号(12)+总数(4)+顺序(4)：基号 = v[:-8]
+            waybill = piece[:-8] if (len(piece) > 8 and piece[-4:].isdigit()) else piece
+        else:
+            waybill = str(r.get(cm["waybill"], "")).strip() or None
+        rows.append(
+            (
+                site,
+                waybill,
+                piece,
+                _parse_time(r.get(cm["scan_time"])),
+                str(r.get(cm["scan_site"], "")).strip() or None,
+                Jsonb(_raw_row(r)),
+            )
+        )
+    if rows:
+        cur.executemany(_SQL_ACTUAL, rows)
+    print(f"   [实到] {site}：{len(rows)} 条扫描")
+    return len(rows)
+
+
+def _ingest_undelivered_baishi(cur):
+    """入库百世应到未到明细（子单级，按 (site, piece_no) UPSERT）。"""
+    path = os.path.join(DOWNLOAD_DIR, eu.BAISHI_FILE)
+    if not os.path.exists(path):
+        print(f"   [跳过] 百世 未到：文件不存在 {eu.BAISHI_FILE}")
+        return 0
+    df = pd.read_excel(path, dtype=str).fillna("")
+    rows = []
+    for r in df.to_dict("records"):
+        rows.append(
+            (
+                "百世",
+                str(r.get("运单号", "")).strip() or None,
+                str(r.get("子单号", "")).strip() or None,
+                str(r.get("类型", "")).strip() or None,
+                str(r.get("最新扫描记录", "")).strip() or None,
+                Jsonb(_raw_row(r)),
+            )
+        )
+    if rows:
+        cur.executemany(_SQL_UNDELIVERED, rows)
+    print(f"   [未到] 百世：{len(rows)} 条")
+    return len(rows)
+
+
+def ingest(site=None):
+    """入库：指定 site 则单站（百世只入未到），否则全站。返回总条数。"""
+    dates = _read_business_dates()
+    total = 0
+    with _connect(_load_pg_config()["dbname"]) as conn:
+        with conn.cursor() as cur:
+            sites = ALL_SITES if site in (None, "百世") else [site]
+            for s in sites:
+                if s not in ALL_SITES:
+                    print(f"   [跳过] 不支持应到/实到的站点: {s}")
+                    continue
+                total += _ingest_expected(cur, s, dates.get(s))
+                total += _ingest_actual(cur, s)
+            if site in (None, "百世"):
+                total += _ingest_undelivered_baishi(cur)
+        conn.commit()
+    print(f">> [ingest] 完成，共 {total} 条")
+    return total
+
+
+# ============================== 命令行 ==============================
+
+
+def _cli():
+    cmd = sys.argv[1] if len(sys.argv) > 1 else "all"
+    site = sys.argv[2] if len(sys.argv) > 2 else None
+    if cmd == "createdb":
+        create_database()
+    elif cmd == "init":
+        init_schema()
+    elif cmd == "ingest":
+        ingest(site)
+    elif cmd == "all":
+        create_database()
+        init_schema()
+        ingest()
+    else:
+        print(__doc__)
+        sys.exit(1)
+
+
+if __name__ == "__main__":
+    _cli()

requirements.txt

@@ -6,3 +6,4 @@ websocket-client>=1.0.0
 fastapi>=0.110.0
 uvicorn>=0.27.0
 apscheduler>=3.10.0
+psycopg[binary]>=3.1

schema.sql

@@ -0,0 +1,54 @@
+-- =====================================================================
+-- CQHXDB / 到货核销数据持久化 schema
+-- 设计：统一核心表 + JSONB raw 兜底（保留站点原始全列，一字段不丢）
+-- 幂等：业务唯一键 UPSERT，重复下载天然合并、零冗余
+-- 表：expected_record（运单级）/ actual_record（扫描件级）
+--     / undelivered_record（百世站点直供未到明细，子单级）
+-- =====================================================================
+
+-- 隔离到专用 schema（不污染 public）；表名无需再加前缀。
+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,   -- 运单基号（业务唯一键，去重键）
+    handover_no     TEXT,                     -- 交接单号
+    handover_pieces INTEGER,                  -- 交接件数（应到件数口径）
+    order_pieces    INTEGER,                  -- 录单件数
+    business_date   DATE,                     -- 业务日期（属性，非唯一键；读不到则 NULL）
+    raw             JSONB        NOT NULL,    -- 站点原始全列（key=原列名）
+    ingested_at     TIMESTAMPTZ  NOT NULL DEFAULT now(),
+    UNIQUE (site, waybill_no)
+);
+CREATE INDEX IF NOT EXISTS idx_expected_site_date ON expected_record (site, business_date);
+
+-- 实到货物（扫描件级：一扫描一行；每扫描一件系统生成一个单号）
+CREATE TABLE IF NOT EXISTS actual_record (
+    id           BIGSERIAL    PRIMARY KEY,
+    site         TEXT         NOT NULL,
+    waybill_no   TEXT         NOT NULL,   -- 运单基号（由扫描单号推导）
+    piece_no     TEXT         NOT NULL,   -- 扫描单号 / 子单号（业务唯一键）
+    scan_time    TIMESTAMPTZ,              -- 扫描时间（尽力解析，失败存 NULL，原始值在 raw）
+    scan_site    TEXT,                     -- 扫描网点
+    raw          JSONB        NOT NULL,
+    ingested_at  TIMESTAMPTZ  NOT NULL DEFAULT now(),
+    UNIQUE (site, piece_no)
+);
+CREATE INDEX IF NOT EXISTS idx_actual_waybill   ON actual_record (site, waybill_no);
+CREATE INDEX IF NOT EXISTS idx_actual_scan_time ON actual_record (scan_time);
+
+-- 应到未到（百世站点直供未到明细：子单级）
+CREATE TABLE IF NOT EXISTS undelivered_record (
+    id           BIGSERIAL    PRIMARY KEY,
+    site         TEXT         NOT NULL,   -- 百世
+    waybill_no   TEXT,                     -- 运单号
+    piece_no     TEXT,                     -- 子单号（业务唯一键）
+    biz_type     TEXT,                     -- 类型
+    last_scan    TEXT,                     -- 最新扫描记录
+    raw          JSONB        NOT NULL,
+    ingested_at  TIMESTAMPTZ  NOT NULL DEFAULT now(),
+    UNIQUE (site, piece_no)
+);