36b204abe5
将未到统计从实到驱动（实到锚点反推批次）改为应到驱动（batch_out_date 归属日直接取应到批次），以吸收应到任务提前 1~2 天提交的扰动（韵达固定 +1、中通偶发 +1）。 - expected_record 新增 out_date / batch_out_date + 索引；入库解析出库时间并聚合批次归属日，支持历史回填 - 新增 compare_site_outdate 应到驱动入口，保留 compare_site_date 实到驱动作对照 - 未到任务 / POST /compare / 全站汇总切换到应到驱动 - 附现状梳理、设计、实现总结三篇文档
670 lines
25 KiB
Python
670 lines
25 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
store.py — 到货核销数据持久化（PostgreSQL）
|
||
|
||
职责：把 downloads/ 下各站点下载的应到 / 实到 / 未到 Excel 解析后，幂等写入 PostgreSQL。
|
||
与下载流程解耦：本模块只读 downloads/ 现有文件入库，不关心谁触发下载、下载了几次。
|
||
|
||
设计要点：
|
||
- 三张表：expected_record（运单级）/ actual_record（扫描件级）
|
||
/ undelivered_record（百世站点直供未到明细，子单级）
|
||
- 每行 = 统一核心列 + raw JSONB（站点原始全列，key=原列名，一字段不丢）
|
||
- 幂等：业务唯一键 UPSERT，重复下载天然合并、零冗余
|
||
- 单号一律按文本读写（dtype=str），防长数字被科学计数 / 精度丢失
|
||
|
||
命令行：
|
||
python -m inbound_verify.store createdb 创建数据库（幂等）
|
||
python -m inbound_verify.store init 建表（幂等 CREATE TABLE IF NOT EXISTS）
|
||
python -m inbound_verify.store ingest [site] 入库全站或单站（幂等 UPSERT）
|
||
python -m inbound_verify.store ingest-one <site> <kind> 仅入库指定站/类（钩子同款路由）
|
||
python -m inbound_verify.store 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 inbound_verify.paths import BASE_DIR, CONFIG_PATH, DOWNLOAD_DIR
|
||
from inbound_verify.domain import (
|
||
BAISHI_FILE,
|
||
_site_cfg,
|
||
) # 站点 / 文件名配置（单一来源）
|
||
|
||
from inbound_verify import compare # _read_business_dates（比对侧业务日期读取）
|
||
|
||
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"),
|
||
"auto_ingest": bool(pg.get("auto_ingest", True)),
|
||
"connect_timeout_seconds": int(pg.get("connect_timeout_seconds", 5)),
|
||
}
|
||
|
||
|
||
def _connect(dbname):
|
||
"""用关键字参数连接（避开 conninfo 对密码特殊字符的解析）。
|
||
options 设 search_path 到专用 schema + 会话级 statement_timeout=30s
|
||
（cpolar 隧道上防失控查询；connect_timeout 守连接阶段）。"""
|
||
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']} -c statement_timeout=30s",
|
||
connect_timeout=c["connect_timeout_seconds"],
|
||
)
|
||
|
||
|
||
def ingest_enabled():
|
||
"""是否启用下载后自动入库（config.yaml postgres.auto_ingest，默认 True）。
|
||
供 runtime 钩子判定开关，避免它伸手进 _load_pg_config。"""
|
||
return _load_pg_config()["auto_ingest"]
|
||
|
||
|
||
# ============================== 建库 / 建表 ==============================
|
||
|
||
|
||
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] 表结构已就绪")
|
||
|
||
|
||
# ============================== 解析辅助 ==============================
|
||
|
||
# 实到单号列 / 基号列映射（口径取自 domain，与 STATIONS 对齐）：
|
||
# 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 _batch_out_date_map(df, cfg, out_col="出库时间"):
|
||
"""按交接单号分组，计算批次归属日：出库日期众数，并列取较早日期。
|
||
返回 {handover_no: date}；无出库时间/无交接单号的行不参与。"""
|
||
from collections import Counter
|
||
|
||
jd_col = cfg.get("exp_jd", "交接单号")
|
||
if jd_col not in df.columns or out_col not in df.columns:
|
||
return {}
|
||
counter: dict[str, Counter] = {}
|
||
for _, r in df.iterrows():
|
||
hn = str(r.get(jd_col, "")).strip()
|
||
if not hn or hn == "nan":
|
||
continue
|
||
d = _parse_time(r.get(out_col))
|
||
if d is None:
|
||
continue
|
||
counter.setdefault(hn, Counter())[d.date()] += 1
|
||
out = {}
|
||
for hn, cnt in counter.items():
|
||
if not cnt:
|
||
continue
|
||
max_n = max(cnt.values())
|
||
earliest = min(d for d, n in cnt.items() if n == max_n)
|
||
out[hn] = earliest
|
||
return out
|
||
|
||
|
||
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 compare._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, 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,
|
||
handover_pieces = EXCLUDED.handover_pieces,
|
||
order_pieces = EXCLUDED.order_pieces,
|
||
business_date = COALESCE(EXCLUDED.business_date, expected_record.business_date),
|
||
out_date = COALESCE(EXCLUDED.out_date, expected_record.out_date),
|
||
batch_out_date = COALESCE(EXCLUDED.batch_out_date, expected_record.batch_out_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()
|
||
"""
|
||
|
||
_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()
|
||
"""
|
||
|
||
|
||
# ============================== 入库 ==============================
|
||
|
||
|
||
def _ingest_expected(cur, site, business_date):
|
||
"""入库单站应到（运单级，按 waybill_no 去重 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")
|
||
biz = _parse_date(business_date)
|
||
out_col = "出库时间"
|
||
has_out_col = out_col in df.columns
|
||
# 批次归属日：同交接单号出库日众数，并列取较早
|
||
batch_out_date = _batch_out_date_map(df, cfg, out_col)
|
||
rows = []
|
||
for r in df.to_dict("records"):
|
||
wb = str(r.get(cfg["exp_wb"], "")).strip()
|
||
if not wb:
|
||
continue
|
||
out_date = None
|
||
if has_out_col:
|
||
out_dt = _parse_time(r.get(out_col))
|
||
if out_dt is not None:
|
||
out_date = out_dt.date()
|
||
hn = str(r.get(cfg["exp_jd"], "")).strip() or None
|
||
rows.append(
|
||
(
|
||
site,
|
||
wb,
|
||
hn,
|
||
_to_int(r.get(cfg["exp_qty"])),
|
||
_to_int(r.get("录单件数")),
|
||
biz,
|
||
out_date,
|
||
batch_out_date.get(hn),
|
||
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 = _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, BAISHI_FILE)
|
||
if not os.path.exists(path):
|
||
print(f" [跳过] 百世 未到：文件不存在 {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 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()
|
||
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 ingest_task(site, kind):
|
||
"""按 (site, kind) 入库本次刚下载的文件（幂等 UPSERT），返回总条数。
|
||
与 ingest(site) 的区别：只入本次刷新的那一类，避免重读写另一类文件（同步钩子里减少阻塞）。
|
||
kind 路由：
|
||
expected/actual 各入其列；
|
||
undelivered 百世 入未到；
|
||
undelivered 4 站 _site_undelivered_handler 内部连带下了 expected+actual，故入两者；
|
||
__compare__ / 其它组合 返回 0。
|
||
"""
|
||
if site == "__compare__":
|
||
return 0
|
||
# 百世无应到/实到（只有站点直供的未到）；非 undelivered 直接返回 0，避免
|
||
# _ingest_expected/_ingest_actual 走到 _site_cfg(百世)=None 而 TypeError。
|
||
if site == "百世" and kind != "undelivered":
|
||
return 0
|
||
dates = _read_business_dates()
|
||
total = 0
|
||
with _connect(_load_pg_config()["dbname"]) as conn:
|
||
with conn.cursor() as cur:
|
||
if kind == "expected":
|
||
total += _ingest_expected(cur, site, dates.get(site))
|
||
elif kind == "actual":
|
||
total += _ingest_actual(cur, site)
|
||
elif kind == "undelivered":
|
||
if site == "百世":
|
||
total += _ingest_undelivered_baishi(cur)
|
||
else: # 顺心/中通/韵达/安能
|
||
total += _ingest_expected(cur, site, dates.get(site))
|
||
total += _ingest_actual(cur, site)
|
||
# 其它组合（如 百世/expected，正常不经钩子触发）：防御性返回 0
|
||
conn.commit()
|
||
return total
|
||
|
||
|
||
def backfill_out_date(site=None):
|
||
"""历史数据回填：从 raw->>'出库时间' 解析出库日，写入 out_date；
|
||
再按交接单号聚合出库日众数，回填 batch_out_date。
|
||
site 为空时处理全部站点。返回回填 out_date 条数。"""
|
||
sites = [site] if site else ALL_SITES
|
||
total = 0
|
||
with _connect(_load_pg_config()["dbname"]) as conn:
|
||
with conn.cursor() as cur:
|
||
for s in sites:
|
||
# ── Step 1: 回填 out_date（仅 NULL 行）──
|
||
cur.execute(
|
||
"SELECT id, raw FROM expected_record "
|
||
"WHERE site=%s AND out_date IS NULL",
|
||
(s,),
|
||
)
|
||
rows = cur.fetchall()
|
||
updates = []
|
||
for rid, raw in rows:
|
||
if not isinstance(raw, dict):
|
||
continue
|
||
out_dt = _parse_time(raw.get("出库时间"))
|
||
if out_dt is None:
|
||
continue
|
||
updates.append((out_dt.date(), rid))
|
||
if updates:
|
||
cur.executemany(
|
||
"UPDATE expected_record SET out_date=%s WHERE id=%s",
|
||
updates,
|
||
)
|
||
total += len(updates)
|
||
print(f" [回填] {s}：{len(updates)}/{len(rows)} 条")
|
||
|
||
# ── Step 2: 回填 batch_out_date（仅 NULL 行）──
|
||
cur.execute(
|
||
"SELECT id, handover_no, out_date FROM expected_record "
|
||
"WHERE site=%s AND batch_out_date IS NULL",
|
||
(s,),
|
||
)
|
||
rows = cur.fetchall()
|
||
if rows:
|
||
from collections import Counter
|
||
|
||
cnt: dict[str, Counter] = {}
|
||
for _, hn, od in rows:
|
||
if not hn or od is None:
|
||
continue
|
||
cnt.setdefault(hn, Counter())[od] += 1
|
||
batch_map = {}
|
||
for hn, c in cnt.items():
|
||
max_n = max(c.values())
|
||
batch_map[hn] = min(d for d, n in c.items() if n == max_n)
|
||
if batch_map:
|
||
cur.executemany(
|
||
"UPDATE expected_record SET batch_out_date=%s "
|
||
"WHERE site=%s AND handover_no=%s",
|
||
[(d, s, hn) for hn, d in batch_map.items()],
|
||
)
|
||
print(f" [回填] {s} batch_out_date：{len(batch_map)} 个批次")
|
||
conn.commit()
|
||
print(f">> [回填] out_date 完成，共 {total} 条")
|
||
return total
|
||
|
||
|
||
def get_existing_handover_nos(site):
|
||
"""查该站点已落库的交接单号集合（expected_record.handover_no）。
|
||
供"提交导出任务前"去重：已落库的交接单号不再重复提交导出任务。
|
||
PG 不可用（cpolar 抖动等）时返回空集 + 告警，调用方按"未确认存在"处理
|
||
（继续提交导出，UPSERT 兜底，绝不因去重查询失败而漏数据）。"""
|
||
try:
|
||
with _connect(_load_pg_config()["dbname"]) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(
|
||
"SELECT handover_no FROM expected_record "
|
||
"WHERE site=%s AND handover_no IS NOT NULL AND handover_no <> ''",
|
||
(site,),
|
||
)
|
||
return {str(r[0]).strip() for r in cur.fetchall()}
|
||
except Exception as e:
|
||
print(f">> [去重] 查询已落库交接单号失败（{site}），本次不去重: {e}")
|
||
return set()
|
||
|
||
|
||
# ============================== PG 数据存在性查询 ==============================
|
||
|
||
|
||
def has_data(site, kind, target_date):
|
||
"""查询 PG：指定站点在 target_date 是否有业务数据。
|
||
target_date: str 'YYYY-MM-DD' 或 date 对象。
|
||
返回 (has_rows: bool, count: int)。
|
||
PG 不可达时返回 (False, 0)，不抛异常——调用方按「未确认存在」处理。
|
||
|
||
kind 路由：
|
||
expected → expected_record (business_date)
|
||
actual → actual_record (scan_time::date)
|
||
undelivered → 百世: baishi_daily_stats；4 站: 不单独查（由调用方 expected∧actual 派生）
|
||
"""
|
||
if site == "百世" and kind == "undelivered":
|
||
sql = (
|
||
"SELECT COUNT(*) FROM baishi_daily_stats"
|
||
" WHERE site = %s AND business_date = %s"
|
||
)
|
||
params = (site, target_date)
|
||
elif kind == "expected":
|
||
sql = (
|
||
"SELECT COUNT(*) FROM expected_record"
|
||
" WHERE site = %s AND business_date = %s"
|
||
)
|
||
params = (site, target_date)
|
||
elif kind == "actual":
|
||
sql = (
|
||
"SELECT COUNT(*) FROM actual_record"
|
||
" WHERE site = %s AND scan_time::date = %s"
|
||
)
|
||
params = (site, target_date)
|
||
else:
|
||
return (False, 0)
|
||
try:
|
||
with _connect(_load_pg_config()["dbname"]) as conn:
|
||
with conn.cursor() as cur:
|
||
cur.execute(sql, params)
|
||
row = cur.fetchone()
|
||
cnt = int(row[0]) if row else 0
|
||
return (cnt > 0, cnt)
|
||
except Exception as e:
|
||
print(f">> [状态] PG 查询 {site}/{kind}/{target_date} 失败: {e}")
|
||
return (False, 0)
|
||
|
||
|
||
# ============================== 命令行 ==============================
|
||
|
||
|
||
def main():
|
||
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()
|
||
elif cmd == "ingest-one":
|
||
kind = sys.argv[3] if len(sys.argv) > 3 else None
|
||
if not site or kind not in ("expected", "actual", "undelivered"):
|
||
print(
|
||
"用法: python -m inbound_verify.store ingest-one <site> <expected|actual|undelivered>"
|
||
)
|
||
sys.exit(1)
|
||
total = ingest_task(site, kind)
|
||
print(f">> [ingest-one] {site}/{kind} 入库 {total} 条")
|
||
elif cmd == "backfill-out-date":
|
||
backfill_out_date(site)
|
||
else:
|
||
print(__doc__)
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|