diff --git a/config.example.yaml b/config.example.yaml index eae651c..3d691f8 100644 --- a/config.example.yaml +++ b/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 diff --git a/db_store.py b/db_store.py new file mode 100644 index 0000000..3510ef0 --- /dev/null +++ b/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() diff --git a/requirements.txt b/requirements.txt index 1652680..dc6c473 100644 --- a/requirements.txt +++ b/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 diff --git a/schema.sql b/schema.sql new file mode 100644 index 0000000..862dee4 --- /dev/null +++ b/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) +);