"""yun8 运输流程编排器。 将 Postman Collection 中运输流程 18 个接口 + SQL 写入串联为完整流水线， 变量传递模拟 Postman 的 Collection Variables 机制（步骤 N 的输出写入 self.vars，供后续步骤读取）。 与旧 yun8_sync/pipeline.py 的差异（均为迁移计划内的有意行为变化）： - _check_response 业务 code != 0 时抛 Yun8StepError（旧版仅 print） - step17 未提取到 payNo 时抛 Yun8StepError（旧版警告后继续，写库 front_seq_no 为空） - step20 上报运单 / step21 滇运通审核失败不再吞异常，向上传播 - SQL 全部参数化，reset 序列包进事务，连接复用 - print → 统一 logger（output/logs/ 文件 + 控制台） Usage: from tests.apis.yun8.pipeline import Yun8Pipeline pipe = Yun8Pipeline(config["yun8_api"], db) pipe.set_auth(driver_token, admin_token) pipe.run_driver_flow() # 步骤 1-5 """ import random import time import uuid from datetime import datetime, timedelta from tests.core.db_helper import Db from tests.core.logger import get_logger from tests.apis.yun8.client import Yun8ApiClient from tests.apis.yun8.sql_templates import ( RECEIPT_FILE_INSERT_SQL, ETC_INVOICE_INSERT_SQL, ETC_PUSH_INSERT_SQL, etc_invoice_insert_params, etc_status_update, post_unload_reset_update, receipt_file_insert_params, driver_reset_statements, risk_control_update, ) class Yun8StepError(RuntimeError): """yun8 流程步骤失败（业务 code != 0 或关键变量提取失败）。""" def __init__(self, step_name: str, code=None, msg: str = ""): self.step_name = step_name self.code = code self.msg = msg detail = f"{step_name} 失败" if code is not None: detail += f" (code={code}, msg={msg})" elif msg: detail += f" ({msg})" super().__init__(detail) def _gen_code() -> str: """按现有数据规则生成 code: 32 位小写十六进制 (UUID4 hex 去横线)""" return uuid.uuid4().hex # ETC 发票路段链（广东→河北，按行驶方向依次经过；每条发票取相邻两站作为进/出站） _ETC_STATION_CHAIN = [ "广东古镇东站", "湖南郴州西站", "湖南岳阳东站", "湖北孝感东站", "河南信阳站", "河北成安西站", ] # ============================================================ # Yun8Pipeline — 变量上下文 # ============================================================ class Yun8Pipeline: """ 模拟 Postman Collection Variables 的变量上下文。 每个步骤的输出写入 self.vars，后续步骤从 self.vars 读取。 """ def __init__(self, config: dict, db: Db): # config 为 test_config.yaml 的 yun8_api section self.config = config self.client = Yun8ApiClient(config) self.db = db self.log = get_logger("apis.yun8.pipeline") # 所有步骤间共享的变量池（对应 Postman collectionVariables） self.vars: dict = {} # 回单 file_name 格式模板（旧 config.sql.file_name_template） self.file_name_template = config.get("sql", {}).get( "file_name_template", "{file_prefix}_1_signed.pdf") # 动态 weight 模式：每次循环随机生成 1~10 之间的小数，per_weight 和 weight 保持一致 dp = config.get("driver_params", {}) if dp.get("weight_mode", "fixed") == "random": val = str(round(random.uniform(1.0, 10.0), 1)) self.client.dp["per_weight"] = val self.client.dp["weight"] = val self.log.info(f" [random] weight = {val}") # ------------------------------------------------------------------ # 辅助方法 # ------------------------------------------------------------------ def _set_vars(self, **kwargs): """批量写入变量""" self.vars.update(kwargs) def _get(self, key: str, default: str = "") -> str: """读取变量""" return self.vars.get(key, default) def _log_step(self, num: int, name: str): self.log.info(f"\n{'─' * 50}") self.log.info(f"[Step {num:02d}] {name}") self.log.info(f"{'─' * 50}") def _check_response(self, data: dict, step_name: str): """检查 API 返回 code 是否为 0，非 0 视为步骤失败并中断流程。""" code = data.get("code") msg = data.get("msg", "") if code is not None and code != 0: raise Yun8StepError(step_name, code=code, msg=msg) def _load_lat(self) -> str: """装货纬度: config.load_latitude → 货源 shipperLatitude → config.latitude""" dp = self.config.get("driver_params", {}) return dp.get("load_latitude") or self._get("shipperLatitude") or dp.get("latitude", "") def _load_lng(self) -> str: """装货经度: config.load_longitude → 货源 shipperLongitude → config.longitude""" dp = self.config.get("driver_params", {}) return dp.get("load_longitude") or self._get("shipperLongitude") or dp.get("longitude", "") def _unload_lat(self) -> str: """卸货纬度: config.unload_latitude → 货源 consigneeLatitude → config.latitude""" dp = self.config.get("driver_params", {}) return dp.get("unload_latitude") or self._get("consigneeLatitude") or dp.get("latitude", "") def _unload_lng(self) -> str: """卸货经度: config.unload_longitude → 货源 consigneeLongitude → config.longitude""" dp = self.config.get("driver_params", {}) return dp.get("unload_longitude") or self._get("consigneeLongitude") or dp.get("longitude", "") # ================================================================== # 步骤 1-5 : 司机端流程 # ================================================================== def step01_query_goods(self) -> dict: self._log_step(1, "查询货源") result = self.client.step01_query_goods() self._set_vars(mainOrderCode=result["mainOrderCode"], residueNumber=result["residueNumber"], shipperLatitude=result["shipperLatitude"], shipperLongitude=result["shipperLongitude"], consigneeLatitude=result["consigneeLatitude"], consigneeLongitude=result["consigneeLongitude"]) self.log.info(f" mainOrderCode = {result['mainOrderCode']}") self.log.info(f" residueNumber = {result['residueNumber']}") return result def step02_accept_order(self) -> dict: self._log_step(2, "接单") data = self.client.step02_accept_order( main_order_code=self._get("mainOrderCode"), residue_number=self._get("residueNumber"), latitude=self._load_lat(), longitude=self._load_lng()) self._check_response(data, "接单") self.log.info(f" 响应: code={data.get('code')}, msg={data.get('msg')}") return data def step03_query_pending_transport(self) -> dict: self._log_step(3, "查询待运输单") result = self.client.step03_query_pending_transport() self._set_vars(code=result["code"], vehicleCode=result["vehicleCode"], orderCode=result["orderCode"]) self.log.info(f" code = {result['code']}") self.log.info(f" vehicleCode = {result['vehicleCode']}") self.log.info(f" orderCode = {result['orderCode']}") return result def step04_load_goods(self) -> dict: self._log_step(4, "装货") data = self.client.step04_load_goods( code=self._get("code"), vehicle_code=self._get("vehicleCode"), latitude=self._load_lat(), longitude=self._load_lng()) self._check_response(data, "装货") self.log.info(f" 响应: code={data.get('code')}, msg={data.get('msg')}") return data def step05_unload_goods(self) -> dict: self._log_step(5, "卸货") data = self.client.step05_unload_goods( code=self._get("code"), latitude=self._unload_lat(), longitude=self._unload_lng()) self._check_response(data, "卸货") self.log.info(f" 响应: code={data.get('code')}, msg={data.get('msg')}") return data # ================================================================== # 步骤 6-9 : 运费核算流程 # ================================================================== def step06_query_for_accounting(self) -> dict: self._log_step(6, "运费核算查询运单") record = self.client.step06_query_for_accounting( order_code=self._get("orderCode")) for k, v in record.items(): self.vars[k] = str(v) if v is not None else "" self.log.info(f" 提取字段 ({len(record)}): {', '.join(list(record.keys())[:10])}...") return record def step07_query_order_expenses(self) -> dict: self._log_step(7, "查询运单详情") result = self.client.step07_query_order_expenses(code=self._get("code")) self._set_vars(**result) for k, v in result.items(): self.log.info(f" {k} = {v}") return result def step08_calculate_order_detail(self) -> dict: self._log_step(8, "核算试算 (calculateOrderDetail)") result = self.client.step08_calculate_order_detail(code=self._get("code")) self._set_vars(**result) for k, v in result.items(): self.log.info(f" {k} = {v[:80] if len(str(v)) > 80 else v}") return result def step09_submit_accounting(self) -> dict: self._log_step(9, "核算提交 (orderHandleEditSubmitHandle)") data = self.client.step09_submit_accounting(self.vars) self._check_response(data, "核算提交") self.log.info(f" 响应: code={data.get('code')}, msg={data.get('msg')}") return data # ================================================================== # 步骤 10-16 : 审核 & 付款流程 # ================================================================== def step10_query_audit(self) -> dict: self._log_step(10, "查询运单审核") data = self.client.step10_query_audit(order_code=self._get("orderCode")) self._check_response(data, "查询运单审核") records = data.get("data", {}).get("records", []) self.log.info(f" 记录数: {len(records)}") return data def step11_approve_audit(self) -> dict: self._log_step(11, "核算审核通过") data = self.client.step11_approve_audit(code=self._get("code")) self._check_response(data, "核算审核通过") self.log.info(f" 响应: code={data.get('code')}, msg={data.get('msg')}") return data # Step 12-14 wrappers 不单独定义（已内联在 run_audit_payment_flow 中） def step15_pay_to_driver(self) -> dict: self._log_step(15, "垫资-财务打款") data = self.client.step15_pay_to_driver( code=self._get("code"), delivery_cash_fee=self._get("deliveryCashFee"), driver_actual_freight=self._get("driverActualFreight"), plan_driver_payout=self._get("planDriverPayout", "0"), plan_captain_payout=self._get("planCaptainPayout", "0")) self._check_response(data, "垫资-财务打款") self.log.info(f" 响应: code={data.get('code')}, msg={data.get('msg')}") return data def step16_apply_for_payment(self) -> dict: self._log_step(16, "垫资-货主打款 (申请付款)") data = self.client.step16_apply_for_payment( code=self._get("code"), total_amount=self._get("shipmentActualCash")) self._check_response(data, "垫资-货主打款") self.log.info(f" 响应: code={data.get('code')}, msg={data.get('msg')}") return data # ================================================================== # 步骤 17-21 : 回单/文件名前缀 # ================================================================== def step17_query_upload_for_payno(self) -> dict: """ 14. 查询运单上传 → payNo ───────────────── GET /ntocc-basic-api/dytBusiness/orderDetail (status=5) Test: records[0].sendMsg → regex【(payNo)】 Input: orderCode """ self._log_step(17, "查询运单上传 (提取 payNo)") result = self.client.step17_query_upload_for_payno( order_code=self._get("orderCode"), ) pay_no = result["payNo"] self._set_vars(payNo=pay_no) self.log.info(f" payNo = {pay_no}") if not pay_no: # 迁移行为变化：旧版仅警告并继续（INSERT 空 front_seq_no），现视为失败 raise Yun8StepError("查询运单上传", msg="未提取到 payNo（sendMsg 中无【】包裹内容）") return result def step18_query_remit_records(self) -> list[dict]: """ 15. 查询打款记录 ───────────────── GET /ntocc-tms-api/remitDefinite (分页) Test: records[0].code → remitCode Input: orderCode """ self._log_step(18, "查询打款记录") records = self.client.step18_query_remit_records( order_code=self._get("orderCode"), ) self.log.info(f" 共 {len(records)} 条打款记录") if records: self.log.info(f" 首条字段: {list(records[0].keys())}") return records def step19_query_file_prefix(self, remit_code: str) -> str: """ 16. 查询文件名前缀 ───────────────── GET /ntocc-tms-api/centralizedPayment/downLoadElectronicReceipt?code={remitCode} Test: msg → '文件名前缀:{filePrefix}' """ file_prefix = self.client.step19_query_file_prefix(remit_code) return file_prefix # ================================================================== # 组装流程 # ================================================================== def run_driver_flow(self) -> dict: """仅跑司机端（步骤 1-5）""" self.log.info("\n" + "=" * 60) self.log.info(" 司机端流程 (Step 1-5)") self.log.info("=" * 60) self.step01_query_goods() time.sleep(0.3) self.step02_accept_order() time.sleep(0.3) self.step03_query_pending_transport() # ---- 重置运单 (条件触发) ---- dp = self.config.get("driver_params", {}) if dp.get("reset_order_detail", False): code = self._get("code") order_code = self._get("orderCode") self.log.info(f"\n [reset] 重置运单 orderCode={order_code}") reset_code = dp.get("reset_order_detail_code", "") with self.db.transaction(): for sql, params in driver_reset_statements(reset_code, order_code): rows = self.db.execute(sql, params) self.log.info(f" SQL: {self.db.mogrify(sql, params)[:80]}... → {rows} rows") time.sleep(0.3) # 重新查询待运输单，刷新变量 self.step03_query_pending_transport() time.sleep(0.3) self.step04_load_goods() time.sleep(0.3) self.step05_unload_goods() # 卸货后 SQL if dp.get("reset_order_detail", False): reset_code = dp.get("reset_order_detail_code", "") sql, params = post_unload_reset_update(reset_code, self._get("orderCode")) rows = self.db.execute(sql, params) self.log.info(f" [reset] 卸货后 UPDATE: {rows} rows") else: self.log.info(" [...] 等待 10s ...") time.sleep(10) sql, params = risk_control_update(self._get("code")) rows = self.db.execute(sql, params) self.log.info(f" 卸货后 risk_control_status UPDATE: {rows} rows") return self.vars def run_accounting_flow(self) -> dict: """运费核算流程（步骤 6-9）""" self.log.info("\n" + "=" * 60) self.log.info(" 运费核算流程 (Step 6-9)") self.log.info("=" * 60) self.step06_query_for_accounting() time.sleep(0.5) self.step07_query_order_expenses() time.sleep(0.3) self.step08_calculate_order_detail() time.sleep(0.3) self.step09_submit_accounting() return self.vars def _retry_on_processing(self, step_fn, step_name: str, max_retries: int = 5): """ 遇到 "处理中" 错误时自动重试，每次间隔递增。 服务端审核通过后需要异步处理，可能持续数秒到十几秒。 """ for attempt in range(1, max_retries + 1): data = step_fn() msg = str(data.get("msg", "")) code = data.get("code") if code == 0 or (msg and ("成功" in msg or "success" in msg.lower())): return data if "处理中" in msg or "稍后" in msg: wait = 3 + attempt * 2 # 5, 7, 9, 11, 13 秒 self.log.info(f" [...] {step_name} 服务端正处理中，第 {attempt}/{max_retries} 次重试（等待 {wait}s）...") time.sleep(wait) else: # 非 "处理中" 错误，直接返回 return data return data def run_audit_payment_flow(self) -> dict: """审核 & 付款流程（步骤 10-16）""" self.log.info("\n" + "=" * 60) self.log.info(" 审核 & 付款流程 (Step 10-16)") self.log.info("=" * 60) # 确保前置依赖已获取（独立运行 --stage payment 时 vars 可能为空） self._ensure_code_and_order_code() self.step10_query_audit() time.sleep(0.5) self.step11_approve_audit() self.log.info(" [...] 等待服务端处理审核结果...") ac = self.config.get("accounting", {}) if not ac.get("execute_after_audit", True): self.log.info(f"\n [stop] execute_after_audit=false，跳过步骤 12-21") return self.vars # ---- 队长分账 (条件触发：payerReceiverPair 非空 + captainSplitMode=2) ---- payer_receiver = ac.get("payer_receiver", "") payer_link_man = ac.get("payer_receiver_link_man", "") captain_split_mode = ac.get("captain_split_mode") or self._get("captainSplitMode") if (payer_receiver or payer_link_man) and captain_split_mode == "2": self._log_step(12, "队长分账预览 (captainSplitAdjustPreview)") preview = self.client.step12_captain_split_preview( order_detail_code=self._get("code"), ) self._check_response(preview, "队长分账预览") latest_driver = float(preview.get("data", {}).get("latestDriverAmount", 0)) oil_fee = float(preview.get("data", {}).get("deliveryOilFee", 0)) self.log.info(f" latestDriverAmount = {latest_driver}") self.log.info(f" deliveryOilFee = {oil_fee}") time.sleep(0.5) self._log_step(13, "队长分账调整 (adjustCaptainSplit)") # 计算 driverAmount/captainAmount (与 client 中计算逻辑一致) import math as _m _driver_amt = _m.ceil(latest_driver * 0.8 * 100) / 100.0 _captain_amt = round(latest_driver - _driver_amt, 2) adjust = self.client.step13_adjust_captain_split( order_detail_code=self._get("code"), latest_driver_amount=latest_driver, delivery_oil_fee=oil_fee, ) self._check_response(adjust, "队长分账调整") self.log.info(f" 响应: code={adjust.get('code')}, msg={adjust.get('msg')}") # 存入 vars，供 step13 财务打款使用 self._set_vars(planDriverPayout=str(_driver_amt), planCaptainPayout=str(_captain_amt)) self.log.info(f" planDriverPayout = {_driver_amt}") self.log.info(f" planCaptainPayout = {_captain_amt}") else: self.log.info(f" [skip] 队长分账 (payerReceiver='{payer_receiver}', captainSplitMode={captain_split_mode})") time.sleep(0.5) # ---- Step 14: singleOil 带重试 ---- self._log_step(14, "单油卡充值 (singleOil)") oil_amount = self._get("oilFee") data_12 = self._retry_on_processing( lambda: self.client.step14_single_oil( order_detail_code=self._get("code"), amount=oil_amount if oil_amount else "0.03", ), "单油卡充值", ) self._check_response(data_12, "singleOil") self.log.info(f" 响应: code={data_12.get('code')}, msg={data_12.get('msg')}") # Step 13 带重试 self._log_step(15, "垫资-财务打款") data_12 = self._retry_on_processing( lambda: self.client.step15_pay_to_driver( code=self._get("code"), delivery_cash_fee=self._get("deliveryCashFee"), driver_actual_freight=self._get("driverActualFreight"), plan_driver_payout=self._get("planDriverPayout", "0"), plan_captain_payout=self._get("planCaptainPayout", "0"), ), "垫资-财务打款", ) self._check_response(data_12, "垫资-财务打款") self.log.info(f" 响应: code={data_12.get('code')}, msg={data_12.get('msg')}") # Step 14 带重试 (totalAmount 取 step06 的 shipmentActualCash) self._log_step(16, "垫资-货主打款 (申请付款)") data_13 = self._retry_on_processing( lambda: self.client.step16_apply_for_payment( code=self._get("code"), total_amount=self._get("shipmentActualCash"), ), "垫资-货主打款", ) self._check_response(data_13, "垫资-货主打款") self.log.info(f" 响应: code={data_13.get('code')}, msg={data_13.get('msg')}") return self.vars def run_receipt_file_flow(self, dry_run: bool = False) -> list[dict]: """回单文件流程（步骤 15-18）+ SQL INSERT""" self.log.info("\n" + "=" * 60) self.log.info(" 回单文件流程 (Step 14-16 + SQL)") self.log.info("=" * 60) # 确保 orderCode 已就绪（独立运行 --stage receipt 时需要） self._ensure_code_and_order_code() # 14. 获取 payNo self.step17_query_upload_for_payno() time.sleep(0.3) # 15. 获取所有打款记录 remit_records = self.step18_query_remit_records() if not remit_records: self.log.info(" [EXIT] 无打款记录") return [] # 16 + SQL: 遍历每条打款记录 pay_no = self._get("payNo") results = [] ok = 0 ng = 0 for i, rec in enumerate(remit_records): remit_code = rec.get("code", "") if not remit_code: self.log.info(f"\n [{i+1}/{len(remit_records)}] 跳过：缺 code 字段") ng += 1 results.append({"remit_code": remit_code, "status": "skip_no_code"}) continue self.log.info(f"\n [{i+1}/{len(remit_records)}] remitCode = {remit_code}") # 16. 查询文件名前缀 file_prefix = self.step19_query_file_prefix(remit_code) self.log.info(f" filePrefix = {file_prefix}") if not file_prefix: self.log.info(f" [!] filePrefix 为空，跳过 SQL INSERT") ng += 1 results.append({"remit_code": remit_code, "status": "skip_no_prefix"}) continue file_name = self.file_name_template.format(file_prefix=file_prefix) self.log.info(f" file_name = {file_name}") self.log.info(f" payNo = {pay_no}") # SQL INSERT（参数化；dry-run 用 mogrify 渲染预览） sql = RECEIPT_FILE_INSERT_SQL params = receipt_file_insert_params(file_prefix, pay_no, self.file_name_template) if dry_run: self.log.info(f" [DRY RUN] SQL:\n{self.db.mogrify(sql, params)}") ok += 1 results.append({"remit_code": remit_code, "file_prefix": file_prefix, "status": "dry_run"}) else: try: rowcount = self.db.execute(sql, params) self.log.info(f" [OK] INSERT 成功 (rows: {rowcount})") ok += 1 results.append({"remit_code": remit_code, "file_prefix": file_prefix, "status": "inserted", "rows": rowcount}) except Exception as exc: self.log.info(f" [FAIL] INSERT 失败: {exc}") ng += 1 results.append({"remit_code": remit_code, "file_prefix": file_prefix, "status": "error", "error": str(exc)}) # 汇总 self.log.info(f"\n ───────────────────────") self.log.info(f" 回单文件流程完成: 成功 {ok}, 失败 {ng}, 总计 {len(remit_records)}") # Step 18: 上报运单（迁移行为变化：失败不再吞异常，向上传播） self._log_step(20, "上报运单 (reportOrderDetail)") data_17 = self.client.step20_report_order_detail( codes=self._get("code"), ) self._check_response(data_17, "上报运单") self.log.info(f" 响应: code={data_17.get('code')}, msg={data_17.get('msg')}") # Step 19: 滇运通审核 (独立 token/host，config 控制开关) # （迁移行为变化：失败不再吞异常，向上传播） if self.client.tp_enabled: self._log_step(21, "滇运通审核 (updateOrderDetailExt)") data_19 = self.client.step21_owner_review( order_detail_code=self._get("code"), ) self._check_response(data_19, "滇运通审核") self.log.info(f" 响应: code={data_19.get('code')}, msg={data_19.get('msg')}") else: self.log.info(f"\n [skip] Step 19 滇运通审核 (dianyuntong.enabled=false)") return results # ================================================================== # 步骤 22-24 : ETC 开票 SQL 写入 # ================================================================== def _gen_etc_invoice(self, idx: int, total: int, amount: float) -> dict: """ 生成第 idx 条发票的差异字段（发票号码/金额/进出站/时间）。 金额由调用方传入（已按"所有发票合计 ≤ 运费×50%"约束分配）。 时间规则: transaction_time(=ex_time) 在运单窗口 2026-05-13 15:35:38 ~ 2026-05-15 13:35:43 内按条数均分、逐条推进， en_date=ex_date=该时间日期；invoice_time 在 2026-06-04 ~ 06-06 内随机。 """ invoice_number = "264" + "".join(random.choices("0123456789", k=17)) pair = idx % (len(_ETC_STATION_CHAIN) - 1) w_start = datetime(2026, 5, 13, 15, 35, 38) w_end = datetime(2026, 5, 15, 13, 35, 43) seg = (w_end - w_start) / total trans_time = w_start + seg * idx + timedelta( seconds=random.uniform(0, seg.total_seconds())) invoice_time = datetime(2026, 6, 4) + timedelta( seconds=random.uniform(0, 3 * 86400)) return { "code": _gen_code(), "invoice_number": invoice_number, "total_price": amount, "amount": amount, "import_station": _ETC_STATION_CHAIN[pair], "export_station": _ETC_STATION_CHAIN[pair + 1], "transaction_time": trans_time.strftime("%Y-%m-%d %H:%M:%S"), "ex_time": trans_time.strftime("%Y-%m-%d %H:%M:%S"), "en_date": trans_time.strftime("%Y-%m-%d"), "ex_date": trans_time.strftime("%Y-%m-%d"), "invoice_time": invoice_time.strftime("%Y-%m-%d %H:%M:%S"), } def step22_insert_etc_invoices(self, dry_run: bool = False) -> list[dict]: """ 22. 写入多条 ETC 开票信息 (base_etc_invoice) ───────────────── 每个运单插入 etc_invoice.count 条发票（config.yaml 配置，默认 3）， 每条的 发票号码/金额/进站/出站/时间 各不相同，其余字段按原固定值： order_code ← 步骤03的 orderCode push_order_no ← orderCode + "-1" order_detail_code ← 步骤03的 code code 按现有数据规则自动生成 (32 位十六进制)，每条独立 金额约束: 所有发票金额合计 ≤ 运费(优先 driverActualFreight，其次 shipmentActualFreight) × 50%；运费未知或上限不足时跳过发票写入。 """ self._log_step(22, "ETC 开票信息写入 (base_etc_invoice × N)") order_code = self._get("orderCode") detail_code = self._get("code") if not order_code or not detail_code: self.log.info(" [!] orderCode/code 为空，无法写入 base_etc_invoice") return [{"table": "base_etc_invoice", "status": "skip_no_vars"}] raw_count = self.config.get("etc_invoice", {}).get("count", 3) n = int(raw_count) if raw_count is not None else 3 push_order_no = f"{order_code}-1" self.log.info(f" order_code = {order_code}") self.log.info(f" push_order_no = {push_order_no}") self.log.info(f" order_detail_code = {detail_code}") self.log.info(f" 发票条数 = {n}") if n <= 0: self.log.info(" [skip] etc_invoice.count <= 0，跳过") return [] # ---- 金额约束: 所有发票金额合计 ≤ 运费 × 50% ---- freight_raw = self._get("driverActualFreight") or self._get("shipmentActualFreight") try: freight = float(freight_raw) except (TypeError, ValueError): freight = 0.0 freight_src = ("driverActualFreight" if self._get("driverActualFreight") else "shipmentActualFreight") cap = freight * 0.5 cap_floor = int(cap * 100) / 100 # 向下取整到分，确保不超上限 self.log.info(f" 运费({freight_src}) = {freight:.2f}，发票金额上限(运费×50%) = {cap_floor:.2f}") if cap_floor < 0.01 * n: self.log.info(f" [!] 上限 {cap_floor:.2f} 不足以分配 {n} 张发票（每张至少 0.01），跳过发票写入") return [{"table": "base_etc_invoice", "status": "skip_amount_cap"}] # 随机总额: 上限的 30%~100%，且保证每张至少 0.01；按整数分分配，合计精确 total_amt = min(cap_floor, max(0.01 * n, round(cap_floor * random.uniform(0.3, 1.0), 2))) total_cents = int(round(total_amt * 100)) alloc = [0] * n # 每张保底 1 分 pool = total_cents - n if pool > 0: weights = [random.uniform(0.5, 1.5) for _ in range(n)] wsum = sum(weights) alloc = [int(pool * w / wsum) for w in weights] for _ in range(pool - sum(alloc)): alloc[random.randrange(n)] += 1 amounts = [(1 + a) / 100 for a in alloc] # 写入前显式判断（按构造不应触发，作兜底保护） if round(sum(amounts), 2) > round(cap, 2): self.log.info(f" [!] 发票合计 {sum(amounts):.2f} 超过运费×50% {cap:.2f}，跳过发票写入") return [{"table": "base_etc_invoice", "status": "skip_amount_cap"}] self.log.info(f" 发票金额合计 = {sum(amounts):.2f}（{n} 张，各 {['%.2f' % a for a in amounts]}）≤ 上限，校验通过") results = [] for i in range(n): inv = self._gen_etc_invoice(i, n, amounts[i]) sql = ETC_INVOICE_INSERT_SQL params = etc_invoice_insert_params(inv, order_code, push_order_no, detail_code) self.log.info(f" [{i+1}/{n}] 发票号={inv['invoice_number']} 金额={inv['total_price']:.2f} " f"{inv['import_station']}→{inv['export_station']} " f"交易时间={inv['transaction_time']} 开票时间={inv['invoice_time']}") if dry_run: self.log.info(f" [DRY RUN] SQL:\n{self.db.mogrify(sql, params)}") results.append({"table": "base_etc_invoice", "invoice_number": inv["invoice_number"], "status": "dry_run"}) continue try: rows = self.db.execute(sql, params) self.log.info(f" [OK] INSERT 成功 (rows: {rows})") results.append({"table": "base_etc_invoice", "invoice_number": inv["invoice_number"], "status": "inserted", "rows": rows}) except Exception as exc: self.log.info(f" [FAIL] INSERT 失败: {exc}") results.append({"table": "base_etc_invoice", "invoice_number": inv["invoice_number"], "status": "error", "error": str(exc)}) ok = sum(1 for r in results if r["status"] in ("inserted", "dry_run")) ng = len(results) - ok self.log.info(f" base_etc_invoice 完成: 成功 {ok}, 失败 {ng}, 总计 {n}") return results def step23_insert_etc_push(self, dry_run: bool = False) -> dict: """ 23. 写入 ETC 推送记录 (tms_order_etc_push) ───────────────── INSERT INTO ntocc.tms_order_etc_push order_no ← 步骤03的 orderCode push_order_no ← orderCode + "-1" order_detail_code ← 步骤03的 code code 按现有数据规则自动生成 (32 位十六进制) """ self._log_step(23, "ETC 推送记录写入 (tms_order_etc_push)") order_code = self._get("orderCode") detail_code = self._get("code") if not order_code or not detail_code: self.log.info(" [!] orderCode/code 为空，无法写入 tms_order_etc_push") return {"table": "tms_order_etc_push", "status": "skip_no_vars"} etc_code = _gen_code() push_order_no = f"{order_code}-1" sql = ETC_PUSH_INSERT_SQL params = (etc_code, push_order_no, order_code, detail_code) self.log.info(f" order_no = {order_code}") self.log.info(f" push_order_no = {push_order_no}") self.log.info(f" order_detail_code = {detail_code}") self.log.info(f" code = {etc_code}") if dry_run: self.log.info(f" [DRY RUN] SQL:\n{self.db.mogrify(sql, params)}") return {"table": "tms_order_etc_push", "status": "dry_run"} try: rows = self.db.execute(sql, params) self.log.info(f" [OK] INSERT 成功 (rows: {rows})") return {"table": "tms_order_etc_push", "status": "inserted", "rows": rows} except Exception as exc: self.log.info(f" [FAIL] INSERT 失败: {exc}") return {"table": "tms_order_etc_push", "status": "error", "error": str(exc)} def step24_update_order_etc_status(self, dry_run: bool = False) -> dict: """ 24. 更新运单 ETC 开票状态 (tms_order_detail) ───────────────── UPDATE ntocc.tms_order_detail SET etc_invoice_status = 3 WHERE order_code = 步骤03的 orderCode """ self._log_step(24, "更新运单 ETC 开票状态 (tms_order_detail.etc_invoice_status=3)") order_code = self._get("orderCode") if not order_code: self.log.info(" [!] orderCode 为空，无法更新 tms_order_detail") return {"table": "tms_order_detail", "status": "skip_no_vars"} sql, params = etc_status_update(order_code) self.log.info(f" order_code = {order_code}") if dry_run: self.log.info(f" [DRY RUN] SQL:\n{self.db.mogrify(sql, params)}") return {"table": "tms_order_detail", "status": "dry_run"} try: rows = self.db.execute(sql, params) self.log.info(f" [OK] UPDATE 成功 (rows: {rows})") return {"table": "tms_order_detail", "status": "updated", "rows": rows} except Exception as exc: self.log.info(f" [FAIL] UPDATE 失败: {exc}") return {"table": "tms_order_detail", "status": "error", "error": str(exc)} def run_etc_invoice_flow(self, dry_run: bool = False) -> list[dict]: """ETC 开票 SQL 写入（步骤 22-24）""" self.log.info("\n" + "=" * 60) self.log.info(" ETC 开票 SQL 写入 (Step 22-24)") self.log.info("=" * 60) # 确保 orderCode / code 已就绪（独立运行 --stage receipt 时需要） self._ensure_code_and_order_code() results = [] results.extend(self.step22_insert_etc_invoices(dry_run=dry_run)) results.append(self.step23_insert_etc_push(dry_run=dry_run)) results.append(self.step24_update_order_etc_status(dry_run=dry_run)) return results def _ensure_code_and_order_code(self): """ 确保 code / orderCode 等关键变量已就绪。 独立运行 --stage payment/receipt 时 vars 为空， 需自动补充。优先尝试 driver 端 step03（waitTrans 状态）， 若订单已过 driver 阶段则降级为 admin 端查询（step10 接口）。 """ if self._get("code") and self._get("orderCode"): self.log.info(f" (复用已有 code={self._get('code')}, orderCode={self._get('orderCode')})") return self.log.info(" [*] 自动补充前置数据...") # 方式 A: 通过 driver 端 step03（status=waitTrans） driver_auth = self.client.driver_http.get_header("authorization") if driver_auth: try: result = self.client.step03_query_pending_transport() self._set_vars( code=result["code"], vehicleCode=result["vehicleCode"], orderCode=result["orderCode"], ) self.log.info(f" [driver] code={self._get('code')}, orderCode={self._get('orderCode')}") except Exception: self.log.info(" driver 端无待运输单，切换 admin 端查询...") driver_auth = "" # 标记失败，走降级 # 方式 B: 通过 admin 端（status=6 已结算 / status=5 已打款） if not self._get("code") or not self._get("orderCode"): for status_val in ["6", "5"]: try: # status=6 用 step10（/ntocc-tms-api/orderDetail?status=6&isSettle=1） if status_val == "6": data = self.client.step10_query_audit( order_code="", page_size=5, ) records = data.get("data", {}).get("records", []) else: # status=5 用 step17（/ntocc-basic-api/dytBusiness/orderDetail?status=5） data = self.client.step17_query_upload_for_payno( order_code="", page_size=5, ) records = data.get("raw_records", []) if records: r = records[0] self._set_vars( code=r.get("code", ""), orderCode=r.get("orderCode", ""), ) if self._get("code"): self.log.info(f" [admin status={status_val}] code={self._get('code')}, orderCode={self._get('orderCode')}") break except Exception as e: self.log.debug(f" [admin status={status_val}] 查询失败: {e}") continue if not self._get("code"): self.log.info(" [!] 无法自动获取 code，请使用 --stage all 完整运行或手动指定") return # 补充核算数据（payment 阶段需要 deliveryCashFee/driverActualFreight） if not self._get("deliveryCashFee"): try: record = self.client.step06_query_for_accounting( order_code=self._get("orderCode"), ) for k, v in record.items(): self.vars[k] = str(v) if v is not None else "" expenses = self.client.step07_query_order_expenses( code=self._get("code"), ) self._set_vars(**expenses) self.log.info(f" deliveryCashFee={self._get('deliveryCashFee')}, driverActualFreight={self._get('driverActualFreight')}") except Exception as e: self.log.info(f" [!] 补充核算字段失败: {e}") def set_auth(self, driver_token: str, admin_token: str): """设置授权 Token""" self.client.set_driver_auth(driver_token) self.client.set_admin_auth(admin_token) self.log.info(f" Driver auth: {driver_token[:30]}...") self.log.info(f" Admin auth: {admin_token[:30]}...") def close(self): """关闭 HTTP 会话连接池（数据库连接由调用方管理）。""" self.client.close()