Files
auto-pickup-v2/field_mapping.py
T
SkyLiCoder e99dfaef4d 初始化项目 v2.0:中国邮政揽收系统自动填单工具
基于 HTTP API + Node.js token bridge 方案,不依赖浏览器自动化。
2026-05-15 14:39:31 +08:00

226 lines
6.7 KiB
Python

"""字段映射模块 — 将 Excel 列名映射到页面 DOM 元素 ID。
所有中文名→DOM ID 的对应关系写死在 BUILTIN_FIELD_MAP 中。
config.json 的 field_mapping 用于追加/覆盖自定义映射。
"""
import json
from pathlib import Path
from typing import Any
# 内置字段映射:中文名 → 页面 DOM 元素 ID(写死在代码中)
BUILTIN_FIELD_MAP: dict[str, str] = {
# ── 业务字段 ──
"单号": "waybillNo",
"运单号": "waybillNo",
"产品名称": "bizProductName",
"客户名称": "senders",
# ── 收件人 ──
"收件人": "receiverLinker",
"收件人手机": "receiverMobile",
"收件地址": "receiverAddr",
"收件邮编": "receiverPostcode",
"收件人固话": "receiverFixtel",
# ── 寄件人 ──
"寄件人": "senderLinker",
"寄件人手机": "senderMobile",
"寄件地址": "senderAddr",
"寄件邮编": "senderPostcode",
"寄件人固话": "senderFixtel",
# ── 货物 ──
"重量": "realWeight",
"内件品名": "reserved5",
# ── 电商 / 支付 ──
"电商平台": "ecommerceNo",
"付款方式": "paymentMode",
"证件类型": "consignorIdType",
"保价金额": "insuranceAmount",
"代收货款": "codAmount",
# ── 系统保留字段(通常不需要在 Excel 中填写,只用默认值)──
"运输类型": "transferType",
"内件属性": "contentsAttribute",
"件数": "quantity",
"保价标志": "insuranceFlag",
"回单标志": "receiptFlag",
"代收标志": "codFlag",
"投递方式": "deliverType",
# ── 其他 ──
"寄达地": "receiverArriveOrgName",
"寄达地区划": "receiverDistrictNo",
"备注": "senderNotes",
}
def load_config(config_path: str = "config.json") -> dict[str, Any]:
"""加载配置文件。"""
# 支持开发环境和 PyInstaller 打包后的路径
paths_to_try = [
Path(config_path),
Path(__file__).parent / config_path,
Path.cwd() / config_path,
]
# PyInstaller _MEIPASS
import sys
if getattr(sys, "frozen", False):
paths_to_try.insert(0, Path(sys._MEIPASS) / config_path)
for p in paths_to_try:
if p.exists():
with open(p, "r", encoding="utf-8") as f:
return json.load(f)
raise FileNotFoundError(f"配置文件未找到: {config_path}")
def get_field_mapping(config: dict) -> dict[str, str]:
"""获取 Excel 列名 → DOM ID 的映射(内置 + 配置覆盖)。"""
mapping = dict(BUILTIN_FIELD_MAP)
mapping.update(config.get("field_mapping", {}))
return mapping
def get_defaults(config: dict) -> dict[str, str]:
"""获取默认值(中文键 → 值)。"""
return config.get("defaults", {})
def get_system_config(config: dict) -> dict[str, Any]:
"""获取系统配置。"""
return config.get("system", {})
def get_timing_config(config: dict) -> dict[str, Any]:
"""获取延时配置。"""
return config.get("timing", {})
def get_fixed_values(config: dict) -> dict[str, dict]:
"""获取固定值设置,返回 {字段名: {value, force}}。"""
return config.get("fixed_values", {})
def get_fixed_value_str(fixed_values: dict[str, dict], field: str) -> str:
"""从固定值配置中提取字段的字符串值。"""
return fixed_values.get(field, {}).get("value", "")
# 中文显示名 → 系统内部编码(用于默认值/固定值的翻译)
# 数据来源:data.har 抓包分析
CODE_MAP: dict[str, dict[str, str]] = {
"电商平台": {
"无": "WU",
"菜鸟电子面单": "TAOBAOWAYBILL",
"淘宝": "TAOBAO",
"拼多多": "PINDUODUO",
"电子面单": "DZMD",
"大客户订单接口": "DKH",
"云打印": "CLOUD",
"批量导入": "IMPO",
"客户端": "EMSRM",
"公安部": "GAB",
"电子返单": "DZFD",
"仓库配送": "CKPS",
"无": "WU",
},
"付款方式": {
"寄付": "1",
"到付": "2",
},
"证件类型": {
"身份证": "01",
},
"运输类型": {
"陆运": "1",
},
"内件属性": {
"其他": "00",
"身份证": "01",
"录取通知书": "02",
"法院专递": "03",
"护照": "04",
"通行证": "05",
"签证": "06",
"驾驶证": "07",
"保险文书": "08",
"银行票据": "09",
"行驶证": "10",
"交通违章通知书": "11",
"纳税申报": "12",
"到货通知书": "13",
"公证书": "14",
"学生档案": "15",
"准考证": "16",
"成绩单": "17",
"药监批件": "18",
"移动配送": "19",
"劳动仲裁": "20",
"合格证": "21",
"健康证": "22",
"火车票": "23",
"房证": "24",
"企业代码证": "25",
"教材": "26",
"福利彩票": "27",
"病历": "28",
"银企对账单": "29",
"学位证书": "30",
"车辆档案": "31",
"交房通知书": "32",
"车牌": "36",
"信用卡": "41",
"检察专递": "53",
"法院专递回执": "54",
},
"保价标志": {
"不保价": "1",
},
"回单标志": {
"无回单": "1",
},
"代收标志": {
"不代收": "9",
},
"投递方式": {
"普通投递": "3",
},
}
def translate_value(excel_col: str, chinese_value: str) -> str:
"""将中文显示名翻译为系统内部编码(如 "寄付" → "1")。
如果该字段没有翻译表,或值已经是编码格式,直接返回原值。
"""
if not chinese_value:
return chinese_value
mapping = CODE_MAP.get(excel_col, {})
return mapping.get(chinese_value, chinese_value)
def reverse_translate(excel_col: str, code_value: str) -> str:
"""将系统编码翻译回中文显示名(如 "1" → "寄付")。"""
if not code_value:
return code_value
mapping = CODE_MAP.get(excel_col, {})
for chinese, code in mapping.items():
if code == code_value:
return chinese
return code_value
def resolve_field_value(
row: dict[str, str],
excel_col: str,
defaults: dict[str, str],
fixed_values: dict[str, str] = None,
) -> str:
"""解析字段值,优先级:固定值 > Excel > 默认值。结果自动翻译为系统编码。"""
fixed_values = fixed_values or {}
fixed = fixed_values.get(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, defaults.get(excel_col, ""))