84 lines
2.7 KiB
Python
84 lines
2.7 KiB
Python
"""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
|