多站点商品数据合并：统一名称、规格与来源记录
用 Python 把两个来源的商品记录整理成统一报价数据，区分字段映射、单位换算、商品匹配与待确认记录，避免同名商品误合并。
两个网站都卖“经典红茶”，不代表它们卖的是同一个规格。一个页面写 500 g，另一个写两盒各 250 g；即使净含量一样，包装、库存和价格也未必能够直接比较。
本文面向已经拥有多个采集器结果的数据开发者。我们接收两份结构不同的 JSON，把名称、规格和报价整理成统一格式，再用明确的商品标识合并来源。最终输出三类数据：能够确认是同一商品的报价组、等待人工确认的记录、无法进入合并流程的坏记录。
这里不重新抓网页，也不重复 URL 去重。输入是采集器已经取得的字段；重点是把这些字段变成不容易误用的数据。
先把商品身份与一次报价分开
同一个商品可以在两个网站、两个地区或两个时间点有不同报价。不要为了只保留一行而覆盖来源价格。
本文采用两层记录：商品层保留标识、品牌和包装规格；报价层保留来源、原名称、原价格、币种和观测时间。两种货币不直接比较，不把运费或税费未知的价格宣布为“最低总价”。
自动匹配需要明确证据。本例要求来源提供相同 GTIN，品牌与包装数量也一致。缺少 GTIN 的相似名称只进入待确认列表。这里把 GTIN 当作两个采集器已提供并需要进一步核查的商品标识，没有实现条码合法性或供应商真实性验证。
安装与准备输入
使用 Python 3.12 或更新版本，只需要标准库。创建 merge_products.py。下面的完整示例把两份输入放在正文中，方便看清映射；实际接入时再换成 JSON 文件读取。
来源 A 使用 name/grams/price，来源 B 使用 label/weight/amount。B 的 0.5 kg 应映射成 500 g；没有可靠编号的两条“茶叶礼盒”不能直接合并。
import json
import unicodedata
from collections import defaultdict
from decimal import Decimal, InvalidOperation
feeds = {
"A": [
{"id": "a1", "name": "山谷红茶 500g", "brand": "Valley",
"gtin": "0123456789012", "grams": "500", "price": "12.90",
"currency": "CNY", "url": "https://example.16yun.cn/a/a1"},
{"id": "a2", "name": "茶叶礼盒", "brand": "Valley",
"gtin": None, "grams": "500", "price": "18.00",
"currency": "CNY", "url": "https://example.16yun.cn/a/a2"},
],
"B": [
{"id": "b1", "label": "Valley black tea", "brand": "Valley",
"gtin": "0123456789012", "weight": "0.5", "unit": "kg",
"amount": "11.90", "currency": "CNY",
"url": "https://example.16yun.cn/b/b1"},
{"id": "b2", "label": "茶叶礼盒", "brand": "Valley",
"gtin": None, "weight": "1", "unit": "kg",
"amount": "25.00", "currency": "CNY",
"url": "https://example.16yun.cn/b/b2"},
],
}
observed_at = "2026-10-28T09:00:00Z"
def normalized_key(value):
return " ".join(unicodedata.normalize("NFKC", value).casefold().split())
def normalize(source, raw):
if source == "A":
title, grams, price = raw["name"], raw["grams"], raw["price"]
grams = Decimal(grams)
else:
title, price = raw["label"], raw["amount"]
multiplier = {"g": Decimal(1), "kg": Decimal(1000)}[raw["unit"]]
grams = Decimal(raw["weight"]) * multiplier
amount = Decimal(price)
if not grams.is_finite() or grams <= 0:
raise ValueError("包装重量必须是正的有限值")
if not amount.is_finite() or amount < 0:
raise ValueError("价格必须是非负有限值")
brand = raw["brand"].strip()
currency = raw["currency"]
if not title.strip() or not brand or currency not in {"CNY", "USD", "GBP"}:
raise ValueError("名称、品牌或币种不符合约定")
return {
"source": source, "source_id": raw["id"],
"title_raw": title, "brand_key": normalized_key(brand),
"gtin": raw.get("gtin"), "grams": str(grams.normalize()),
"price": str(amount), "currency": currency,
"source_url": raw["url"], "observed_at": observed_at,
}
groups = defaultdict(list)
review = []
rejected = []
for source, items in feeds.items():
for raw in items:
try:
item = normalize(source, raw)
except (KeyError, ValueError, InvalidOperation) as error:
rejected.append({"source": source, "raw": raw, "reason": str(error)})
continue
if not item["gtin"]:
review.append({"reason": "missing_identifier", "offer": item})
continue
groups[item["gtin"]].append(item)
matched = []
for gtin, offers in groups.items():
signatures = {(row["brand_key"], row["grams"]) for row in offers}
if len(signatures) != 1:
review.append({"reason": "identifier_conflict", "gtin": gtin, "offers": offers})
else:
matched.append({"gtin": gtin, "offers": offers})
result = {"matched": matched, "review": review, "rejected": rejected}
with open("merged.json", "w", encoding="utf-8") as stream:
json.dump(result, stream, ensure_ascii=False, indent=2)
print({"matched_groups": len(matched), "review_items": len(review),
"rejected_items": len(rejected)})运行：
python merge_products.py预期输出是一个确认组、两个待确认记录、零个坏记录。它是根据正文输入推导的结果，不代表实际商品目录的匹配准确率。
映射与标准化分别在修正什么
normalize 把来源字段映射成内部字段，但原名称仍保留为 title_raw。不要用清洗后的短名称覆盖原始名称，否则以后无法判断某个规格是不是被清洗规则删掉。
重量使用 Decimal 从字符串转换。0.5 kg 与 500 g 会得到相同的标准重量；如果先用二进制浮点乘除，再把差异当成规格变化，容易产生不必要的冲突。金额同样不经过浮点；相关取舍见 Decimal 文档。
normalized_key 使用 NFKC、大小写折叠和空白归并，只用于品牌匹配键。Unicode 兼容规范化可能改变展示形式，因此它不能直接用来改写合同名称、型号或法律标识。规范化类型见 Unicode 文档。
这里没有把 2 × 250 g 当成 500 g 单袋。接入这种页面时，应增加 pack_count、unit_weight 和包装类型，而不是只保存总重量。业务上把两种包装视作可替代品，也应该是单独的分析关系，不是悄悄改成同一个商品身份。
为什么相同编号还要检查规格
来源可能取错字段，也可能把商品级编号重复赋给多个规格。代码先按 GTIN 形成候选组，再检查品牌和重量；同编号却不同规格时整组进入 identifier_conflict，保留全部报价，等待查明。
它没有依据“多数来源都写 500 g”就覆盖少数记录。多数不自动代表正确。回到来源页面核对编号属于商品还是规格，确定哪条记录错了，再修复对应采集器或数据映射。
相反，两个没有编号的礼盒虽然名称相同，重量却不同，更不能用标题直接匹配。后续可以加入品牌、制造商型号、规格和图片等证据生成候选，但候选相似度不应直接变成确认匹配。
怎样使用合并结果比较报价
merged.json 中同一个确认组保留 A 和 B 的各自报价。对本例数据，两个报价都是 CNY，并且单位规格一致，因此可以比较页面展示价格 12.90 与 11.90；不能进一步声称 B 含运费更便宜，输入并没有提供运费规则。
数据还应保留观测时间。本例给定固定时间来说明字段结构，真实采集器应记录实际观测时间，不能把合并程序启动时间冒充网页采集时间。相隔几天的两条价格，不能当作同一时刻的市场对照。
如果需要在多个批次维护当前报价，建议用“来源 + 来源记录 ID + 地区/规格”标识报价，再保存每次观测版本。GTIN 负责跨来源关联，不负责覆盖同一来源的所有历史。
常见问题与修复
本来同一商品却没有合并。 先看缺少编号还是编号冲突。若一个来源缺编号，补采详情里的制造商标识；不要先把匹配条件放宽成相同标题。修复后应看到两条来源记录进入同一个确认组，原始字段仍在。
规格差一千倍。 检查来源数字的单位是不是 kg、g 或 mg。修改映射表并重新处理原始记录；不要直接把离群值截断到平均值。未知单位应被拒绝，而不是默认按 g 解释。
同编号不同品牌。 先回查页面和编号提取位置。可能取到了平台内部分类码，也可能来源使用的不是 GTIN。确认原因后更改编号类型或来源映射，不能单靠品牌别名字典抹平冲突。
金额转换失败。 原始值可能包含货币符号、千位分隔符或“起”。当前输入合同要求已经分离好的十进制金额。应在来源适配层明确处理格式和价格含义，不对所有字符一删了之。
导出后找不到某个来源。 检查 review 和 rejected，不要只统计 matched。合并完成不意味着全部记录自动获得商品身份。只有待确认记录经过核对，才应该进入下游自动比较。
跨站合并最重要的产物，是保留证据的商品关系和报价记录。无法确认的关系保持待确认，比给每条记录强行分组更便于后续修正。
需要企业代理方案？
我们可根据目标站点、并发规模与稳定性目标提供定制方案。