diff --git a/docs/2026-07-31-顺心DB差缺对比实施计划.md b/docs/2026-07-31-顺心DB差缺对比实施计划.md new file mode 100644 index 0000000..6054bae --- /dev/null +++ b/docs/2026-07-31-顺心DB差缺对比实施计划.md @@ -0,0 +1,264 @@ +# 顺心 DB 差缺对比 — 实施计划 + +> 日期：2026-07-31 +> 目标：将顺心站点差缺对比从 Excel 读取改为 PostgreSQL 查询，并修正 SF 运单特殊处理逻辑 + +--- + +## 一、背景 + +### 当前状态（Excel 方式） + +``` +compare.py:process("顺心") + ├── 读 downloads/顺心-应到货物数据.xlsx + ├── 读 downloads/顺心-实到货物数据.xlsx + ├── arrived_pieces_by_cols("运单号", "子单号") ← SF/non-SF 无区分 + └── 产出 {站}-未到数据.xlsx + 统计 dict +``` + +### 需要解决的两个问题 + +1. **从 Excel 切换到 DB**：数据已持久化到 PostgreSQL，比对应直接从 DB 查询 +2. **顺心 SF 运单特殊处理**：SF 运单的子单号为随机号码，不能用于去重计数，应使用行计数 + +--- + +## 二、数据结构 + +### PostgreSQL 表 + +**expected_record**（关键列）： + +| 列 | 类型 | 说明 | +|----|------|------| +| site | TEXT | 站点 | +| waybill_no | TEXT | 运单号（唯一键之一，SF 以 "SF" 开头） | +| handover_no | TEXT | 交接单号（批次标识） | +| handover_pieces | INTEGER | 交接件数（应到口径） | +| order_pieces | INTEGER | 录单件数（参考） | +| business_date | DATE | 下载目标日期 | + +**actual_record**（关键列）： + +| 列 | 类型 | 说明 | +|----|------|------| +| site | TEXT | 站点 | +| waybill_no | TEXT | 运单基号（关联 expected_record） | +| piece_no | TEXT | 扫描单号（non-SF：运单号+顺序号；SF：随机号码） | +| scan_time | TIMESTAMPTZ | 扫描时间（可靠，当天数据=当天扫描） | + +### SF 数据特征（已验证） + +- 顺心 actual_record 中 SF 运单：148 条 +- `piece_no == waybill_no`：86 条（58%） +- `piece_no != waybill_no`：62 条（42%）← 随机 SF 号码 +- SF 运单 expected：99 条，分布在 31 个交接批次中 + +--- + +## 三、算法设计 + +### 核心思路：以实到为锚，通过交接单号反推批次 + +``` +输入: site="顺心", date="2026-07-25" + +Step 1 — 取实到锚点 + SELECT DISTINCT waybill_no FROM actual_record + WHERE site='顺心' AND scan_time::date = '2026-07-25' + +Step 2 — 反推交接批次 + SELECT DISTINCT handover_no FROM expected_record + WHERE site='顺心' + AND waybill_no IN (Step 1 的运单集合) + +Step 3 — 展开批次全量应到 + SELECT waybill_no, handover_no, handover_pieces + FROM expected_record + WHERE site='顺心' + AND handover_no IN (Step 2 的交接单号集合) + +Step 4 — 取批次全量实到 + SELECT waybill_no, piece_no FROM actual_record + WHERE site='顺心' + AND waybill_no IN (Step 3 的运单集合) + +Step 5 — 逐运单比对 + for each waybill in Step 3: + if waybill_no LIKE 'SF%': + arrived_cnt = COUNT(*) ← 行计数，不去重 + else: + arrived_cnt = COUNT(DISTINCT piece_no) ← 子单号去重 + if arrived_cnt < handover_pieces → 差缺 +``` + +### SF vs non-SF 处理差异 + +| | non-SF | SF | +|------|--------|-----| +| piece_no 含义 | 运单号 + 顺序号（可推导） | 随机 SF 号码（无推导意义） | +| 实到计数方式 | `COUNT(DISTINCT piece_no)` | `COUNT(*)`（行计数） | +| 已到单号列表 | 列出去重后的子单号 | 列出所有 piece_no（含重复） | + +### 统计指标 + +| 指标 | 公式 | +|------|------| +| 运单数 | Step 3 去重运单数 | +| 应到件 | Σ handover_pieces | +| 已到件 | Σ arrived_cnt | +| 未到件 | max(0, 应到件 − 已到件) | +| 涉及运单 | arrived_cnt < handover_pieces 的运单数 | +| 完全未到 | arrived_cnt = 0 的运单数 | +| 部分未到 | 0 < arrived_cnt < handover_pieces 的运单数 | +| 未到率 | 未到件 ÷ 应到件 | + +### 边界情况覆盖 + +| 情况 | 覆盖方式 | +|------|----------| +| 同日多批次 | Step 2 查出全部涉及的 handover_no | +| 跨天到达（延迟） | Step 4 不限 scan_time，历史扫描全计入 | +| 溢到（实到 > 应到） | arrived_cnt >= n 跳过，不进差缺表 | +| 完全沉默批次 | 一件未扫 = 实到无锚点，该批次不会被触发——在首次有扫描那天被纳入 | +| SF 子单号重复 | 用 COUNT(*) 而非 COUNT(DISTINCT)，不会漏计 | + +--- + +## 四、模块设计 + +### 新增文件 + +**`inbound_verify/db_compare.py`** — DB 比对引擎（纯 PostgreSQL + Python） + +```python +# 核心函数签名 + +def compare_site_date(site: str, date: str) -> CompareResult | None: + """对指定站点和日期执行 DB 差缺比对。 + + 返回 CompareResult（stats + undelivered_rows）， + 当天无实到数据时返回 None。 + """ + +def compare_site_batch(site: str, handover_no: str) -> CompareResult | None: + """按指定交接单号执行全批次比对（不依赖实到锚点）。""" +``` + +**数据类型**： + +```python +@dataclass +class CompareResult: + stats: dict # 统计指标 + rows: list[dict] # 差缺明细行 + batches: list[str] # 涉及的交接批次 + +@dataclass +class UndeliveredRow: + handover_no: str # 交接单号 + waybill_no: str # 运单号 + total_pieces: int # 总件数（=交接件数） + arrived_pieces: int # 已到件数 + arrived_list: list[str] # 已到单号列表 + is_sf: bool # 是否 SF 运单 +``` + +### 修改文件 + +**`inbound_verify/cli/server.py`** — 新增 API 端点 + +```python +@app.post("/compare") +def run_compare(req: CompareRequest): + """DB 比对：{site, date} → 返回差缺结果""" + +@app.get("/compare/{site}/{date}") +def get_compare(site: str, date: str): + """查询某站点某日的差缺结果（缓存）""" +``` + +### 现有文件保持不动 + +- `compare.py` — 保留不动，Excel 比对继续可用 +- `domain.py` — 可能需要新增 DB 版站点配置（或复用现有） +- `runtime.py` — 暂不改动，`_site_undelivered_handler` 仍走 Excel 路径 + +--- + +## 五、实施步骤 + +### Phase 1 — `db_compare.py` 核心引擎 + +- [ ] 新建 `inbound_verify/db_compare.py` +- [ ] 实现 `compare_site_date("顺心", date)` +- [ ] SF/non-SF 分支处理 +- [ ] 返回 `CompareResult` +- [ ] 终端手动验证（直接调函数，打印结果） + +### Phase 2 — API 端点 + +- [ ] 在 `server.py` 新增 `POST /compare` +- [ ] `CompareRequest { site, date }` +- [ ] 调用 `db_compare.compare_site_date()` +- [ ] 返回 JSON：stats + undelivered rows +- [ ] HTTP 验证：curl 调 `/compare` 对比不同日期结果 + +### Phase 3 — Excel 输出（可选） + +- [ ] `db_compare` 生成 Excel 报告（复用现有 `compare.py` 的 openpyxl 样式） +- [ ] 输出到 `output/顺心-{date}-未到数据.xlsx` +- [ ] 或者只输出 JSON，前端自行渲染 + +### Phase 4 — 替换 undelivered 任务流 + +- [ ] `runtime.py` 新增 `_db_undelivered_handler` +- [ ] 下载完成后不再调 Excel 比对，改调 DB 比对 +- [ ] 逐步替换 `TASK_HANDLERS` 中的顺心 undelivered handler + +### Phase 5 — 扩展到中通/韵达/安能 + +- [ ] 各站适配（主要是 piece_no 去重方式差异） +- [ ] 中通：`COUNT(DISTINCT piece_no)`，无 SF 问题 +- [ ] 韵达：同上 +- [ ] 安能：同上 + +--- + +## 六、测试策略 + +### 手工验证（Phase 1） + +```python +# 终端直接调 +from inbound_verify.db_compare import compare_site_date +result = compare_site_date("顺心", "2026-07-25") +print(result.stats) +# 对比基于 Excel 版的 compare.process("顺心") 结果 +``` + +### API 验证（Phase 2） + +```bash +curl -X POST http://127.0.0.1:8000/compare \ + -H "Content-Type: application/json" \ + -d '{"site":"顺心","date":"2026-07-25"}' +``` + +### 回归验证 + +- 新 DB 比对结果 vs 旧 Excel 比对结果（同一份数据） +- SF 运单的 arrived_cnt 对比：DB 版（COUNT(*)）vs Excel 版（COUNT DISTINCT piece_no） +- 确认 SF 运单不再被漏计 + +--- + +## 七、风险与注意事项 + +| 风险 | 缓解 | +|------|------| +| DB 连接超时（cpolar 隧道） | 加 connect_timeout + try/except 降级 | +| 全表扫描性能 | 依赖 (site, waybill_no) 和 (site, scan_time) 索引 | +| SF 运单数据量小（~1%） | 测试覆盖可能不足——需找有 SF 差缺的日期验证 | +| `scan_time` 时区 | 统一用 `::date` cast，确认与服务器时区一致 | diff --git a/inbound_verify/cli/server.py b/inbound_verify/cli/server.py index 50e0a91..c927949 100644 --- a/inbound_verify/cli/server.py +++ b/inbound_verify/cli/server.py @@ -36,6 +36,7 @@ from inbound_verify.runtime import ( launch_and_prepare, run_heartbeat, ) +from inbound_verify import db_compare # 全部站点；百世固定下载当天，不可配置偏移 ALL_SITES = ["顺心", "百世", "中通", "韵达", "安能"] @@ -245,6 +246,71 @@ def list_tasks(limit: int = 20): return state_store.list_tasks(limit) +# ── DB 比对（基于 PostgreSQL，不依赖 Excel 文件）── + + +class CompareRequest(BaseModel): + site: str + date: str # YYYY-MM-DD + + +@app.post("/compare") +def run_compare(req: CompareRequest): + """DB 差缺比对：以实到扫描日期为锚点，反推交接批次，展开全量比对。 + 返回统计指标 + 差缺明细。 + """ + # 合法性校验 + if req.site not in db_compare.SITE_COMPARE_CONFIG: + raise HTTPException( + status_code=400, + detail=f"不支持的站点: {req.site}（支持: {list(db_compare.SITE_COMPARE_CONFIG.keys())}）", + ) + try: + target_date = datetime.strptime(req.date, "%Y-%m-%d").date() + except ValueError: + raise HTTPException( + status_code=400, detail=f"date 格式非法，需 YYYY-MM-DD: {req.date}" + ) + today = datetime.now().date() + if target_date > today: + raise HTTPException(status_code=400, detail=f"date 不可为未来日期: {req.date}") + + result = db_compare.compare_site_date(req.site, req.date) + if result is None: + raise HTTPException( + status_code=404, + detail=f"{req.site} {req.date}: 当天无实到数据，无法比对", + ) + + return { + "site": result.site, + "date": result.date, + "batches": result.batches, + "stats": { + "waybill_count": result.stats.waybill_count, + "sf_wb_count": result.stats.sf_wb_count, + "expected_pieces": result.stats.expected_pieces, + "arrived_pieces": result.stats.arrived_pieces, + "undelivered_pieces": result.stats.undelivered_pieces, + "undelivered_wb": result.stats.undelivered_wb, + "full_miss": result.stats.full_miss, + "part_miss": result.stats.part_miss, + "sf_undelivered": result.stats.sf_undelivered, + }, + "rows": [ + { + "handover_no": r.handover_no, + "waybill_no": r.waybill_no, + "total_pieces": r.total_pieces, + "arrived_pieces": r.arrived_pieces, + "arrived_list": r.arrived_list, + "is_sf": r.is_sf, + } + for r in result.rows + ], + } + + @app.get("/status") def get_status(): """各站登录态 + 数据态 + 入库态（前端状态盘用），另含 worker 就绪状态。""" diff --git a/inbound_verify/db_compare.py b/inbound_verify/db_compare.py new file mode 100644 index 0000000..0e57f36 --- /dev/null +++ b/inbound_verify/db_compare.py @@ -0,0 +1,534 @@ +# -*- coding: utf-8 -*- +""" +db_compare.py — 基于 PostgreSQL 的应到未到差缺比对引擎。 + +与 compare.py（Excel 版）并行：本模块直接从 DB 查询数据进行比对， +不依赖 downloads/ 下的 Excel 文件。 + +核心思路：以实到扫描日期为锚点 → 反推交接批次 → 展开批次全量比对。 + +每个站点只需提供配置（waybill 列名 / piece 列名 / 是否有 SF 特殊处理）， +核心比对逻辑完全通用。 + +顺心站点 SF 运单特殊处理：SF 运单的子单号（piece_no）为随机号码，不能用 +COUNT(DISTINCT piece_no) 去重计数，改为 COUNT(*) 行计数。 + +用法： + from inbound_verify.db_compare import compare_site_date, SITE_COMPARE_CONFIG + + result = compare_site_date("顺心", "2026-07-25") + if result: + print(result.stats) + for row in result.rows: + print(row) +""" + +import os +from dataclasses import dataclass, field +from datetime import date, datetime + +import psycopg +import yaml +from openpyxl import Workbook +from openpyxl.styles import Font, PatternFill, Alignment, Border, Side + +from inbound_verify.paths import CONFIG_PATH, OUTPUT_DIR, DOWNLOAD_DIR +from inbound_verify.domain import _site_cfg, ALL_REPORT_SITES + +# ============================== 结果类型 ============================== + + +@dataclass +class CompareStats: + """单站点/单批次比对统计。""" + + waybill_count: int = 0 # 应到运单数 + expected_pieces: int = 0 # 应到件数 + arrived_pieces: int = 0 # 实到件数 + undelivered_pieces: int = 0 # 未到件数 + undelivered_wb: int = 0 # 差缺运单数 + full_miss: int = 0 # 完全未到 + part_miss: int = 0 # 部分未到 + sf_wb_count: int = 0 # SF 运单数 + sf_undelivered: int = 0 # SF 差缺数 + + +@dataclass +class UndeliveredRow: + """单条差缺明细。""" + + handover_no: str = "" # 交接单号 + waybill_no: str = "" # 运单号 + total_pieces: int = 0 # 总件数（交接件数） + arrived_pieces: int = 0 # 已到件数 + arrived_list: list = field(default_factory=list) # 已到单号列表 + is_sf: bool = False # 是否 SF 运单 + + +@dataclass +class CompareResult: + """一次比对的完整结果。""" + + site: str = "" + date: str = "" + batches: list = field(default_factory=list) # 涉及的交接批次 + stats: CompareStats = field(default_factory=CompareStats) + rows: list = field(default_factory=list) # UndeliveredRow 列表 + + +# ============================== 站点比对配置 ============================== + + +@dataclass +class SiteCompareConfig: + """DB 比对的站点参数。""" + + name: str # 站点名 + has_sf: bool = False # 是否需要区分 SF 运单 + + +# 四站点 DB 比对配置（百世不参与 4 站比对） +SITE_COMPARE_CONFIG: dict[str, SiteCompareConfig] = { + "顺心": SiteCompareConfig(name="顺心", has_sf=True), + "中通": SiteCompareConfig(name="中通", has_sf=False), + "韵达": SiteCompareConfig(name="韵达", has_sf=False), + "安能": SiteCompareConfig(name="安能", has_sf=False), +} + + +# ============================== DB 连接 ============================== + + +def _load_pg_config(): + """从 config.yaml 读 postgres 段。与 store.py 共用同一配置源。""" + 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"), + "connect_timeout_seconds": int(pg.get("connect_timeout_seconds", 5)), + } + + +def _connect(): + c = _load_pg_config() + return psycopg.connect( + host=c["host"], + port=c["port"], + dbname=c["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 compare_site_date(site: str, target_date: str) -> CompareResult | None: + """对指定站点和日期执行 DB 差缺比对。 + + 算法： + 1. 取 scan_time::date = target_date 的实到运单（锚点） + 2. 反推这些运单所属的交接批次（handover_no） + 3. 展开批次全量应到运单 + 4. 查询批次全量实到扫描 + 5. 逐运单比对差缺（SF/non-SF 分支处理） + + Args: + site: 站点名（"顺心"/"中通"/"韵达"/"安能"） + target_date: 日期 "YYYY-MM-DD" + + Returns: + CompareResult 或 None（当天无实到数据时返回 None） + """ + cfg = SITE_COMPARE_CONFIG.get(site) + if cfg is None: + print(f"[db_compare] 不支持的站点: {site}") + return None + + try: + conn = _connect() + cur = conn.cursor() + + # ── Step 1: 取实到锚点 ── + cur.execute( + """ + SELECT DISTINCT waybill_no FROM actual_record + WHERE site = %s AND scan_time::date = %s + """, + (site, target_date), + ) + anchor_wbs = [r[0] for r in cur.fetchall()] + if not anchor_wbs: + print(f"[db_compare] {site} {target_date}: 当天无实到数据") + conn.close() + return None + + # ── Step 2: 反推交接批次 ── + cur.execute( + """ + SELECT DISTINCT e.handover_no FROM expected_record e + WHERE e.site = %s AND e.waybill_no = ANY(%s) + """, + (site, anchor_wbs), + ) + batches = [r[0] for r in cur.fetchall()] + + # ── Step 3: 展开批次全量应到 ── + cur.execute( + """ + SELECT waybill_no, handover_no, handover_pieces + FROM expected_record + WHERE site = %s AND handover_no = ANY(%s) + ORDER BY handover_no, waybill_no + """, + (site, batches), + ) + exp_rows = cur.fetchall() # [(waybill_no, handover_no, handover_pieces), ...] + + if not exp_rows: + conn.close() + return None + + all_wbs = [r[0] for r in exp_rows] + + # ── Step 4: 取批次全量实到 ── + cur.execute( + """ + SELECT waybill_no, piece_no FROM actual_record + WHERE site = %s AND waybill_no = ANY(%s) + ORDER BY waybill_no, piece_no + """, + (site, all_wbs), + ) + act_rows = cur.fetchall() # [(waybill_no, piece_no), ...] + + conn.close() + + # ── Step 5: 逐运单比对 ── + return _do_compare(site, target_date, batches, exp_rows, act_rows, cfg) + + except Exception as e: + print(f"[db_compare] {site} {target_date} 比对异常: {e}") + return None + + +def compare_site_batch(site: str, handover_no: str) -> CompareResult | None: + """按指定交接单号执行全批次比对（不依赖实到锚点）。 + + 用于已知交接单号后精确比对某一批次。 + """ + cfg = SITE_COMPARE_CONFIG.get(site) + if cfg is None: + print(f"[db_compare] 不支持的站点: {site}") + return None + + try: + conn = _connect() + cur = conn.cursor() + + cur.execute( + """ + SELECT waybill_no, handover_no, handover_pieces + FROM expected_record + WHERE site = %s AND handover_no = %s + ORDER BY waybill_no + """, + (site, handover_no), + ) + exp_rows = cur.fetchall() + if not exp_rows: + conn.close() + return None + + all_wbs = [r[0] for r in exp_rows] + + cur.execute( + """ + SELECT waybill_no, piece_no FROM actual_record + WHERE site = %s AND waybill_no = ANY(%s) + ORDER BY waybill_no, piece_no + """, + (site, all_wbs), + ) + act_rows = cur.fetchall() + + conn.close() + + return _do_compare( + site, + f"batch:{handover_no}", + [handover_no], + exp_rows, + act_rows, + cfg, + ) + + except Exception as e: + print(f"[db_compare] {site} batch:{handover_no} 比对异常: {e}") + return None + + +# ============================== 比对核心 ============================== + + +def _do_compare( + site: str, + label: str, + batches: list[str], + exp_rows: list[tuple], # [(waybill_no, handover_no, handover_pieces), ...] + act_rows: list[tuple], # [(waybill_no, piece_no), ...] + cfg: SiteCompareConfig, +) -> CompareResult: + """执行逐运单比对，产出统计 + 差缺明细。 + + 与 compare.py:process() 口径一致： + - 应到件数 = handover_pieces（交接件数） + - 实到件数 = SF ? COUNT(*) : COUNT(DISTINCT piece_no) + - arrived_cnt >= handover_pieces → 足额到货，跳过 + """ + # 构建实到索引: waybill_no → [piece_no, ...]（保留所有行，不去重） + act_by_wb: dict[str, list[str]] = {} + for wb, piece in act_rows: + act_by_wb.setdefault(wb, []).append(piece) + + stats = CompareStats() + rows: list[UndeliveredRow] = [] + max_arrived = 0 + + for wb, handover_no, handover_pcs in exp_rows: + handover_pcs = handover_pcs or 0 + if handover_pcs <= 0: + continue + + stats.waybill_count += 1 + stats.expected_pieces += handover_pcs + + is_sf = cfg.has_sf and wb.startswith("SF") + if is_sf: + stats.sf_wb_count += 1 + + all_pieces = act_by_wb.get(wb, []) + + if is_sf: + # SF: 行计数，不去重（piece_no 是随机号码） + arrived_cnt = len(all_pieces) + arrived_list = list(all_pieces) + else: + # non-SF: 子单号去重 + unique_pieces = list(dict.fromkeys(all_pieces)) # 保序去重 + arrived_cnt = len(unique_pieces) + arrived_list = unique_pieces + + stats.arrived_pieces += arrived_cnt + + if arrived_cnt >= handover_pcs: + continue # 足额或溢到，不进差缺表 + + if arrived_cnt == 0: + stats.full_miss += 1 + else: + stats.part_miss += 1 + + if is_sf: + stats.sf_undelivered += 1 + + max_arrived = max(max_arrived, arrived_cnt) + rows.append( + UndeliveredRow( + handover_no=handover_no, + waybill_no=wb, + total_pieces=handover_pcs, + arrived_pieces=arrived_cnt, + arrived_list=arrived_list, + is_sf=is_sf, + ) + ) + + stats.undelivered_pieces = max(0, stats.expected_pieces - stats.arrived_pieces) + stats.undelivered_wb = stats.full_miss + stats.part_miss + + result = CompareResult( + site=site, + date=label, + batches=batches, + stats=stats, + rows=rows, + ) + + # 打印摘要 + print( + f"[db_compare] {site} {label}: " + f"batches={len(batches)}, " + f"wb={stats.waybill_count}(SF:{stats.sf_wb_count}), " + f"exp={stats.expected_pieces}, arr={stats.arrived_pieces}, " + f"miss={stats.undelivered_pieces}, " + f"miss_wb={stats.undelivered_wb}(full={stats.full_miss}, part={stats.part_miss})" + ) + if stats.sf_undelivered: + print(f" SF 差缺: {stats.sf_undelivered} 个运单") + + return result + + +# ============================== Excel 输出 ============================== + + +# 样式常量（与 compare.py 对齐） +_FONT = "微软雅黑" +_BLUE = "305496" + +_HEADER_FILL = PatternFill("solid", fgColor=_BLUE) +_HEADER_FONT = Font(name=_FONT, bold=True, color="FFFFFF", size=11) +_BODY_FONT = Font(name=_FONT, size=10) +_THIN = Side(style="thin", color="D9D9D9") +_BORDER = Border(left=_THIN, right=_THIN, top=_THIN, bottom=_THIN) + + +def write_result_excel(result: CompareResult, output_path: str | None = None) -> str: + """将比对结果写入 Excel 文件。 + + Args: + result: compare_site_date 或 compare_site_batch 的返回值 + output_path: 输出路径，为 None 时自动生成： + output/{站}-{日期}-未到数据.xlsx + + Returns: + 实际写入的文件路径 + """ + if output_path is None: + os.makedirs(OUTPUT_DIR, exist_ok=True) + date_tag = result.date.replace(":", "-").replace("batch:", "batch-") + output_path = os.path.join( + OUTPUT_DIR, f"{result.site}-{date_tag}-未到数据.xlsx" + ) + + wb = Workbook() + ws = wb.active + ws.title = result.site + + _write_sheet(ws, result) + wb.save(output_path) + print(f"[db_compare] Excel 已输出: {output_path}") + return output_path + + +def _write_sheet(ws, result: CompareResult): + """写单个站点的差缺明细 sheet。""" + s = result.stats + rows = result.rows + + # 动态列: 交接单号 | 运单号 | 总件数 | 已到单号1 | 已到单号2 | ... + max_arrived = max((len(r.arrived_list) for r in rows), default=0) + columns = ["交接单号", "运单号", "总件数"] + [ + f"已到单号{i + 1}" for i in range(max_arrived) + ] + + ws.sheet_view.showGridLines = False + + # 表头 + ws.append(columns) + for c in range(1, len(columns) + 1): + cell = ws.cell(row=1, column=c) + cell.fill = _HEADER_FILL + cell.font = _HEADER_FONT + cell.alignment = Alignment(horizontal="center", vertical="center") + cell.border = _BORDER + + # 数据行 + for row in rows: + values = { + "交接单号": row.handover_no, + "运单号": row.waybill_no, + "总件数": row.total_pieces, + } + for i, piece in enumerate(row.arrived_list): + values[f"已到单号{i + 1}"] = piece + ws.append([values.get(c, "") for c in columns]) + + # 格式 + for r in range(2, ws.max_row + 1): + for c, col in enumerate(columns, start=1): + cell = ws.cell(row=r, column=c) + cell.font = _BODY_FONT + cell.border = _BORDER + if col == "总件数": + cell.number_format = "#,##0" + cell.alignment = Alignment(horizontal="right", vertical="center") + else: + cell.number_format = "@" + + # 列宽 + for c, col in enumerate(columns, start=1): + body_lens = [ + len(str(ws.cell(row=r, column=c).value or "")) + for r in range(2, ws.max_row + 1) + ] + width = min(max([len(str(col))] + body_lens) + 4, 36) + ws.column_dimensions[ws.cell(row=1, column=c).column_letter].width = max( + width, 12 + ) + + ws.freeze_panes = "A2" + + +# ============================== 终端验证入口 ============================== + + +def main(): + """命令行验证入口： + python -m inbound_verify.db_compare 顺心 2026-07-25 + """ + import sys + + site = sys.argv[1] if len(sys.argv) > 1 else "顺心" + target_date = sys.argv[2] if len(sys.argv) > 2 else "2026-07-25" + + result = compare_site_date(site, target_date) + if result is None: + print(f"{site} {target_date}: 无结果") + return + + print(f"\n=== {result.site} {result.date} 差缺明细 ===") + print(f"涉及批次: {result.batches}") + print(f"应到运单: {result.stats.waybill_count} (SF: {result.stats.sf_wb_count})") + print(f"应到件数: {result.stats.expected_pieces}") + print(f"实到件数: {result.stats.arrived_pieces}") + print(f"未到件数: {result.stats.undelivered_pieces}") + print( + f"差缺运单: {result.stats.undelivered_wb} (完全未到: {result.stats.full_miss}, 部分未到: {result.stats.part_miss})" + ) + if result.stats.sf_undelivered: + print(f"SF 差缺: {result.stats.sf_undelivered}") + + if result.rows: + print(f"\n--- 差缺明细 (共 {len(result.rows)} 条) ---") + for row in result.rows[:20]: + sf = "[SF]" if row.is_sf else "" + arrived_preview = row.arrived_list[:5] + print( + f" {sf} {row.waybill_no}: " + f"应到{row.total_pieces}件, 实到{row.arrived_pieces}件" + f" {f'已到: {arrived_preview}' if arrived_preview else ''}" + ) + if len(result.rows) > 20: + print(f" ... 还有 {len(result.rows) - 20} 条") + + # 输出 Excel + path = write_result_excel(result) + print(f"\n结果文件: {path}") + + +if __name__ == "__main__": + main() diff --git a/inbound_verify/runtime.py b/inbound_verify/runtime.py index 6081d09..5943b23 100644 --- a/inbound_verify/runtime.py +++ b/inbound_verify/runtime.py @@ -476,23 +476,44 @@ def _web_handler(site, download_func): def _site_undelivered_handler(site): - """4 站未到：下应到+实到 → 比对写 downloads/<站>-未到数据.xlsx。 - 任一下载失败 → 清掉旧未到文件、返回 False（前端不展示陈旧未到）。""" + """4 站未到：下应到+实到 → DB 比对 → 写 output/<站>-<日期>-未到数据.xlsx。 + 应到全量去重（已落库则跳过导出），因此比对不依赖 Excel 文件，走数据库查询。 + 下载成功则返回 True（比对失败不影响任务判定，数据已入库）。""" def handler(ctx, force=False, date=None): - # 各站下载入口约定返回 True/False；顺心历史返回 None（视为成功，与 dispatch 一致） exp_ok = TASK_HANDLERS[(site, "expected")](ctx, force, date) is not False act_ok = ( (TASK_HANDLERS[(site, "actual")](ctx, force, date) is not False) if exp_ok else False ) - if exp_ok and act_ok: - return compare.write_site_file(site) - stale = os.path.join(DOWNLOAD_DIR, SITE_UNDELIVERED_FILE.format(name=site)) - if os.path.exists(stale): - os.remove(stale) - return False + if not exp_ok or not act_ok: + stale = os.path.join(DOWNLOAD_DIR, SITE_UNDELIVERED_FILE.format(name=site)) + if os.path.exists(stale): + os.remove(stale) + return False + + # ── DB 比对（替代旧 Excel 比对）── + try: + from inbound_verify import db_compare # 懒导入，避免成环 + + if date: + target_date = date + else: + offset = state_store.get_offset(site, "expected") + target_date = (datetime.now().date() - timedelta(days=offset)).strftime( + "%Y-%m-%d" + ) + + result = db_compare.compare_site_date(site, target_date) + if result is not None: + db_compare.write_result_excel(result) + else: + print(f">> [未到] {site} {target_date}: 当天无实到数据，跳过比对") + except Exception as e: + print(f">> [未到] {site} DB 比对异常（不影响下载结果）: {e}") + + return True # 下载成功即返回 True，比对失败不影响任务判定 return handler