学完能做到 外部系统每 5 分钟同步一次订单，不重复、不漏单、失败能查。
开始之前
- 已完成 API 快速开始（能登录并调通 GET）
- 外部订单有稳定的唯一编号
操作步骤
先确认三张对照关系
同步前必须解决三件事：① 客户怎么对应（外部客户 ID → 本系统客户 ID）；② 型号怎么对应（外部 SKU → 本系统型号 ID）；③ 订单号怎么对应（外部单号 → 本系统订单号）。三张对照表建议存在你自己的中间表里，不要每次全量查本系统做模糊匹配。
型号对不上就不要自动建
遇到本系统没有的型号，正确做法是标记为「待人工映射」并告警，而不是自动创建型号 —— 自动建出来的型号没有工艺路线和工价，订单确认后会拆不出任务，问题被推到生产环节才暴露。
幂等靠外部单号，不靠时间戳
用外部订单号作为幂等键：先查本系统有没有这个外部单号（建议在订单备注或扩展字段里存），有就跳过。用「最近 N 分钟的时间窗」做去重迟早会出错 —— 补同步历史单时会重复，跨零点会漏单。
下单接口只创建，不直接确认
POST /api/admin/production/orders 创建订单后处于草稿/待确认状态。确认动作会拆分出工单与任务，这一步建议保留人工确认（或按规则自动确认）。理由：型号工艺路线配错时，确认后要作废重来，代价比确认前发现大得多。
失败要有队列和重试
同步任务失败时不要丢掉这条数据。建议本地落一张同步表，记录 {外部单号, 状态, 重试次数, 最后错误}，失败的定时重试。连续失败 3 次的进人工队列。没有这张表，出问题时你无法回答「昨天那 12 单到底同步了没有」。
用「拉」不用「推」
外部系统主动推给你，你要处理签名、重放、乱序。自己做定时拉取，节奏可控、可补数据、失败可重跑。除非对方明确要求推模式，否则一律拉。
代码示例
python带幂等的同步骨架
import sqlite3, requests, time
from datetime import datetime
BASE = "https://YOUR_HOST"
LIMIT_PER_RUN = 200
# 本地同步表：外部单号唯一索引，这就是幂等的根
DB = sqlite3.connect("sync.db")
DB.execute("""CREATE TABLE IF NOT EXISTS synced_order(
ext_no TEXT PRIMARY KEY, -- 外部唯一单号
inner_id INTEGER, -- 本系统订单 ID
status TEXT, -- pending/created/failed
retry INTEGER DEFAULT 0,
last_error TEXT,
updated_at TEXT
)""")
DB.commit()
def api_post(token, path, payload):
r = requests.post(f"{BASE}{path}", json=payload,
headers={"Authorization": f"Bearer {token}"}, timeout=30)
if r.status_code == 429:
time.sleep(3)
return api_post(token, path, payload)
body = r.json()
if body.get("code") != 200:
raise RuntimeError(f"{body.get('code')}: {body.get('msg')}")
return body["data"]
def sync_one(token, ext_order, customer_map, sku_map):
ext_no = ext_order["order_no"]
# 1) 幂等检查（唯一索引兜底，并发安全）
row = DB.execute("SELECT status FROM synced_order WHERE ext_no=?", (ext_no,)).fetchone()
if row and row[0] == "created":
return "skip"
# 2) 对照关系必须都有，缺一个就挂起
cid = customer_map.get(ext_order["customer_code"])
if not cid:
raise RuntimeError(f"客户未映射: {ext_order['customer_code']}")
items = []
for it in ext_order["items"]:
sid = sku_map.get(it["sku"])
if not sid:
raise RuntimeError(f"型号未映射: {it['sku']}")
items.append({"sku_id": sid, "quantity": it["qty"]})
# 3) 创建订单（不自动确认）
data = api_post(token, "/api/admin/production/orders", {
"customer_id": cid,
"remark": f"ext:{ext_no}", # 外部单号留痕，便于反查
"items": items,
})
DB.execute("""INSERT INTO synced_order(ext_no, inner_id, status, updated_at)
VALUES(?,?,?,?)
ON CONFLICT(ext_no) DO UPDATE SET
inner_id=excluded.inner_id, status='created', updated_at=excluded.updated_at""",
(ext_no, data.get("id"), "created", datetime.now().isoformat()))
DB.commit()
return "created"示例中的域名、租户编码、密钥请替换成你自己的。生产环境不要把密钥写进前端代码。
容易踩的坑
同一订单被同步了两次，拆出两套工单？
幂等键没生效。检查是否用外部单号判重，以及判重是在创建之前还是之后。并发场景还要加锁或唯一索引，不能只靠「先查再插」。
订单里的客户名对不上，建了一堆重复客户？
客户必须走对照表，禁止按名称模糊匹配自动创建。第一次上线时人工把主要客户映射一遍，之后新客户进待映射队列。
同步很慢，1000 单要跑十几分钟？
逐条 POST 加逐条查询必然慢。优化顺序：① 对照表全量缓存到内存（只查一次）；② 判重改成批量查（一次查 200 个单号）；③ 请求并发控制在 5–10，再高会被限流反而更慢。
订单数量改了怎么办？
已确认的订单不能直接改数量。正确做法是作废未开工任务、按新数量重新生成。外部系统改单要能识别出「这是改单不是新单」，所以幂等键之外还需要一个版本或更新时间字段。