"""HTTP 自动化引擎 — 通过直接调用后端 API 完成表单填写和提交。 不依赖浏览器，使用 Node.js token bridge 计算 $_ts 安全 token。 """ import asyncio import copy from datetime import datetime from typing import Any, Callable from log import debug as _ld, trace as _lt, user as _lu from pickup_api import PickupAPI, PickupSession, TokenBridge from field_mapping import get_fixed_value_str, translate_value # ─── Submit 表单构建器 ───────────────────────────────────────────────── # HAR 成功提交的完整字段模板（所有字段初始化为空，逐层覆盖） _FORM_TEMPLATE: dict[str, str] = { "QRcodeTransId": "", "allCargoFileNames": "", "baoLiuTiHuoDan": "", "baoLiuTiHuoDanInsuranceAmount": "", "batteryFlag": "", "bizProductId": "", "bizProductName": "", "bizProductNo": "", "cargoOptionName": "", "cargoOptionNo": "", "cargoPriceNumber": "", "cargoTypeNameEn": "", "caseNo": "", "cha": "", "chang": "", "changeMinGanFlag": "", "changeNoFlag": "", "codAmount": "", "codFlag": "", "consignorIdName": "", "consignorIdNo": "", "consignorIdType": "", "consignorMobile": "", "consignorOtherFlag": "", "contentsAttribute": "", "contentsQuantity": "", "contentsWeight": "", "custRecvPhoneIsEmpty": "", "customerManagerId": "", "customerManagerName": "", "customerManagerNo": "", "deliverDate": "", "deliverPreDate": "", "deliverSign": "", "deliverType": "", "deliverTypeErjiText": "", "deliverTypeSource": "", "discountRate": "", "ecommerceNo": "", "fayuanName": "", "fayuanNo": "", "fffs": "", "gao": "", "goodsDesc": "", "height": "", "identifyFlag": "", "ifInspection": "", "imClient": "", "importantAddress": "", "innerChannel": "", "insuranceAmount": "", "insuranceAmountTempHide": "", "insuranceFlag": "", "isBreakable": "", "isForceFillRecevier": "", "isSpecialMarketing": "", "isStoredCard": "", "jdds": "", "jsxbsx": "", "ksms": "", "kuan": "", "length": "", "lockCustomer": "", "logisticsOrderNo": "", "mailBoxKeyDown": "", "mdjmbs": "", "more3": "", "more4": "", "njxzs": "", "noChangeProductXunShi": "", "oldBizOccurDate": "", "oldConsignorIdName": "", "oldConsignorIdNo": "", "oldConsignorMobile": "", "omsCodFlag": "", "omsContentsAttribute": "", "omsInfoId": "", "omsInfoReserved35": "", "omsMinGanFlag": "", "omsPackageMaterialTypeNo": "", "omsPaymentMode": "", "omsReceiverDistrictNo": "", "omsReceiverMobile": "", "omsSenderMobile": "", "omsXunshiFlag": "", "oneBillFlag": "", "orderId": "", "orderWeight": "", "orgCityCode": "", "orgProvinceCode": "", "packageMaterial": "", "packagehj": "", "packaging": "", "patrolProductFlag": "", "paymentMode": "", "paymentType": "", "pickupOrgIntegrate": "", "pickupPersonName": "", "pickupPersonNo": "", "pickupType": "", "pkpGoodsFlag": "", "pkpWaybillId": "", "postBlend": "", "preWaybillNo": "", "printFlag": "", "printtype": "", "productType": "", "projectId": "", "qinQingFlag": "", "quantity": "", "readRemotePayMobileFlag": "", "realNameCheckFlag": "", "realNameFlag": "", "realNameNotAdjust": "", "realNameNotCheck": "", "realWeight": "", "reasonForTemporaryStorage": "", "receiptFeeAmount": "", "receiptFlag": "", "receiptWaybillNo": "", "receiver": "", "receiverAddr": "", "receiverAddrAdditional": "", "receiverArriveOrgName": "", "receiverDistrictNo": "", "receiverDuty": "", "receiverFixtel": "", "receiverGis": "", "receiverId": "", "receiverImId": "", "receiverImType": "", "receiverLinker": "", "receiverMobile": "", "receiverNo": "", "receiverNotes": "", "receiverOrg": "", "receiverPostcode": "", "receiverSafetyCode": "", "receiverType": "", "receiverWarehouseId": "", "receiverWarehouseName": "", "registeredCustomerNo": "", "remotePayMobile": "", "reserved10": "", "reserved12": "", "reserved19": "", "reserved29": "", "reserved31": "", "reserved4": "", "reserved45": "", "reserved5": "", "reserved9": "", "reusablePackagingCode": "", "secretType": "", "securityProtocol": "", "securityProtocolTitle": "", "sender": "", "senderAddr": "", "senderAddrAdditional": "", "senderAgentIdNo": "", "senderAgentIdType": "", "senderClientType": "", "senderDistrictNo": "", "senderFixtel": "", "senderGis": "", "senderId": "", "senderIdEncryptedCode": "", "senderIdNo": "", "senderIdType": "", "senderImId": "", "senderImType": "", "senderIsCheckAddress": "", "senderLinker": "", "senderMobile": "", "senderNo": "", "senderNotes": "", "senderOrg": "", "senderPostcode": "", "senderRemotePayMobile": "", "senderSafetyCode": "", "senderSignType": "", "senderType": "", "senders": "", "settlementMode": "", "signatureURL": "", "signingTime": "", "singleSealFlag": "", "storedValueCardCheckFlag": "", "subBillDelFlag": "", "templateCode": "", "tiHuoDanHaoTemp": "", "transId": "", "transferType": "", "valuableFlag": "", "waybillNo": "", "width": "", "yangGuangBaoXianOmsQuantity": "", "yangGuangBaoXianQuantity": "", "zhudanhao": "", } # 系统固定标志（每次提交都需要，值固定） _SYSTEM_FLAGS: dict[str, str] = { "ksms": "0", "fffs": "0", "jdds": "0", "njxzs": "0", "insuranceFlag": "1", "receiptFlag": "1", "codFlag": "9", "deliverSign": "01", "quantity": "1", "reserved29": "0", "imClient": "0", "yangGuangBaoXianQuantity": "0", "yangGuangBaoXianOmsQuantity": "0", "omsXunshiFlag": "0", "senderClientType": "2", "noChangeProductXunShi": "0", "isForceFillRecevier": "1", "reserved10": '{"receiverAdderCode":"0","secretType":"1"}', "realNameCheckFlag": "1", "pkpGoodsFlag": "1", "consignorOtherFlag": "0", "pickupOrgIntegrate": "0", "secretType": "1", "senderSignType": "2", "mdjmbs": "0", "reserved31": "0", "innerChannel": "0", "settlementMode": "3", "senderType": "1", "printFlag": "0", "printtype": "0", "isSpecialMarketing": "2", "packagehj": "0", "more3": "0", "more4": "0", "cargoPriceNumber": "0", "paymentType": "1", "cha": "1", "deliverType": "3", "pkpWaybillId": "", "reserved4": "0000", "orderWeight": "1000", } class PickupEngine: """揽收系统自动化引擎（HTTP API 模式）。 工作模式： 1. 启动 token bridge（Node.js 子进程） 2. Python 管理 HTTP 会话和 cookies 3. 所有业务操作通过后端 API 直接完成 4. 用户通过 choose/confirm 回调参与首批决策 """ def __init__( self, url: str, field_mapping: dict[str, str], defaults: dict[str, str], timing: dict[str, Any], log_callback: Callable[[str], None] = None, fixed_values: dict[str, str] = None, choose_callback: Callable = None, confirm_callback: Callable = None, ): self.url = url self.field_mapping = field_mapping self.defaults = defaults self.timing = timing self._log_cb = log_callback or print self.fixed_values = fixed_values or {} self._choose = choose_callback self._confirm = confirm_callback # 单次 batch 缓存（batch 结束后清除） self._product_choice: dict | None = None self._customer_choice: dict | None = None self._first_row_confirmed: bool = False # 基础设施 self._sess: PickupSession | None = None self._api: PickupAPI | None = None self._bridge: TokenBridge | None = None self._stop_requested = False self._pause_requested = False self._infra_injected = False # 基础设施从外部注入时跳过 start/begin_processing # 跨行复用的产品/客户详情（从 API 获取） self._product_detail: dict = {} self._customer_detail: dict = {} # 经办人实名信息（证件类型/姓名/证件号/手机，来自 findrealnamehistorybysender API，与客户绑定） self._sender_realname: dict = {} # 地址匹配结果 self._address_match: dict = {} # 运输类型（由 UI 下拉框选择，来自 transmode 数组的 propertyCode） self._transport_type: str = "" def _ulog(self, msg: str) -> None: _lu(msg) def _dlog(self, msg: str) -> None: _ld(msg) # ─── 字段值解析 ────────────────────────────────────────── def _get_value(self, row: dict[str, str], excel_col: str) -> str: """优先级：固定值 > Excel 表格值 > 默认值。中文值自动翻译为系统编码。""" fixed = get_fixed_value_str(self.fixed_values, excel_col) if fixed: return translate_value(excel_col, fixed) value = row.get(excel_col, "").strip() if value: return translate_value(excel_col, value) return translate_value(excel_col, self.defaults.get(excel_col, "")) # ─── 基础设施注入（由外部预先初始化）────────────────────── def inject_infrastructure(self, sess: PickupSession, bridge: TokenBridge, api: PickupAPI): """注入已初始化的 session / bridge / api，跳过 start/begin_processing。""" self._sess = sess self._bridge = bridge self._api = api self._infra_injected = True self._dlog("基础设施已从外部注入") # ─── 生命周期 ──────────────────────────────────────────── async def start(self) -> None: """启动 token bridge 并初始化 HTTP 会话。""" if self._infra_injected: self._dlog("start 跳过（基础设施已注入）") return self._ulog("启动 HTTP API 引擎...") # 启动 token bridge self._bridge = TokenBridge() self._bridge.start() self._dlog("Token bridge 已启动") # 创建 HTTP 会话 self._sess = PickupSession(self.url) # 尝试加载已保存的 cookies session_path = self._get_session_path() if self._sess.load_session(session_path): cookie_names = {c.name for c in self._sess._http.cookies} missing = [n for n in ("JSESSIONID", "jdpt.session.id") if n not in cookie_names] if missing: self._ulog(f"缺少关键 cookie: {', '.join(missing)}，请重新登录并手动填写") else: self._dlog("已加载登录会话") else: self._ulog("未找到登录会话，请先登录") async def begin_processing(self) -> None: """加载 pickup 页面，初始化 token bridge。""" if not self._sess: raise RuntimeError("未启动") if self._infra_injected: self._dlog("begin_processing 跳过（基础设施已注入）") self._ulog("后台自动化已就绪") return self._ulog("加载页面资源...") resources = self._sess.fetch_page_resources() # 构造最小 HTML：只设置 $_ts.cd/nsd，不包含任何会阻塞 jsdom 的内联脚本 minimal_html = ( f"" ) self._dlog(f"初始化 token bridge (cd={resources['ts_cd'][:20]}..., " f"security JS: {len(resources['scripts'])} files)") self._bridge.initialize( minimal_html, resources["scripts"], resources["base_url"], ) self._dlog("Token bridge 已初始化，$_ts 安全框架就绪") # 创建 API 客户端 self._api = PickupAPI(self._sess, self._bridge) # 加载初始数据（电商列表、投递日期等） try: self._dlog("预加载: query_ecommerce...") self._api.query_ecommerce("B") self._dlog("预加载: query_deliver_date...") self._api.query_deliver_date() self._dlog("预加载: check_sheet_warning...") self._api.check_sheet_warning() self._dlog("预加载完成") except Exception as e: self._dlog(f" [warn] 预加载部分失败: {e}") self._ulog("后台自动化已就绪") async def stop(self) -> None: if self._stop_requested: return # 防止重复调用 self._stop_requested = True # 保存 session cookies if self._sess: try: self._sess.save_session(self._get_session_path()) self._dlog("Session 已保存") except Exception: pass # 外部注入的基础设施不销毁，由外部管理生命周期 if self._infra_injected: self._dlog("跳过 bridge 销毁（基础设施由外部管理）") self._clear_batch_cache() return # 停止 token bridge if self._bridge: try: self._bridge.stop() except Exception: pass self._bridge = None self._api = None self._ulog("引擎已停止") # 清除 batch 缓存 self._clear_batch_cache() def _get_session_path(self) -> str: import os as _os, sys as _sys if getattr(_sys, "frozen", False): base = _os.path.dirname(_sys.executable) else: base = _os.path.dirname(_os.path.abspath(__file__)) return _os.path.join(base, "session.json") def _clear_batch_cache(self): self._product_choice = None self._customer_choice = None self._product_detail = {} self._customer_detail = {} self._sender_realname = {} self._first_row_confirmed = False # ─── 暂停控制 ──────────────────────────────────────────── def request_stop(self) -> None: self._stop_requested = True def request_pause(self) -> None: self._pause_requested = True def resume(self) -> None: self._pause_requested = False async def _check_pause(self) -> None: while self._pause_requested and not self._stop_requested: await asyncio.sleep(0.5) # ─── 单行处理（核心流程）───────────────────────────────── async def process_single_row(self, row: dict[str, str]) -> dict: start_time = datetime.now() result = {"status": "失败", "message": "", "postage": "", "time": "", "寄达地": "", "单号": row.get("单号", "").strip(), "重量": row.get("重量", "").strip(), "产品": "", "客户": "", "运输类型": "", "寄达地区划代码": "", "寄件人姓名": "", "寄件地址": ""} try: await self._check_pause() # 1. 产品选择 self._dlog(" 选择产品...") await self._handle_product(self._get_value(row, "产品名称")) if self._stop_requested: return result await self._check_pause() # 2. 客户选择 self._dlog(" 选择客户...") await self._handle_customer(self._get_value(row, "客户名称")) if self._stop_requested: return result await self._check_pause() # 3. 查询运单 self._dlog(" 查询运单...") await self._handle_waybill(row) if self._stop_requested: return result await self._check_pause() # 4. 收件地址 self._dlog(" 匹配收件地址...") await self._handle_address(row) await self._check_pause() # 5. 货物信息 self._dlog(" 校验货物...") await self._handle_cargo_checks(row) # 6. 组装表单 form_data = self._assemble_form(row) # 7. 首行确认 if not self._first_row_confirmed and self._confirm: if not await self._confirm(form_data): self._stop_requested = True result.update(status="取消", message="用户取消确认，已停止", time=datetime.now().strftime("%H:%M:%S")) self._ulog(" STOP: 用户取消确认") return result self._first_row_confirmed = True await self._check_pause() # 8. 提交 self._dlog(" 提交中...") submit_result = await self._submit_form(form_data) result["寄达地"] = form_data.get("receiverArriveOrgName", "") result["单号"] = row.get("单号", "").strip() result["重量"] = form_data.get("realWeight", "") result["产品"] = self._product_choice.get("productName", "") if self._product_choice else "" result["客户"] = self._customer_choice.get("customerName", "") if self._customer_choice else "" result["寄达地区划代码"] = form_data.get("receiverDistrictNo", "") result["寄件人姓名"] = form_data.get("senderLinker", "") result["寄件地址"] = form_data.get("senderAddr", "") result["运输类型"] = form_data.get("transferType", "") elapsed = (datetime.now() - start_time).total_seconds() if submit_result.get("success"): result.update(status="成功", message=submit_result.get("message", ""), postage=str(submit_result.get("postage", "")), time=datetime.now().strftime("%H:%M:%S")) self._ulog(f" OK ({elapsed:.1f}s): {result['message']}") else: result.update(status="失败", message=submit_result.get("message", "提交失败"), time=datetime.now().strftime("%H:%M:%S")) self._ulog(f" FAIL ({elapsed:.1f}s): {result['message']}") except Exception as e: elapsed = (datetime.now() - start_time).total_seconds() result.update(status="异常", message=str(e), time=datetime.now().strftime("%H:%M:%S")) self._ulog(f" ERROR ({elapsed:.1f}s): {e}") return result # ─── 产品处理 ──────────────────────────────────────────── async def _handle_product(self, product_name: str) -> None: if not product_name: return if self._product_choice is not None: if self._product_detail: return biz_id = self._product_choice.get("productId", "") biz_no = self._product_choice.get("bizProductNo", "") if biz_id and biz_no: self._product_detail = self._api.select_product(biz_id, biz_no) return # 首次选择：搜索 → 让用户选 → 缓存 candidates = self._api.query_product(product_name) self._dlog(f" query_product('{product_name}') → {len(candidates)} candidates: " f"{[c.get('productName','')[:20] for c in candidates[:5]]}") if not candidates: self._ulog(f" [warn] 产品 '{product_name}' 无候选结果，请在固定值中设置正确的产品名称") self.request_stop() return if self._choose: choice = await self._choose( "选择产品", [c.get("productName", c.get("name", str(c))) for c in candidates], ) if choice is None: self._ulog("用户取消产品选择，停止 batch") self.request_stop() return if 0 <= choice < len(candidates): self._product_choice = candidates[choice] if not self._product_choice and candidates: self._product_choice = candidates[0] chosen = self._product_choice if chosen: biz_id = chosen.get("productId", "") biz_no = chosen.get("bizProductNo", "") if biz_no and biz_id: self._product_detail = self._api.select_product(biz_id, biz_no) # ─── 客户处理 ──────────────────────────────────────────── async def _handle_customer(self, customer_name: str) -> None: if not customer_name: return if self._customer_choice is not None: if self._customer_detail: return cust_id = self._customer_choice.get("customerId", "") if cust_id: self._customer_detail = self._api.select_customer(str(cust_id)) if not isinstance(self._customer_detail, dict) or "_text" in self._customer_detail: self._dlog(f" [warn] customerins 返回异常: {str(self._customer_detail)[:200]}") return base = self._customer_detail.get("baseCustome", {}) if isinstance(self._customer_detail.get("baseCustome"), dict) else {} sender_no = base.get("businessCode") or base.get("customerSubCode") or "" if sender_no: try: self._sender_realname = self._api.find_sender_history( str(sender_no) ) except Exception: pass return # 首次选择 candidates = self._api.query_customer(customer_name) self._dlog(f" query_customer('{customer_name}') → {len(candidates)} candidates: " f"{[c.get('customerName','')[:20] for c in candidates[:5]]}") if not candidates: self._ulog(f" [warn] 客户 '{customer_name}' 无候选结果，请在固定值中设置正确的客户名称") self.request_stop() return if self._choose: choice = await self._choose( "选择客户", [c.get("customerName", c.get("name", str(c))) for c in candidates], ) if choice is None: self._ulog("用户取消客户选择，停止 batch") self.request_stop() return if 0 <= choice < len(candidates): self._customer_choice = candidates[choice] if not self._customer_choice and candidates: self._customer_choice = candidates[0] chosen = self._customer_choice if chosen: cust_id = chosen.get("customerId", "") if cust_id: self._customer_detail = self._api.select_customer(str(cust_id)) if not isinstance(self._customer_detail, dict) or "_text" in self._customer_detail: self._dlog(f" [warn] customerins 返回异常: {str(self._customer_detail)[:200]}") return base = self._customer_detail.get("baseCustome", {}) if isinstance(self._customer_detail.get("baseCustome"), dict) else {} sender_no = base.get("businessCode") or base.get("customerSubCode") or "" if sender_no: try: self._sender_realname = self._api.find_sender_history( str(sender_no) ) except Exception: pass # ─── 运单查询 ──────────────────────────────────────────── async def _handle_waybill(self, row: dict[str, str]) -> None: waybill_no = self._get_value(row, "单号") ecommerce = self._get_value(row, "电商平台") payment = self._get_value(row, "付款方式") biz_product_no = ( self._product_choice.get("bizProductNo", "") if self._product_choice else "" ) biz_product_id = self._product_detail.get("bizProductId", "") result = self._api.query_waybill( waybill_no=waybill_no, ecommerce_no=ecommerce, biz_product_no=biz_product_no, payment_mode=payment, biz_product_id=str(biz_product_id), ) self._waybill_result = result if isinstance(result.get("data"), dict): wd = result["data"] key_vals = {k: wd[k] for k in ["senderLinker","senderMobile","receiverLinker","receiverMobile","receiverAddr","senderAddr","senderNo","internalInfoId","bizProductNo","bizProductName"] if wd.get(k)} self._dlog(f" waybill key_vals: {key_vals}") else: self._dlog(f" waybill result top: {list(result.keys())[:5]}, data type: {type(result.get('data'))}") # ─── 地址处理 ──────────────────────────────────────────── async def _handle_address(self, row: dict[str, str]) -> None: receiver_addr = self._get_value(row, "收件地址") if not receiver_addr: return # 1. 用实际地址查询区划代码（addressMatch API） dist_code = "" try: match = self._api.address_match_raw(receiver_addr) if isinstance(match, dict): dist_code = match.get("districtCode", "") or match.get("contyCode", "") if not dist_code: dist_code = match.get("provCode", "") self._address_match = match self._dlog(f" addressMatch → dist_code={dist_code}") except Exception: self._address_match = {} self._dlog(" [warn] addressMatch 失败") # 如果 addressMatch 没结果，尝试用运单数据中的区划代码 if not dist_code: if isinstance(self._waybill_result, dict): wr_data = self._waybill_result.get("data", {}) if isinstance(wr_data, str): try: import json as _json wr_data = _json.loads(wr_data) except Exception: wr_data = {} if isinstance(wr_data, dict): dist_code = wr_data.get("receiverDistrictNo", "") if dist_code: try: self._api.query_district(dist_code) except Exception: self._dlog(" [warn] 区划查询失败") # 2. 地址网格匹配 biz_product_no = ( self._product_choice.get("bizProductNo", "") if self._product_choice else "" ) waybill_no = self._get_value(row, "单号") sender_no = "" if self._customer_detail: sender_no = (self._customer_detail.get("baseCustome", {})).get("businessCode") or "" transfer_type = self._get_value(row, "运输类型") contents_attr = self._get_value(row, "内件属性") try: grid_result = self._api.match_address( receiver_arrive_org_no=dist_code, base_product_no=biz_product_no, waybill_no=waybill_no, receiver_addr=receiver_addr, sender_no=sender_no, transfer_type=transfer_type, contents_attribute=contents_attr, ) self._grid_result = grid_result except Exception as e: self._dlog(f" [warn] 地址匹配失败: {e}") self._grid_result = {} # ─── 货物校验 ──────────────────────────────────────────── async def _handle_cargo_checks(self, row: dict[str, str]) -> None: cargo_names = self._get_value(row, "内件品名") waybill_no = self._get_value(row, "单号") weight = self._get_value(row, "重量") biz_product_no = ( self._product_choice.get("bizProductNo", "") if self._product_choice else "" ) # 寄达区划代码 dist_code = "" if hasattr(self, '_grid_result') and isinstance(self._grid_result, dict): dist_code = self._grid_result.get("receiverDistrictNo", dist_code) # 客户代码 cust_code = "" if self._customer_detail: cust_code = "" if isinstance(self._customer_detail, dict): base = self._customer_detail.get("baseCustome", {}) if isinstance(base, dict): cust_code = base.get("businessCode") or "" # 校验产品物料 try: self._api.check_product_material(cargo_names, waybill_no, biz_product_no, dist_code) except Exception: pass # 校验可达区域 try: self._api.check_reach_area(biz_product_no, dist_code, cust_code, weight, waybill_no) except Exception: pass # ─── 表单提交 ──────────────────────────────────────────── def _assemble_form(self, row: dict[str, str]) -> dict[str, str]: """组装完整表单数据。优先级：强制固定值 > API > 固定值 > 表格 > 默认值。""" form = dict(_FORM_TEMPLATE) form.update(_SYSTEM_FLAGS) # ── 第一层：默认值 ── for col, param in self.field_mapping.items(): val = self.defaults.get(col, "") if val: form[param] = translate_value(col, val) # ── 第二层：表格值 ── for col, param in self.field_mapping.items(): val = row.get(col, "").strip() if val: form[param] = translate_value(col, val) waybill_no = row.get("单号", "").strip() if waybill_no: form["logisticsOrderNo"] = waybill_no # ── 第三层：固定值（非强制）── for col_name, cfg in self.fixed_values.items(): if not cfg.get("force") and cfg.get("value"): param = self.field_mapping.get(col_name, col_name) form[param] = translate_value(col_name, cfg["value"]) # ── 第四层：API 获取值 ── if self._product_choice: form["bizProductNo"] = self._product_choice.get("bizProductNo", "") form["bizProductName"] = self._product_choice.get("productName", "") if self._product_detail: for k in ("bizProductId", "templateCode"): if self._product_detail.get(k): form[k] = str(self._product_detail[k]) # 产品的内件性质（queryproductbymanager 返回 innerpartsType） ipt = self._product_detail.get("innerpartsType", "") if ipt: form["contentsAttribute"] = str(ipt) # 运输类型（UI 下拉框选择，来自 transmode 数组） if self._transport_type: form["transferType"] = self._transport_type if self._customer_choice: # senders 用全名（带前缀如 qxsm-穹诩商贸），sender 用短名 full_name = self._customer_choice.get("customerFullName", "") short_name = self._customer_choice.get("customerName", "") if full_name: form["senders"] = full_name if short_name: form["sender"] = short_name if isinstance(self._customer_detail, dict): base = self._customer_detail.get("baseCustome", {}) if isinstance(self._customer_detail.get("baseCustome"), dict) else {} cn = base.get("customerNo", "") if cn: form["senderNo"] = str(cn) cid = base.get("id") or self._customer_detail.get("customerId", "") if cid: form["senderId"] = str(cid) wr = self._waybill_result if hasattr(self, '_waybill_result') and isinstance(self._waybill_result, dict) else {} data_raw = wr.get("data", {}) if isinstance(data_raw, str): try: import json as _json wr_data = _json.loads(data_raw) self._dlog(f" waybill data parsed from JSON string, keys={len(wr_data)}") except Exception as e: self._dlog(f" waybill data JSON parse failed: {e}, raw={data_raw[:100]}") wr_data = {} elif isinstance(data_raw, dict): wr_data = data_raw else: wr_data = {} if not wr_data: self._ulog(f" [warn] 运单 data 为空，可能 ecommerceNo 不正确，当前值: {self._get_value(row, '电商平台')}") # 运单中有值的字段填入（白名单：HAR 确认需要的字段） wr_whitelist = { "ecommerceNo", "senderLinker", "senderMobile", "senderPostcode", "senderDistrictNo", "senderFixtel", "receiverLinker", "receiverMobile", "receiverPostcode", "receiverDistrictNo", "receiverArriveOrgNo", "receiverArriveOrgName", "receiverSafetyCode", "internalInfoId", "contentsAttribute", "transferType", "quantity", "deliverType", "reserved10", "senderProvinceName", "senderCityName", "senderCountyName", "receiverProvinceName", "receiverCityName", "receiverCountyName", } for key in wr_whitelist: val = wr_data.get(key) if val is not None and val != "": form[key] = str(val) # 拼接完整地址：省+市+县+地址 sp = wr_data.get("senderProvinceName", "") sc = wr_data.get("senderCityName", "") sn = wr_data.get("senderCountyName", "") sa = wr_data.get("senderAddr", "") full_sender = str(sp or "") + str(sc or "") + str(sn or "") + str(sa or "") if full_sender.strip(): form["senderAddr"] = full_sender.strip() rp = wr_data.get("receiverProvinceName", "") rc = wr_data.get("receiverCityName", "") rn = wr_data.get("receiverCountyName", "") ra = wr_data.get("receiverAddr", "") full_receiver = str(rp or "") + str(rc or "") + str(rn or "") + str(ra or "") if full_receiver.strip(): form["receiverAddr"] = full_receiver.strip() # receiverAddrAdditional 也拼接省市区 if wr_data.get("receiverAddrAdditional"): full_radd = str(rp or "") + str(rc or "") + str(rn or "") + str(wr_data["receiverAddrAdditional"]) form["receiverAddrAdditional"] = full_radd.strip() for key in wr_whitelist: val = wr_data.get(key) if val is not None and val != "": form[key] = str(val) if wr_data.get("internalInfoId"): form["pkpWaybillId"] = str(wr_data["internalInfoId"]) # 寄达地：用 city+county 拼接 city = wr_data.get("receiverCityName", "") county = wr_data.get("receiverCountyName", "") if city or county: form["receiverArriveOrgName"] = str(city) + str(county) # 省份/城市代码 sp = wr_data.get("senderProvinceNo", "") sc = wr_data.get("senderCityNo", "") if sp: form["orgProvinceCode"] = sp[:2] if sc: form["orgCityCode"] = sc[:4] # 地址匹配结果（addressMatch API）的区划代码 if hasattr(self, '_address_match') and isinstance(self._address_match, dict): dc = self._address_match.get("districtCode") or self._address_match.get("contyCode") or "" if dc: form["receiverDistrictNo"] = str(dc) # receiverArriveOrgName 用 prov+city+county 拼接 parts = [self._address_match.get(k, "") for k in ("provName", "cityName", "contyName")] org_name = "".join(p for p in parts if p) if org_name: form["receiverArriveOrgName"] = org_name if hasattr(self, '_grid_result') and isinstance(self._grid_result, dict): for k in ("receiverArriveOrgName", "receiverDistrictNo"): if self._grid_result.get(k): form[k] = str(self._grid_result[k]) # senderNo 使用 businessCode/customerSubCode（不是 customerNo） if self._customer_detail and isinstance(self._customer_detail, dict): base = self._customer_detail.get("baseCustome", {}) if isinstance(base, dict): sn = base.get("businessCode") or base.get("customerSubCode") or "" if sn: form["senderNo"] = str(sn) # customerManager for data_k, form_k in [ ("accountMgrName", "customerManagerName"), ("accountMgrCode", "customerManagerNo"), ("id", "customerManagerId"), ]: v = base.get(data_k) if v: form[form_k] = str(v) # 寄件人地址（客户资料中的完整地址） cust_addr = base.get("addressStreet", "") if cust_addr: form["senderAddr"] = str(cust_addr) # 寄件人联系方式 if base.get("contactName"): form.setdefault("senderLinker", str(base["contactName"])) if base.get("contactPhone"): form.setdefault("senderMobile", str(base["contactPhone"])) # 实名信息（在 detail[0] 下） if isinstance(self._sender_realname, dict): detail0 = self._sender_realname.get("detail", [{}]) sr = detail0[0] if isinstance(detail0, list) and detail0 else {} for sr_key, form_key in [ ("consignorIdType", "consignorIdType"), ("consignorIdName", "consignorIdName"), ("consignorIdNo", "consignorIdNo"), ("consignorMobile", "consignorMobile"), ("consignorIdName", "oldConsignorIdName"), ("consignorIdNo", "oldConsignorIdNo"), ("consignorMobile", "oldConsignorMobile"), ]: if sr.get(sr_key): form[form_key] = str(sr[sr_key]) # ── 第五层（最高）：强制覆盖固定值 ── for col_name, cfg in self.fixed_values.items(): if cfg.get("force") and cfg.get("value"): param = self.field_mapping.get(col_name, col_name) form[param] = translate_value(col_name, cfg["value"]) # ── 衍生字段（setdefault 不覆盖 API 值）── form.setdefault("receiverAddrAdditional", form.get("receiverAddr", "")) if form.get("receiverMobile"): form.setdefault("omsReceiverMobile", form["receiverMobile"]) if form.get("senderMobile"): form.setdefault("senderFixtel", form["senderMobile"]) form.setdefault("senderSafetyCode", form["senderMobile"]) form.setdefault("omsSenderMobile", form["senderMobile"]) if form.get("senderNotes", ""): pass # already set above elif row.get("备注", "").strip(): form["senderNotes"] = row["备注"].strip() return form async def _submit_form(self, form_data: dict[str, str]) -> dict: """提交已组装好的表单数据。""" if not self._api: return {"success": False, "message": "API 未初始化"} # 输出有值的表单字段用于对比 HAR non_empty = {k: v for k, v in sorted(form_data.items()) if v} self._dlog(f" submit fields ({len(non_empty)}): {non_empty}") response = self._api.submit(form_data) res_mess = response.get("resMess", "") comment = response.get("comment", "") return { "success": "收寄成功" in res_mess, "message": res_mess, "postage": str(comment) if comment else "", } # ─── 批量处理 ──────────────────────────────────────────── async def process_all( self, rows: list[dict[str, str]], progress_callback: Callable[[int, int], None] = None, ) -> list[dict]: results = [] total = len(rows) delay = self.timing.get("delay_between_rows_ms", 3000) / 1000 try: for i, row in enumerate(rows): if self._stop_requested: self._ulog("用户停止") break self._ulog(f"\n[{i+1}/{total}] 运单号: {row.get('运单号', 'N/A')}") results.append(await self.process_single_row(row)) if progress_callback: progress_callback(i + 1, total) if i < total - 1 and not self._stop_requested: await asyncio.sleep(delay) finally: # batch 结束后清除缓存，下次重新选择 self._clear_batch_cache() return results