海外仓每日发货 Agent — 从4小时到3分钟

业务背景 · 跨境电商海外仓

每天发货，到底在做什么？

一个真实的跨境电商海外仓场景——Temu、Amazon、TikTok Shop 多平台运营，每日需要把多家物流商的面单与订单精准配对。

📊

多店铺，多格式

每个平台导出的 WMS Excel 列名不同，需要手动整合成仓库要求的标准格式

📄

PDF 顺序随机

物流商生成的面单 PDF，页面顺序和 Excel 行顺序不一致，仓库打印必须对齐

⏱️

每天重复 1-2 小时

手动翻表、复制粘贴、检查核对——高重复、低价值，但出错代价极高

❌

发错包裹的风险

一旦面单和包裹对应错误，退货率上升，客诉增加，严重时平台封号

适用场景

跨境电商海外仓 · 物流分拣中心
多平台店铺运营（Temu / Amazon / TikTok Shop）
GOFO / UPS / USPS / SWX 等多物流商混发

改造前痛点

2名运营 × 每天 3-4 小时手工对单
出错导致发错包裹，退货率上升
繁重重复工作，人员流失率高

Agent 接管后

3 分钟内处理完毕
运单号严格一一对应，零漏单
2 名运营解放，专注更高价值工作

Agent 的目标

输入：散乱的 WMS Excel + 面单 PDF（多文件，顺序不定）
输出：7 个标准文件
      行李箱-WMS.xlsx / 行李箱.xlsx / 行李箱.pdf
      非行李箱-WMS.xlsx / 非行李箱.xlsx / 非行李箱.pdf
      拣货表.xlsx
要求：零容错 — 运单号无法匹配必须报错，不允许跳过

逐步拆解

每一步的方法与思考

点击展开每个步骤，查看输入输出、核心算法、工具选型和演示场景。

初始化分类 + 早期校验

按文件名分行李箱/非行李箱，并立即验证 PDF 总页数 = Excel 总行数

▼

输入

输入目录下所有 .xlsx 和 .pdf 文件
文件名包含「行李箱」/「非行李箱」

输出

分类子目录（行李箱/excel、非行李箱/pdf 等）
校验报告：通过 ✅ 或精确报错 ❌

文件名分类逻辑

NON_LUGGAGE_KEYWORDS = ['非行李箱', 'non-luggage']
LUGGAGE_KEYWORDS     = ['行李箱', 'luggage']

def is_luggage(filename):
    # 先检查排除词，防止「非行李箱」被误判为行李箱
    if any(k in filename for k in NON_LUGGAGE_KEYWORDS):
        return False
    return any(k in filename for k in LUGGAGE_KEYWORDS)

早期校验 — Fail Fast

excel_rows = sum(len(pd.read_excel(f, dtype=str)) for f in excels)
pdf_pages  = sum(len(pymupdf.open(f)) for f in pdfs)

if excel_rows != pdf_pages:
    raise_error(f"行李箱: Excel {excel_rows} 行 ≠ PDF {pdf_pages} 页")

工具选型

pathlib.Path — 文件扫描
pd.read_excel(nrows=1) — 快速读表头，不读全表
pymupdf.open() — 快速统计页数
shutil.copy — 分类到子目录

演示场景

故意少放一页 PDF，展示精确报错：

❌ 非行李箱: Excel 33行 ≠ PDF 30页
❌   Excel: ['非行李箱-wms.xlsx']
❌   PDF: ['非行李箱.pdf']

💡 思考方式：Fail Fast 原则

越早发现错误越好。如果 PDF 少一页，与其让后续步骤跑完一半再崩溃，不如第一步就终止并给出明确提示。这是生产系统最重要的设计原则之一。

Excel 合并与标准化

多店铺 Excel 合并，中文列名映射为 43 列英文标准，缺失字段自动填充

▼

输入

原始 WMS Excel（23列中文表头）
可能有多个文件需合并

输出

标准 WMS Excel（43列英文）
简化版 Excel（7列）

列名映射（中文 → 英文）

COLUMN_MAPPING = {
    '仓库代码': 'Warehouse Code/仓库代码',
    '销售平台': 'Sales Platform/销售平台',
    '店铺账号': 'Store/店铺',
    '运单号':   'Tracking No./物流跟踪号',
    # ... 共 23 个映射
}

# 目标列名从模版文件读，不硬编码
wb_tpl = load_workbook(TPL_LUGGAGE_WMS)
target_columns = [c.value for c in wb_tpl['小包出库单'][1] if c.value]

⚠️ 关键陷阱：22 位数字精度丢失

USPS 运单号是 22 位整数，超出 float64 精度（~15位）。pandas 默认把数字单元格转 float，读回来就变成错误的值。
解决：pd.read_excel(f, dtype=str) 强制所有列读为字符串。

演示：加 dtype=str 前后的差异

# ❌ 默认读取（float64 精度丢失）
df['运单号'].iloc[-1]  →  9234690390868213319132  # 错的！

# ✅ dtype=str
df['运单号'].iloc[-1]  →  9234690390868213725049  # 正确

承运商自动推断

CARRIER_PREFIXES = [
    ('GFUS', 'GOFO'), ('1Z', 'UPS'),
    ('SWX',  'SWX'),  ('9234', 'USPS'),
]
def infer_carrier(tracking_no):
    for prefix, carrier in CARRIER_PREFIXES:
        if tracking_no.startswith(prefix):
            return carrier
    return ''

PDF 条形码匹配与重排 ★ 核心算法

扫描 PDF 每页条形码，按 Excel 行顺序建立一一对应，重组输出 PDF

▼

输入

WMS Excel（有序，n 行）
面单 PDF（无序，n 页）

输出

重排 PDF（页顺序 = Excel 行顺序）
或精确的匹配失败报错

三个阶段串联执行：

阶段 1：扫描所有 PDF 页，建注册表

page_registry = {}  # {条码字符串: (pdf路径, 第几页)}

for page in pdf_doc:
    # 方法1: pyzbar 条形码识别（主力）
    img = page.get_pixmap(dpi=300)  # pymupdf 渲染
    for bc in pyzbar.decode(PIL_image):
        code = bc.data.decode().strip()
        code = re.sub(r'[\x00-\x1f]', '', code)  # 去GS1控制字符
        page_registry[code] = (pdf_file, page_num)

    # 方法2: pymupdf 文字层兜底（条码扫不出时）
    for token in page.get_text().split():
        if len(token) >= 8:
            page_registry[token] = (pdf_file, page_num)

阶段 2：双向尾部匹配算法

def _matches(excel_t, barcode):
    a = re.sub(r'[^A-Z0-9]', '', excel_t.upper())   # Excel运单号
    b = re.sub(r'[^A-Z0-9]', '', barcode.upper())   # PDF条码
    tail_len = max(min(len(a), 10), 8)
    tail = a[-tail_len:]
    return (tail in b) or (a in b) or (b in a)

# USPS GS1-128 示例：
# PDF 条码原始: 42091744 + \x1d + 9234690390868213725049
# 去控制字符后: 420917449234690390868213725049  ← 更长
# Excel 运单号:           9234690390868213725049  ← 是子串
# → (a in b) = True ✅

阶段 3：按 Excel 顺序重组 PDF

used_pages = set()  # 防止一页被两个运单号抢占

writer = PdfWriter()
for tracking_no in excel_trackings:  # 严格按 Excel 行顺序
    pdf_file, page_num = tracking_to_page[tracking_no]
    writer.add_page(PdfReader(pdf_file).pages[page_num - 1])

writer.write(output_pdf)  # 输出页序 = Excel 行序 ✅

匹配失败时的报错输出

❌ 未匹配 2 个运单号，无法在 PDF 中找到对应页面：

❌ 运单号: 9234690390868213319132

❌ 订单号: PO-211-18097245105273269

❌ 店铺: MOPlus

→ 运营可直接定位是哪个店铺的哪单出问题，去核查面单

工具选型

pymupdf — PDF渲染，纯Python无外部依赖
pyzbar — 条形码识别，支持CODE128/QR
pypdf — 拆页、重组PDF
PIL/Pillow — 格式桥接

为什么不直接按页码匹配？

面单是多物流商、多批次混合 PDF，没有固定顺序。只有条形码里的运单号才是全局唯一标识。

添加水印

每页叠加 SCS 标识 + 日期，直接覆盖原文件，不产生额外副本

▼

for page in reader.pages:
    packet = io.BytesIO()
    c = canvas.Canvas(packet, pagesize=(w, h))
    c.setFont("Helvetica-Bold", 30)
    c.setFillColorRGB(0.6, 0.6, 0.6, alpha=0.4)  # 半透明
    c.drawRightString(w-15, h-25, "SCS")
    c.drawString(15, 15, display_date)
    c.save()

    wm_page = PdfReader(packet).pages[0]
    page.merge_page(wm_page)  # 叠合水印层
    writer.add_page(page)

# 直接覆盖原文件（不生成 -水印.pdf 副本）
writer.write(pdf_file)

💡 设计决策：水印原地覆盖

早期版本生成 -水印.pdf 副本，导致每类输出变成 4 个文件。用户需要的是精确 3 个文件（WMS Excel / 简化 Excel / PDF）。改为原地覆盖，输出目录整洁，无冗余文件。

生成拣货表

汇总行李箱 + 非行李箱所有 SKU 数量，合并输出一张拣货表

▼

sku_counts = defaultdict(int)

for excel_file in [行李箱_WMS, 非行李箱_WMS]:
    df = pd.read_excel(excel_file, dtype=str)
    for i in range(len(df)):
        sku = df[sku_col].iloc[i]
        qty = int(df[qty_col].iloc[i] or 0)
        sku_counts[sku] += qty

# 写入拣货表模版
for row_i, (sku, qty) in enumerate(sorted(sku_counts.items()), 2):
    ws.cell(row=row_i, column=1, value=sku)
    ws.cell(row=row_i, column=2, value=qty)

拣货表输出示例

✅ 拣货表 → 拣货表-20260316.xlsx (5 个SKU，共22件)

SKU 数量

250 3

251 1

252 7

264 9

GD-2012 2

关键工程决策

为什么这样设计？

每一个决策背后，都有一个踩过的坑或一个明确的业务约束。

决策 01

模版驱动输出格式

不做在代码里硬写 43 个列名

做从 templates/ 模版文件读列名，按模版写数据

原因格式变了只改模版文件，不改代码；仓库换格式要求时零改动

决策 02

全程 dtype=str 读 Excel

不做让 pandas 自动推断数据类型

做所有 read_excel() 强制 dtype=str

原因22 位 USPS 单号超出 float64 精度，自动推断会读出错误数字

决策 03

零容错匹配，立即报错

不做匹配不上就跳过，生成不完整文件

做任一运单号找不到对应 PDF 页，报错并给出运单号+订单号+店铺

原因发货场景不能有漏单；宁可不发，不能错发

决策 04

Fail Fast 早期校验

不做跑完 Step 3 一半才发现 PDF 少页

做Step 1 立即校验 PDF 总页数 = Excel 总行数

原因越早发现错误，排查成本越低；精确报出哪组文件有问题

决策 05

水印原地覆盖

不做生成 *-水印.pdf 副本，输出 4 个文件/类

做直接覆盖原 PDF，输出严格 3 个文件/类

原因用户要求每类精确 3 个文件；副本增加混淆和误操作风险

决策 06

pymupdf 替代 pdftoppm

不做subprocess 调用 pdftoppm，生成临时图片文件

做pymupdf 直接在内存渲染为 PIL Image

原因无外部系统依赖，无临时文件，速度更快，跨平台

方法论

可复用到其他 Agent 项目的原则

这些原则不只属于这个项目——它们是构建可靠文件处理 Agent 的通用方法。

📋

明确输入输出契约

每步函数有精确的输入类型和返回值，步骤间靠返回值传递，不共享全局状态

⚡

Fail Fast

第一步就做完整性校验，错误不传递到下一步，报错信息要有业务上下文

🎯

数据来源权威性

Excel 是权威，PDF 必须向 Excel 对齐。永远不允许为了让程序通过而修改权威数据源

📁

模版驱动

格式规范在文件里，不在代码里。业务变更只改模版，不改代码

🔧

选对工具

pymupdf 替代 pdftoppm（无外部依赖）；dtype=str 防精度丢失；优先纯 Python 方案

🔍

可观测性

每步有清晰日志，匹配过程逐行打印，失败给出完整业务上下文（不只是技术错误码）

🚫

零容错原则

发货场景宁可报错停止，不可静默跳过。任何不完整的输出比没有输出更危险

🧹

整洁输出

输出文件数量有明确约定（每类 3 个），不产生中间文件、副本、调试文件

2人×4小时的发货工作Agent 3分钟内完成

每天发货，到底在做什么？

多店铺，多格式

PDF 顺序随机

每天重复 1-2 小时

发错包裹的风险

Agent 的目标

五步流水线

每一步的方法与思考

初始化分类 + 早期校验

输入

输出

工具选型

演示场景

Excel 合并与标准化

输入

输出

承运商自动推断

PDF 条形码匹配与重排 ★ 核心算法

输入

输出

阶段 1：扫描所有 PDF 页，建注册表

阶段 2：双向尾部匹配算法

阶段 3：按 Excel 顺序重组 PDF

工具选型

为什么不直接按页码匹配？

添加水印

生成拣货表

为什么这样设计？

模版驱动输出格式

全程 dtype=str 读 Excel

零容错匹配，立即报错

Fail Fast 早期校验

水印原地覆盖

pymupdf 替代 pdftoppm

从原始文件到标准输出

可复用到其他 Agent 项目的原则

明确输入输出契约

Fail Fast

数据来源权威性

模版驱动

选对工具

可观测性

零容错原则

整洁输出

把你团队的重复工作交给 Agent 来做

2人×4小时的发货工作
Agent 3分钟内完成

把你团队的重复工作
交给 Agent 来做