初始化项目 v2.0:中国邮政揽收系统自动填单工具
基于 HTTP API + Node.js token bridge 方案,不依赖浏览器自动化。
This commit is contained in:
@@ -0,0 +1,83 @@
|
||||
"""Excel 读取模块 — 读取输入 Excel 并写入结果 Excel。"""
|
||||
|
||||
import openpyxl
|
||||
|
||||
|
||||
def read_input_excel(file_path: str) -> tuple[list[dict], list[str]]:
|
||||
"""读取输入 Excel 文件,返回 (行数据列表, 列名列表)。
|
||||
|
||||
每行数据为 {列名: 值} 的字典。跳过全空行。
|
||||
"""
|
||||
wb = openpyxl.load_workbook(file_path, read_only=True)
|
||||
ws = wb.active
|
||||
|
||||
rows = list(ws.iter_rows(min_row=1, values_only=True))
|
||||
if not rows:
|
||||
raise ValueError("Excel 文件为空")
|
||||
|
||||
headers = [str(h).strip() if h is not None else "" for h in rows[0]]
|
||||
if not any(headers):
|
||||
raise ValueError("未找到表头行")
|
||||
|
||||
data = []
|
||||
for row in rows[1:]:
|
||||
values = [str(c).strip() if c is not None else "" for c in row]
|
||||
if not any(values): # 跳过全空行
|
||||
continue
|
||||
row_dict = {}
|
||||
for i, header in enumerate(headers):
|
||||
if i < len(values):
|
||||
row_dict[header] = values[i]
|
||||
else:
|
||||
row_dict[header] = ""
|
||||
data.append(row_dict)
|
||||
|
||||
wb.close()
|
||||
return data, headers
|
||||
|
||||
|
||||
def validate_required_columns(
|
||||
headers: list[str],
|
||||
fixed_values: dict[str, str] = None,
|
||||
defaults: dict[str, str] = None,
|
||||
) -> list[str]:
|
||||
"""验证必要字段是否有数据来源(固定值 > Excel > API > 默认值)。
|
||||
|
||||
数据来源优先级:
|
||||
1. 固定值设置(客户名称、寄件人、寄件地址、产品名称、寄达地)
|
||||
2. Excel 表格(运单号、重量、内件品名)
|
||||
3. API 自动获取(querypostwaybillmanager 返回收件人/电话/地址等全部运单信息)
|
||||
4. 默认值(电商平台、付款方式等系统编码)
|
||||
|
||||
只在所有来源都没有值时警告。
|
||||
"""
|
||||
fixed_values = fixed_values or {}
|
||||
defaults = defaults or {}
|
||||
|
||||
# querypostwaybillmanager API 自动返回的字段,Excel 无需填写
|
||||
api_fields = [
|
||||
"收件人", "收件人手机", "收件地址", "收件邮编", "收件人固话",
|
||||
"寄件人", "寄件人手机", "寄件地址", "寄件邮编",
|
||||
]
|
||||
|
||||
# 其余需要人工提供数据的字段
|
||||
check_fields = [
|
||||
"单号", # Excel 必填(API 不返回)
|
||||
"重量", # Excel 必填(API 返回 orderWeight 但可能为 0)
|
||||
"产品名称", # 固定值
|
||||
"客户名称", # 固定值
|
||||
]
|
||||
|
||||
from field_mapping import get_fixed_value_str
|
||||
|
||||
missing = []
|
||||
for field in check_fields:
|
||||
if get_fixed_value_str(fixed_values, field):
|
||||
continue
|
||||
if field in headers:
|
||||
continue
|
||||
if defaults.get(field, ""):
|
||||
continue
|
||||
missing.append(field)
|
||||
|
||||
return missing
|
||||
Reference in New Issue
Block a user