xst
52b73f456c
- tests/core: 新增 BaseHttpClient 统一 HTTP 封装（直连绕行/超时/日志/自定义编码）； db_helper 追加 Db 长连接类（事务/mogrify/自动重连），既有函数不动 - tests/apis/yun8: 21 个 step 方法按域拆分为 5 个 Mixin（driver/accounting/ audit_payment/receipt/third_party），Yun8Pipeline 24 步编排原样迁移， SQL 全部参数化并集中至 sql_templates - tests/api: 5 阶段有序 P0 E2E 用例（业务失败由 print 改为真实报错）+ README 迁移（删除明文密码段，修正步骤编号漂移） - yun8_sync 目录退役：配置并入 test_config.yaml 的 yun8_api 段； 造数 CLI 保留（python -m tests.apis.yun8.cli，兼容旧 config.yaml 结构） - pytest.ini 注册 api/yun8_sync marker；requirements.txt 补 requests>=2.28 - 静态验证: compileall / 全仓 collect 24 条 / marker 过滤 / CLI --help / 客户端与 pipeline 离线实例化 / SQL 占位符核对 / cross_end 兼容
918 lines
42 KiB
Python
918 lines
42 KiB
Python
"""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()
|