适用读者：负责电商站点后端或数据平台的同学，商品页由多套模板渲染、SKU 量级在数千到数十万、需要持续给搜索引擎与生成式引擎输出结构化数据（Structured Data）的团队。示例环境为 Python 3.12 与 MySQL 8.0，思路可平移到其它语言栈。

一次改字段的经历：十几个模板与一个漏掉的字段

去年大促前一周，运营要求把商品页的价格有效期字段 priceValidUntil 从写死的季度末，改成随 SKU 的活动结束时间输出。我在仓库里搜了一遍，命中 14 个文件：PC 详情、移动详情、秒杀、预售、拼团，还有五套只在活动期上线的专题模板。改完上线。两周后富媒体报告里冒出 327 条缺失字段告警，顺着链接点进去，全是专题模板——那一份把字段拼进了另一个变量名，模板渲染不报错，只是输出了空串。

问题不在这一次改动，在于没有任何一处代码知道"商品页该输出哪些字段"这件事。字段散落在模板里，模板散落在仓库里，谁改谁知道。

管线总览：四层结构与一次完整流转

把这件事重写成一条管线后，结构变成四层：事实源、生成器、校验层、注入层。事实源是 MySQL 商品库；生成器读库、渲染、校验、落盘；注入层决定产物怎么出现在 HTML 里。

flowchart TD
    A[MySQL 商品库 pms_sku / pms_brand / pms_price] -->|按 updated_at 水位线增量拉取| B[Python 生成器]
    B --> C[Jinja2 模板渲染 JSON-LD 片段]
    C --> D[pydantic 模型校验与 jsonschema 复核]
    D -->|校验通过| E[产物目录 objects/schemas/版本号/]
    D -->|校验失败| F[拒绝清单 deadletter.jsonl 并告警]
    E --> G{注入方式}
    G -->|方案一| H[CDN 边缘按 URL 规则注入 script 标签]
    G -->|方案二| I[页面渲染时从产物缓存内联]
    H --> J[线上商品页]
    I --> J
    F --> K[人工修数后重跑增量]
    K --> B

四层之间的边界是硬边界。模板层不允许再写任何字段字面量，商品页只认注入层给的字符串。任何一次字段口径变更，落点只有一个：生成器里的映射表。

迁移前后的实测对照

下表取自迁移完成后连续四周的观测，抽样口径为全站 8.6 万个在售 SKU 的商品页。

完整率的提升几乎全部来自校验层：字段缺失不再静默通过，而是在产物落盘前被拦下。

事实源：MySQL 表设计与增量水位线

商品库里真正参与结构化数据输出的字段，建议单独圈出来，不要和交易字段混在同一张宽表里改。下面这张片段是可运行的最小集。

-- 运行环境：MySQL 8.0.36，InnoDB，utf8mb4_0900_ai_ci
-- 商品主表：结构化数据的取值来源，模板层不再持有任何字段默认值
CREATE TABLE pms_sku (
  sku_id        BIGINT UNSIGNED NOT NULL COMMENT 'SKU 内部 ID，与商品页 URL 一一对应',
  spu_id        BIGINT UNSIGNED NOT NULL COMMENT 'SPU 聚合并非必须，但便于复用主图与描述',
  title         VARCHAR(255)    NOT NULL COMMENT '商品标题，对应 schema.org 的 name',
  brand_id      BIGINT UNSIGNED NULL     COMMENT '品牌 ID，为空时生成器回落到店铺名',
  gtin13        CHAR(13)        NULL     COMMENT '国际商品条码，Offer 的可选但强烈建议字段',
  mpn           VARCHAR(64)     NULL     COMMENT '厂商零件号，3C 类目常用',
  main_image    VARCHAR(512)    NULL     COMMENT '主图完整 URL，要求 CDN 可直连',
  description   MEDIUMTEXT      NULL     COMMENT '纯文本描述，生成器会剥离 HTML 标签',
  status        TINYINT         NOT NULL DEFAULT 1 COMMENT '1 在售 2 下架 3 预告',
  updated_at    DATETIME(3)     NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
  PRIMARY KEY (sku_id),
  -- 水位线查询的核心索引：按更新时间游标扫描，避免全表排序
  KEY idx_sku_updated (updated_at, sku_id),
  -- 状态过滤走前缀索引，增量扫描时先收敛状态再回表
  KEY idx_sku_status (status, updated_at)
) ENGINE=InnoDB COMMENT='商品主表';

-- 价格表：价格与库存变动频率高于主表，单独成表以免污染主表水位线
CREATE TABLE pms_price (
  sku_id          BIGINT UNSIGNED NOT NULL,
  sale_price      DECIMAL(12,2)   NOT NULL COMMENT '含税售价，Offer 的 price',
  currency        CHAR(3)         NOT NULL DEFAULT 'CNY' COMMENT 'ISO 4217 货币码',
  price_valid_to  DATE            NULL     COMMENT '价格有效期终点，对应 priceValidUntil',
  stock_qty       INT             NOT NULL DEFAULT 0 COMMENT '可用库存，决定 availability 取值',
  updated_at      DATETIME(3)     NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
  PRIMARY KEY (sku_id),
  KEY idx_price_updated (updated_at)
) ENGINE=InnoDB COMMENT='价格与库存表';

-- 生成器水位线表：记录每个产线跑到的位置，支持按产线独立回放
CREATE TABLE gen_watermark (
  pipeline   VARCHAR(64)     NOT NULL COMMENT '产线名，如 jsonld_product',
  wm_value   DATETIME(3)     NOT NULL COMMENT '水位线时间戳',
  run_rows   INT             NOT NULL DEFAULT 0 COMMENT '本轮处理行数，用于异常比对',
  updated_at DATETIME(3)     NOT NULL DEFAULT CURRENT_TIMESTAMP(3) ON UPDATE CURRENT_TIMESTAMP(3),
  PRIMARY KEY (pipeline)
) ENGINE=InnoDB COMMENT='生成水位线表';

字段设计上有三个容易被忽略的点。价格单独成表，否则一次调价会把主表 updated_at 推高，导致每轮增量都扫出大批实际上没变的 SKU。水位线单独建表而不是写在配置文件里，因为它必须是可查询、可回退、可并发安全更新的状态。updated_at 用毫秒精度，一秒内多次改价的场景在批量改价时很常见，秒级精度会丢数据。

生成器：Jinja2 渲染与双段校验

生成器分三段跑：增量拉取、渲染、校验。三段之间用内存里的列表传递，不落中间文件，失败整体回滚本轮。

# 运行环境：Python 3.12.6 / Jinja2 3.1.4 / pymysql 1.1.1 / pydantic 2.9.2 / jsonschema 4.23.0
# 安装命令：pip install "jinja2==3.1.*" "pymysql==1.1.*" "pydantic==2.9.*" "jsonschema==4.23.*"
# 说明：以下为生成器主干，省略连接池与日志初始化，可直接嵌入调度任务
from __future__ import annotations

# 标准库：os 读取环境变量，datetime 与 timedelta 处理价格有效期缺省值
import os
from datetime import date, datetime, timedelta

# 第三方库：pymysql 直连 MySQL，Jinja2 负责渲染，pydantic 负责结构校验
import pymysql
from jinja2 import Environment, FileSystemLoader
from pydantic import ValidationError

# 常量区：站点域名、单次批大小、价格缺省有效期天数
# 这些值建议从环境变量注入，便于多环境复用同一份生成器代码
SITE_HOST = os.getenv("SITE_HOST", "example.com")
BATCH_SIZE = int(os.getenv("BATCH_SIZE", "5000"))
DEFAULT_VALID_DAYS = 30

# Jinja2 环境：JSON-LD 不是 HTML，必须关闭 HTML 自动转义
# 否则 & 引号会被转成实体，导致 JSON 解析失败
env = Environment(
    loader=FileSystemLoader("templates"),
    autoescape=False,
    trim_blocks=True,
    lstrip_blocks=True,
)
# 模板文件只包含取值与条件拼接，业务判断一律留在 Python 侧
TEMPLATE = env.get_template("product.jsonld.j2")

# 可用性映射：库存与上架状态共同决定 schema.org 的受控枚举值
# 这里集中维护，模板与各调用方都不再各自写字符串字面量
AVAILABILITY = {
    "in_stock": "https://schema.org/InStock",
    "out_of_stock": "https://schema.org/OutOfStock",
    "preorder": "https://schema.org/PreOrder",
}


def fetch_delta(conn, watermark: datetime, batch: int = BATCH_SIZE):
    # 增量拉取：水位线为左开区间，避免同一毫秒的数据被重复处理
    # 价格表用 LEFT JOIN 一次性带出，省掉 N+1 查询
    # row_updated 取两表时间戳的较大值，作为本轮水位线推进依据
    sql = """
        SELECT s.sku_id, s.title, s.gtin13, s.mpn, s.main_image, s.description,
               s.status, p.sale_price, p.currency, p.price_valid_to, p.stock_qty,
               GREATEST(s.updated_at, COALESCE(p.updated_at, s.updated_at)) AS row_updated
        FROM pms_sku s
        LEFT JOIN pms_price p ON p.sku_id = s.sku_id
        WHERE s.updated_at > %s OR p.updated_at > %s
        ORDER BY row_updated
        LIMIT %s
    """
    with conn.cursor(pymysql.cursors.SSDictCursor) as cur:
        # SSDictCursor 为流式游标，8 万行数据不会把内存打满
        cur.execute(sql, (watermark, watermark, batch))
        for row in cur:
            yield row


def pick_availability(status: int, stock: int) -> str:
    # 预告状态优先于库存判断，避免未上架商品被标成可售
    if status == 3:
        return AVAILABILITY["preorder"]
    if status != 1 or stock <= 0:
        return AVAILABILITY["out_of_stock"]
    return AVAILABILITY["in_stock"]


def strip_html(raw: str | None) -> str | None:
    # 描述字段常带运营后台粘贴的 HTML 标签，需剥离成纯文本
    # 结构化数据里出现标签会被判为无效值
    if not raw:
        return None
    # 极简实现：去掉尖括号包裹的内容，生产环境建议换成专业清洗库
    import re as _re

    text = _re.sub(r"<[^>]+>", " ", raw)
    # 压缩连续空白，避免描述字段里出现大段换行
    return _re.sub(r"\s+", " ", text).strip()


def render_one(row: dict, site_host: str) -> str:
    # 渲染前先做字段归一化，模板里只做取值不做判断
    # 归一化是整条管线里仅有的、允许出现字段默认值的位置
    ctx = {
        "sku_id": row["sku_id"],
        "name": (row["title"] or "").strip(),
        "gtin13": row["gtin13"],
        "mpn": row["mpn"],
        # 主图允许为空，但校验层要求至少一个元素，故此处原样透传
        "image": row["main_image"],
        # 描述先剥离 HTML 标签再入模板
        "description": strip_html(row["description"]),
        "price": f"{row['sale_price']:.2f}" if row["sale_price"] is not None else None,
        "currency": row["currency"] or "CNY",
        # 有效期缺省时给 30 天，宁可保守也不要输出空串
        "price_valid_until": (row["price_valid_to"] or (date.today() + timedelta(days=30))).isoformat(),
        "availability": pick_availability(row["status"], row["stock_qty"] or 0),
        "url": f"https://{site_host}/p/{row['sku_id']}.html",
    }
    return TEMPLATE.render(**ctx)

模板本身保持极简，只做取值与条件拼接，判断逻辑全部留在 Python 侧。这样模板可以被运营侧同学读懂，也不会变成第二个业务逻辑藏身处。

校验层：pydantic 拦结构，jsonschema 拦词表

pydantic 负责类型与必填，jsonschema 负责枚举与格式。两者职责不同，合并成一层会让错误信息变得难读。

# 运行环境：Python 3.12.6 / pydantic 2.9.2 / jsonschema 4.23.0
# 说明：ProductModel 对应 schema.org/Product 的最小可用集，字段缺失即落 deadletter
# 两段校验的分工：pydantic 管类型与必填，jsonschema 管枚举与常量
from typing import Optional

# Dict202012Validator 预编译词表，实例化一次即可复用
from jsonschema import Draft202012Validator

# BaseModel 与 Field 描述结构，field_validator 挂载归一化钩子
from pydantic import BaseModel, Field, field_validator


class OfferModel(BaseModel):
    # price 用字符串承载，避免浮点精度在 JSON 序列化时抖动
    # 正则限制两位小数，杜绝 0.1 加 0.2 这类二进制误差产物
    price: str = Field(pattern=r"^\d+(\.\d{1,2})?$")
    # 货币码固定三位，长度约束即 ISO 4217 约束
    priceCurrency: str = Field(min_length=3, max_length=3)
    # 有效期允许为空，由上层按缺省天数补齐
    priceValidUntil: Optional[str] = None
    # 可用性取值来自受控词表，不允许自由文本
    availability: str
    # url 必须是可抓取的完整地址，相对路径在此层拦不住
    url: str

    @field_validator("priceCurrency")
    @classmethod
    def upper_currency(cls, v: str) -> str:
        # ISO 4217 要求大写，库里偶有小写脏数据
        # 归一化放在校验器里，避免污染上游渲染上下文
        return v.upper()


class ProductModel(BaseModel):
    # @context 与 @type 固定，不允许模板改写
    context: str = Field(alias="@context", default="https://schema.org")
    type: str = Field(alias="@type", default="Product")
    # 商品名至少两个字符，单字符多为脏数据
    name: str = Field(min_length=2, max_length=255)
    # image 是数组且至少一个元素，缺主图直接判失败
    image: list[str] = Field(min_length=1)
    # 描述可选，但如果存在则不应为空串
    description: Optional[str] = None
    # sku 落内部 ID，不用商品标题做键，避免同名规格互相覆盖
    sku: str
    # 条码为 13 位数字，位数不对说明上游同步有问题
    gtin13: Optional[str] = Field(default=None, pattern=r"^\d{13}$")
    # 厂商零件号，3C 类目常用，无格式约束
    mpn: Optional[str] = None
    # 品牌缺失时允许降级，由生成器回落店铺名
    brand: Optional[dict] = None
    # offers 为嵌套模型，价格与库存问题在这里暴露
    offers: OfferModel


# 词表级校验：枚举值必须落在 schema.org 定义的受控词表内
# 这一层补的是 pydantic 不擅长表达的常量约束
JSONLD_SCHEMA = {
    "$schema": "https://json-schema.org/draft/2020-12/schema",
    # 顶层必填：上下文、类型、名称、主图、报价，缺一即不产出
    "type": "object",
    "required": ["@context", "@type", "name", "image", "offers"],
    "properties": {
        # 常量约束：类型只能是 Product，防止模板被误改成其它类型
        "@type": {"const": "Product"},
        # offers 子树：价格对象单独约束，枚举封闭
        "offers": {
            "type": "object",
            "properties": {
                # 可用性枚举封闭，拼错变量名在这里会被抓到
                "availability": {
                    "enum": [
                        "https://schema.org/InStock",
                        "https://schema.org/OutOfStock",
                        "https://schema.org/PreOrder",
                        "https://schema.org/BackOrder",
                    ]
                }
            },
        },
    },
}
VALIDATOR = Draft202012Validator(JSONLD_SCHEMA)


def validate_payload(payload: dict, sku_id: int) -> list[str]:
    # 先过 pydantic，结构与必填问题在这里一次性暴露
    # by_alias=True 是因为模型字段用 @context 这类带符号的名字
    try:
        ProductModel.model_validate(payload, by_alias=True)
    except ValidationError as exc:
        # 错误按字段位置聚合返回，便于 deadletter 里按类目统计
        return [f"{sku_id} pydantic: {e['loc']} {e['msg']}" for e in exc.errors()]
    # 再过 jsonschema，兜住枚举与常量这类 pydantic 不擅长描述的约束
    return [f"{sku_id} jsonschema: {e.message}" for e in VALIDATOR.iter_errors(payload)]

校验不通过的 SKU 不写产物，而是追加到 deadletter.jsonl 并按类目聚合告警。这件事的价值在于把"线上静默缺失"变成"离线显式报错"，修复成本从排查一整条链路降到改一条数据。

机制剖析：水位线为什么既不漏也不重

增量生成最容易出问题的地方是边界。水位线机制用三个约束把它钉死。

第一，区间左开。查询条件是 updated_at > watermark，上一轮最后一条数据不会被重复捞取，重复捞取本身无害，但会掩盖"同一条数据被渲染两次且结果不同"这类真实问题。

第二，水位线取自本轮最大 row_updated，而不是本轮结束时间。若取结束时间，在本轮跑动期间发生但时间戳小于结束时间的修改会被永久跳过——这是最隐蔽的一种数据丢失。

第三，主表与价格表的时间戳取 GREATEST 合并后再推进。两侧各自维护水位线会让组合结果出现空洞。

sequenceDiagram
    participant W as 水位线表
    participant G as 生成器
    participant D as MySQL 商品库
    participant O as 产物目录
    participant L as deadletter
    G->>W: 读取 wm_value
    W-->>G: 2026-09-18 10:00:00.000
    G->>D: WHERE updated_at > 水位线 ORDER BY row_updated LIMIT 5000
    D-->>G: 返回 5000 行变更集
    loop 逐行处理
        G->>G: Jinja2 渲染
        G->>G: pydantic 加 jsonschema 双段校验
        alt 校验通过
            G->>O: 写入 sku_id.jsonld
        else 校验失败
            G->>L: 追加错误明细
        end
    end
    G->>W: 以本轮最大 row_updated 推进水位线
    Note over G,W: 未推进到系统时间，避免漏掉跑动期间的写入
    G->>G: 若本轮满 5000 行则继续下一批，否则结束

水位线还能直接支撑回滚。把 wm_value 改回上一个已知良好的时间点，重跑增量，产物目录就回到那个时刻的状态。比从备份恢复快两个数量级。

机制剖析：单一事实源如何消除模板间漂移

模板间不一致的本质，是同一个业务事实有多个可写副本。手写时期，priceValidUntil 在 14 个模板里有 14 份定义，其中 13 份正确，1 份拼错变量名。任何依赖于"所有副本同时被改对"的方案，在副本数增长时失败概率趋于 1。

换成单一事实源之后，副本数降到 1，但代价是新增了一层间接：商品页不再直接知道自己要输出什么，而是问注入层要。这层间接带来两个额外收益。校验可以集中做一次，不必在每个模板里重复防御。字段口径变更的审计也集中了，谁在什么时候把 availability 的库存阈值从 0 改成 1，映射表的提交记录里写得清清楚楚。

代价同样要明确：生成器成了新的单点。商品库不可用时，商品页的结构化数据会停留在上一次产物上。因此注入层必须容忍产物缺失——拿不到新产物就输出旧产物，两者都没有就不输出 script 标签，绝不能因为拿不到数据而让页面 500。

注入方式：CDN 边缘注入与渲染时内联

两种注入方式没有普适的优劣，取决于模板栈的收敛程度。

多数存量站点的现实是模板栈短期内收敛不了，边缘注入更稳。新站点或者已经统一到一套渲染框架的系统，直接内联更简单，少一整套基础设施。

产物发布采用版本目录加原子切换，避免边写边读。

# 运行环境：Linux / GNU coreutils / bash 5.x / Python 3.12 生成器已完成一轮产出
# 约定：产物目录按版本号分目录，current 为符号链接，切换是原子操作
# 前置条件：生成器已退出且退出码为 0，否则不进入发布流程

# 生成一轮产物，版本号取当前时间戳，避免并发跑批互相覆盖
VERSION=$(date +%Y%m%d%H%M%S)
python -m gen.jsonld --pipeline jsonld_product --out /data/objects/schemas/$VERSION

# 校验产物规模，行数骤降通常意味着上游查询条件写错
COUNT=$(ls /data/objects/schemas/$VERSION | wc -l)
if [ "$COUNT" -lt 80000 ]; then
  echo "产物数量异常：$COUNT" >&2
  exit 1
fi

# 原子切换：ln -sTf 先建新链接再 rename，读进程不会看到半截目录
ln -sTf /data/objects/schemas/$VERSION /data/objects/schemas/.current.tmp
mv -Tf /data/objects/schemas/.current.tmp /data/objects/schemas/current

# 保留最近 5 个版本，回滚时直接改链接指向即可
ls -1dt /data/objects/schemas/20* | tail -n +6 | xargs -r rm -rf

灰度与回滚

灰度按 SKU 类目分批推进，而不是按流量百分比。按流量分会出现同一 SKU 在两个请求里拿到不同产物，排查时无法复现。

回滚设计留了三层。最轻的一层是配置开关，按类目关注入，秒级生效，产物本身不动。中间一层是版本目录回指，把 current 指回上一个版本目录。最重的一层是水位线回退加增量重跑，用于产物内容本身有误的场景，代价是几十分钟的重算时间。三层覆盖了从"注入出问题"到"数据出问题"的全区间。

三个容易被忽略的坑

第一个是价格与产物的时序差。CDN 边缘注入的产物是分钟级的，秒杀场景下页面已经显示新价格而结构化数据还是旧价格。这类不一致会被判为误导性内容，解决方式是对动销极快的类目直接用渲染时内联，两条链路并存，按类目选路。

第二个是主图字段填相对路径。schema.org 的 image 要求可抓取的完整 URL，相对路径在本地预览时看不出问题，上线后校验全过但抓取失败。这个坑在 pydantic 层拦不住，得加一条域名前缀校验。

第三个是去重逻辑误伤。同名不同规格的商品若共用一套 sku 取值，会被搜索引擎判定为重复实体。SKU 维度必须落内部 ID，不要用商品标题做键。

参考与延伸

• schema.org Product 类型定义：https://schema.org/Product
• Google 结构化数据产品片段文档：https://developers.google.com/search/docs/appearance/structured-data/product
• Jinja2 官方文档：https://jinja.palletsprojects.com
• pydantic 官方文档：https://docs.pydantic.dev

字段映射表里还有一批类目特有属性，比如服饰的尺码体系、生鲜的重量单位，这些字段的口径往往要和外部词表对齐。你们的类目里哪些字段最难维护，评论区聊聊。

JSON-LD 结构化数据, Python 生成器, Jinja2 模板, pydantic 校验, MySQL 增量水位线, CDN 边缘注入, GEO, AI优化AIO