xst/Yb-QaAutomationHub
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(yun8_sync): 重写 ETC 开票步骤 (Step 22-23) 及配套配置文档

Browse Source 
- pipeline.py: ETC 开票信息写入 base_etc_invoice + 推送记录写入
  tms_order_etc_push, 随 receipt 阶段执行; 每单发票条数可配
  (etc_invoice.count), 发票号/金额/进出站/时间自动变化,
  金额合计受运费×50%约束; order_code/order_no 取步骤 03 的
  orderCode, push_order_no 追加 -1, dry-run 仅预览不写库
- config.yaml: 新增 etc_invoice 配置节 + 刷新登录 token
- README.md: 阶段 E 流程图、接口表 Step 22/23 及取值说明
This commit is contained in:
xst
2026-09-02 17:44:52 +08:00
parent 625a589bfc
commit 224fe88ea6
3 changed files with 190 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
yun8_sync/pipeline.py
+178 -49
View File
@@ -16,8 +16,10 @@
"""
import json
import random
import time
import uuid
from datetime import datetime, timedelta
from typing import Optional
from client import Yun8Client
@@ -29,6 +31,13 @@ def _gen_code() -> str:
return uuid.uuid4().hex
# ETC 发票路段链（广东→河北，按行驶方向依次经过；每条发票取相邻两站作为进/出站）
_ETC_STATION_CHAIN = [
"广东古镇东站", "湖南郴州西站", "湖南岳阳东站",
"湖北孝感东站", "河南信阳站", "河北成安西站",
]
# ============================================================
# Pipeline — 变量上下文
# ============================================================
@@ -583,62 +592,154 @@ class Pipeline:
return results
# ==================================================================
# 步骤 22-23 : ETC 开票 SQL 写入
# 步骤 22-24 : ETC 开票 SQL 写入
# ==================================================================
def step22_insert_etc_invoice(self, dry_run: bool = False) -> dict:
def _gen_etc_invoice(self, idx: int, total: int, amount: float) -> dict:
"""
22. 写入 ETC 开票信息 (base_etc_invoice)
生成第 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)
─────────────────
INSERT INTO ntocc.base_etc_invoice
order_code ← 步骤03的 orderCode
push_order_no ← orderCode + "-1"
order_detail_code ← 步骤03的 code
code 按现有数据规则自动生成 (32 位十六进制)
每个运单插入 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)")
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:
print(" [!] orderCode/code 为空，无法写入 base_etc_invoice")
return {"table": "base_etc_invoice", "status": "skip_no_vars"}
return [{"table": "base_etc_invoice", "status": "skip_no_vars"}]
etc_code = _gen_code()
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"
sql = (
"INSERT INTO `ntocc`.`base_etc_invoice` ("
"`id`, `code`, `order_code`, `push_order_no`, `order_detail_code`, "
"`invoice_time`, `sale_tax_code`, `sale_name`, `import_station`, `export_station`, "
"`transaction_time`, `total_price`, `amount`, `tax_rate`, `tax_amount`, "
"`invoice_type`, `invoice_code`, `invoice_number`, `waybill_start_time`, `waybill_end_time`, "
"`invoice_url`, `invoice_html_url`, `en_date`, `ex_date`, `ex_time`, "
"`is_valid`, `task_name`, `plate_num`, `etc_invoice_status`, `tax_deduction_status`, "
"`sign`, `is_deleted`, `created_by`, `modified_by`, `gmt_created`, `gmt_modified`) "
"VALUES ("
f"NULL, '{etc_code}', '{order_code}', '{push_order_no}', '{detail_code}', "
"'2026-06-04 23:23:51', '9141010055833160XE', '河南省视博电子股份有限公司', '广东古镇东站', '河北成安西站', "
"'2026-05-14 19:03:42', 0.02, 0.02, 0.03, 0.03, "
"'征税发票', '*', '26417910230600041165', '2026-05-13 15:35:38', '2026-05-15 13:35:43', "
"'https://hdfs.txffp.com/e/1102e1en66nPPP1ChAeC161e1C6Pei16.zip?Expires=1812122633&OSSAccessKeyId=LTAIYAJbICRhyCD8&Signature=TLjlHgKtCxwWpPliPRQ0xQ6HdKY%3D', "
"'https://gateway.txffp.com/alle/common/htmlurl/getinvoice/41012526221804003093/1102e1en66nPPP1ChAeC161e1C6Pei16', "
"'2026-05-14', '2026-05-14', '2026-05-14 19:03:42', "
"0, 'etc开票申请2026070604', '冀D8E016', 3, 0, "
"NULL, 0, '', '04b7e539c77d4bdbb4fe4e87eaaf41e4', '2026-07-06 09:55:16', '2026-07-06 10:03:26')"
)
print(f" order_code = {order_code}")
print(f" push_order_no = {push_order_no}")
print(f" order_detail_code = {detail_code}")
print(f" code = {etc_code}")
print(f" 发票条数 = {n}")
if n <= 0:
print(" [skip] etc_invoice.count <= 0，跳过")
return []
if dry_run:
print(f" [DRY RUN] SQL:\n{sql}")
return {"table": "base_etc_invoice", "status": "dry_run"}
# ---- 金额约束: 所有发票金额合计 ≤ 运费 × 50% ----
freight_raw = self._get("driverActualFreight") or self._get("shipmentActualFreight")
try:
rows = self.db.execute(sql)
print(f" [OK] INSERT 成功 (rows: {rows})")
return {"table": "base_etc_invoice", "status": "inserted", "rows": rows}
except Exception as exc:
print(f" [FAIL] INSERT 失败: {exc}")
return {"table": "base_etc_invoice", "status": "error", "error": str(exc)}
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 # 向下取整到分，确保不超上限
print(f" 运费({freight_src}) = {freight:.2f}，发票金额上限(运费×50%) = {cap_floor:.2f}")
if cap_floor < 0.01 * n:
print(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):
print(f" [!] 发票合计 {sum(amounts):.2f} 超过运费×50% {cap:.2f}，跳过发票写入")
return [{"table": "base_etc_invoice", "status": "skip_amount_cap"}]
print(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 = (
"INSERT INTO `ntocc`.`base_etc_invoice` ("
"`id`, `code`, `order_code`, `push_order_no`, `order_detail_code`, "
"`invoice_time`, `sale_tax_code`, `sale_name`, `import_station`, `export_station`, "
"`transaction_time`, `total_price`, `amount`, `tax_rate`, `tax_amount`, "
"`invoice_type`, `invoice_code`, `invoice_number`, `waybill_start_time`, `waybill_end_time`, "
"`invoice_url`, `invoice_html_url`, `en_date`, `ex_date`, `ex_time`, "
"`is_valid`, `task_name`, `plate_num`, `etc_invoice_status`, `tax_deduction_status`, "
"`sign`, `is_deleted`, `created_by`, `modified_by`, `gmt_created`, `gmt_modified`) "
"VALUES ("
f"NULL, '{inv['code']}', '{order_code}', '{push_order_no}', '{detail_code}', "
f"'{inv['invoice_time']}', '9141010055833160XE', '河南省视博电子股份有限公司', "
f"'{inv['import_station']}', '{inv['export_station']}', "
f"'{inv['transaction_time']}', {inv['total_price']:.2f}, {inv['amount']:.2f}, 0.03, 0.03, "
f"'征税发票', '*', '{inv['invoice_number']}', '2026-05-13 15:35:38', '2026-05-15 13:35:43', "
"'https://hdfs.txffp.com/e/1102e1en66nPPP1ChAeC161e1C6Pei16.zip?Expires=1812122633&OSSAccessKeyId=LTAIYAJbICRhyCD8&Signature=TLjlHgKtCxwWpPliPRQ0xQ6HdKY%3D', "
"'https://gateway.txffp.com/alle/common/htmlurl/getinvoice/41012526221804003093/1102e1en66nPPP1ChAeC161e1C6Pei16', "
f"'{inv['en_date']}', '{inv['ex_date']}', '{inv['ex_time']}', "
"0, 'etc开票申请2026070604', '冀D8E016', 3, 0, "
"NULL, 0, '', '04b7e539c77d4bdbb4fe4e87eaaf41e4', '2026-07-06 09:55:16', '2026-07-06 10:03:26')"
)
print(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:
print(f" [DRY RUN] SQL:\n{sql}")
results.append({"table": "base_etc_invoice",
"invoice_number": inv["invoice_number"],
"status": "dry_run"})
continue
try:
rows = self.db.execute(sql)
print(f" [OK] INSERT 成功 (rows: {rows})")
results.append({"table": "base_etc_invoice",
"invoice_number": inv["invoice_number"],
"status": "inserted", "rows": rows})
except Exception as exc:
print(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
print(f" base_etc_invoice 完成: 成功 {ok}, 失败 {ng}, 总计 {n}")
return results
def step23_insert_etc_push(self, dry_run: bool = False) -> dict:
"""
@@ -669,8 +770,8 @@ class Pipeline:
"VALUES ("
f"NULL, '{etc_code}', '{push_order_no}', '{order_code}', '{detail_code}', "
"'2026-07-07 15:08:25', '2026-07-07 15:08:28', 2026, 6, 1, "
"NULL, '11111', '11111', 0, '2026-07-07 11:40:21', "
"2, '2026-07-07 15:17:02', '2', 2, '2026-07-07 15:11:04', 'dsdsad', "
"NULL, '11111', '11111', 2, '2026-07-07 11:40:21', "
"1, '2026-07-07 15:17:02', '成功', 1, '2026-07-07 15:11:04', '成功', "
"0, NULL, NULL, '2026-07-07 11:09:17', '2026-07-07 15:15:18')"
)
print(f" order_no = {order_code}")
@@ -689,19 +790,47 @@ class Pipeline:
print(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:
print(" [!] orderCode 为空，无法更新 tms_order_detail")
return {"table": "tms_order_detail", "status": "skip_no_vars"}
sql = (f"UPDATE `ntocc`.`tms_order_detail` SET etc_invoice_status = 3 "
f"WHERE order_code = '{order_code}'")
print(f" order_code = {order_code}")
if dry_run:
print(f" [DRY RUN] SQL:\n{sql}")
return {"table": "tms_order_detail", "status": "dry_run"}
try:
rows = self.db.execute(sql)
print(f" [OK] UPDATE 成功 (rows: {rows})")
return {"table": "tms_order_detail", "status": "updated", "rows": rows}
except Exception as exc:
print(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-23）"""
"""ETC 开票 SQL 写入（步骤 22-24）"""
print("\n" + "=" * 60)
print(" ETC 开票 SQL 写入 (Step 22-23)")
print(" ETC 开票 SQL 写入 (Step 22-24)")
print("=" * 60)
# 确保 orderCode / code 已就绪（独立运行 --stage receipt 时需要）
self._ensure_code_and_order_code()
results = [
self.step22_insert_etc_invoice(dry_run=dry_run),
self.step23_insert_etc_push(dry_run=dry_run),
]
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):