diff --git a/yun8_sync/README.md b/yun8_sync/README.md index b6ee266..5b80c1f 100644 --- a/yun8_sync/README.md +++ b/yun8_sync/README.md @@ -22,8 +22,9 @@ │ 阶段 D: 回单文件 + SQL (Step 15-18) │ │ 运单上传(payNo) → 打款记录 → 文件名前缀 → INSERT → 上报运单 │ ├─────────────────────────────────────────────────────────────────┤ -│ 阶段 E: ETC 开票 SQL (Step 22-23) │ -│ 写入 base_etc_invoice → 写入 tms_order_etc_push │ +│ 阶段 E: ETC 开票 SQL (Step 22-24) │ +│ 写入 base_etc_invoice (多条) → 写入 tms_order_etc_push │ +│ → 更新 tms_order_detail 开票状态 │ └─────────────────────────────────────────────────────────────────┘ ``` @@ -236,5 +237,8 @@ yun8_sync/ | — | SQL INSERT | — | `ntocc.base_gd_mft_file` | — | | 22 | ETC 开票信息写入 | SQL INSERT | — | `ntocc.base_etc_invoice` | — | | 23 | ETC 推送记录写入 | SQL INSERT | — | `ntocc.tms_order_etc_push` | — | +| 24 | 运单 ETC 开票状态更新 | SQL UPDATE | — | `ntocc.tms_order_detail` | — | +> 步骤 22 每个运单插入 `etc_invoice.count` 条发票（config.yaml 配置，默认 3），每条的**发票号码**（随机 20 位数字）、**金额**（随机分配，**所有发票合计 ≤ 运费×50%**，运费优先取 `driverActualFreight`、缺失时取 `shipmentActualFreight`，超限或运费未知则跳过发票写入）、**进站/出站**（按广东→河北路段链取相邻两站）、**时间**（交易时间在运单窗口内逐条推进）各不相同；税额、开票方、发票类型、URL 等其余字段用固定值。 > 步骤 22/23 的 `order_code`/`order_no` 取步骤 03 的 `orderCode`，`push_order_no` 为其后追加 `-1`，`order_detail_code` 取步骤 03 的 `code`，两个表的 `code` 按现有数据规则（32 位十六进制）自动生成，随 receipt 阶段执行（`--dry-run` 仅预览不写库）。 +> 步骤 23 的推送记录固定写入 `invoice_pull_status=2`、`start_status=1`、`end_status=1`、`start_result='成功'`、`end_result='成功'`；步骤 24 将步骤 03 `orderCode` 对应运单的 `etc_invoice_status` 置为 3。 diff --git a/yun8_sync/config.yaml b/yun8_sync/config.yaml index 951cc14..2675ddc 100644 --- a/yun8_sync/config.yaml +++ b/yun8_sync/config.yaml @@ -31,8 +31,8 @@ api: # 通过命令行 --auth-driver / --auth-admin 传入，或在此写默认值 auth: # driver_token: "Bearer NRWt5TGNYgZOMYWimJ21xTzYYgEZMYD45Dz3lZEOYcTOZJG4jTT55QMZYFDj" # 司机端 Bearer token (含 Bearer 前缀) - driver_token: "Bearer mJ25zW3NYIYYZQcYMVQZ2QFNmTj5wkzMYMFZlj2NzDWk2mj12mT33QMMMIjM" # 司机端 Bearer token (含 Bearer 前缀) - admin_token: "Bearer Z=AZ3I3YlTTzhTGNOFDO5DDx1YhYMYz5hTTOOQUZkFTl5kFOMYdY3QWM2Em=" # 管理端 Bearer token (含 Bearer 前缀) + driver_token: "Bearer YQZOMJWOYljNZJGN0XT23kFONQJZmDRNlm20lTTYYAUY4YjhiZTj0cDlhDjw" # 司机端 Bearer token (含 Bearer 前缀) + admin_token: "Bearer Z=BZ1gdZYkDN2Ij4kzTw3EGN0GnN2TYYhJcMzQmYNUZNNEWZjNWwmTTOhQD=" # 管理端 Bearer token (含 Bearer 前缀) # ----- 司机端固定参数（装货/卸货） ----- driver_params: @@ -90,6 +90,10 @@ payment: phone: "13888888888" pay_type: "arpa_2" +# ----- ETC 开票 (步骤 22) ----- +etc_invoice: + count: 3 # 每个运单插入的发票条数; 每条发票号码/金额/进出站/时间自动变化, 其余字段用固定值; 金额合计受运费×50%约束 + # ----- 数据库配置 ----- database: host: "36.137.133.24" diff --git a/yun8_sync/pipeline.py b/yun8_sync/pipeline.py index 4bb6da8..c71561b 100644 --- a/yun8_sync/pipeline.py +++ b/yun8_sync/pipeline.py @@ -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):