Shopify Bulk 商品导出验收：按作业 ID 核对 JSONL 父子关系

应用已经提交 Shopify 商品 Bulk 查询并取得作业 ID，接下来要验收的是这一个作业的结果。本文从指定 ID 的状态回读开始，检查下载入口、Product/Variant 的 JSONL 关系以及两种计数。输入是保存的作业 ID、原查询和结果文件；输出是可追溯的接受或拒绝原因。

本文按 2026 年 10 月 4 日核对的 2026-10 文档编写，限定在现有已授权应用的 backend 和原作业上下文中执行。Bulk 适用于 GraphQL Admin API；Storefront API 不支持，Admin UI extensions 的 Direct API access 也不支持，应交给应用 backend。商品读取需具备相应 read_products 授权。参见Bulk 支持范围、Product 权限要求和实际访问范围说明。

先保存作业身份与这次导出的查询

保存提交响应中的 bulkOperation.id、提交的查询、API 版本及应用与商店上下文；同时检查提交响应中的 userErrors。bulkOperationRunQuery 返回字段区分作业对象与提交错误。这里假定作业已成功取得 ID，不另外提供开户或凭证配置流程。

下面限定一种导出契约：根对象为 Product，每个商品下只导出 Variant 身份字段。它适用于提交时就采用这份查询的作业。若原作业查询不同，应按原查询建立验收规则，而不是把旧文件套进本例。

{
  products {
    edges {
      node {
        id
        __typename
        variants {
          edges { node { id __typename } }
        }
      }
    }
  }
}

__typename 用于区分实际对象类型，属于 GraphQL 元字段，见GraphQL 的类型元字段说明。__parentId 不写进上面的查询；它是 Bulk 展平嵌套连接时自动加入的关系字段，见JSONL 导出示例。

用返回的作业 ID 回读状态

使用 bulkOperation(id:) 回读保存的 ID，并核对响应中的 id、type 和 query。本篇只接受 type: QUERY 的导出作业；MUTATION 是另一种作业类型。该字段可能返回 null，不能把查不到作业当作空导出。字段及参数见bulkOperation 定义，类型值见BulkOperationType。下面的 9001 是演示 ID，实际操作必须替换为原作业返回值。

query ReadExport($id: ID!) {
  bulkOperation(id: $id) {
    id type query status errorCode
    url partialDataUrl rootObjectCount objectCount
  }
}

{
  "id": "gid://shopify/BulkOperation/9001"
}

现有请求层应检查 HTTP、JSON、顶层错误以及 X-Shopify-API-Version；版本响应头的含义见API 版本说明。下方验收器要求 query 字符串与保存值完全一致，这是消费者的保守规则；若只是格式不同，也会暂停并交由人工核对，不据此宣称 Shopify 返回无效查询。

区分成功下载、失败部分数据和过期结果

按BulkOperationStatus处理全部状态，再决定是否读取文件。

成功入口是 url，失败时的部分入口是 partialDataUrl；它们可能为 null。成功 URL 在作业完成后七天失效，部分 URL 在作业失败后七天失效。见BulkOperation 的 URL 与计数字段。不要把失败作业的部分文件晋级为完整导出。官方指南还说明，查询没有结果时，已完成作业的 url 可以为 null；本例在两个计数都为零时返回 EMPTY_SCOPE_REVIEW，仍需核对查询范围后才确认空结果，见完成后的结果下载说明。

按 ID 核对 Product 与 Variant 的关系

JSONL 每行是一个 JSON 对象；嵌套的 Variant 用自动添加的 __parentId 指向父 Product。官方指南明确描述连接的顺序保持不变、嵌套连接出现在父对象之后。本文按 ID 建立关系，是消费者的核对方式，不能据此说 Shopify 文件会随机排序或子行先于父行。参见JSONL 展平与顺序说明。

groupObjects 默认是 false；将其启用会增加分组处理时间及超时风险。该参数定义不证明文件随机或出现子先父顺序，见groupObjects 定义。

对上面的特定查询，rootObjectCount 应对应 Product 根对象数，objectCount 包括 Product 和 Variant，不能把后者当成商品数。两者都是 UnsignedInt64，以十进制数字字符串表示，范围为 0 到 2⁶⁴−1，见两种计数的定义和UnsignedInt64 表示范围。JavaScript 普通 number 的安全整数上限为 2⁵³−1，不能假定它可无损表示上述全部计数，见ECMAScript 的 Number.MAX_SAFE_INTEGER 定义。

以下三行是人工合成的身份文件，包含两个 Product 和一个 Variant，因此演示计数为 rootObjectCount: "2"、objectCount: "3"，并非商家导出实测。

{"id": "gid://shopify/Product/1001", "__typename": "Product"}
{"id": "gid://shopify/ProductVariant/2001", "__typename": "ProductVariant", "__parentId": "gid://shopify/Product/1001"}
{"id": "gid://shopify/Product/1002", "__typename": "Product"}

完整读取文件后才判定关系与计数

下面的 Python 示例接受已经解码的作业对象和本地文本行迭代器。它不发请求、不下载文件；下载层应先确认文件来自这次成功作业，保存取得时间和文件哈希。函数使用 Python 整数处理计数，并拒绝非字符串或超出范围的值。

"""Small offline Product/Variant identity-file acceptance example. No HTTP."""
import json
import re

def unsigned64(value):
    if not isinstance(value, str) or not re.fullmatch(r"[0-9]+", value):
        raise ValueError("base-10 digit string required")
    number = int(value)
    if number > 2**64 - 1:
        raise ValueError("UnsignedInt64 out of range")
    return number

def accept_export(operation, expected_id, expected_query, lines):
    if not isinstance(operation, dict) or operation.get("id") != expected_id:
        return "JOB_ID_MISMATCH", None
    if operation.get("type") != "QUERY":
        return "JOB_TYPE_NOT_QUERY", None
    if operation.get("query") != expected_query:
        return "QUERY_MISMATCH", None
    status = operation.get("status")
    if status in {"CREATED", "RUNNING", "CANCELING"}:
        return "WAIT", None
    if status == "FAILED":
        return "FAILED_PARTIAL" if operation.get("partialDataUrl") else "FAILED_NO_DATA", None
    if status in {"CANCELED", "EXPIRED"}:
        return status, None
    if status != "COMPLETED":
        return "UNKNOWN_STATUS", None
    if operation.get("errorCode") is not None or operation.get("partialDataUrl") is not None:
        return "INCONSISTENT_COMPLETED_RECEIPT", None
    try:
        roots = unsigned64(operation.get("rootObjectCount"))
        objects = unsigned64(operation.get("objectCount"))
    except ValueError:
        return "BAD_COUNTS", None
    if operation.get("url") is None:
        return "EMPTY_SCOPE_REVIEW" if roots == objects == 0 else "MISSING_URL", None
    if not isinstance(operation["url"], str) or not operation["url"]:
        return "MISSING_URL", None
    products, parents, ids = set(), {}, set()
    for number, line in enumerate(lines, 1):
        if not isinstance(line, str) or not line.strip():
            return "BAD_JSONL_LINE", {"line": number}
        try:
            row = json.loads(line)
        except json.JSONDecodeError:
            return "BAD_JSONL_LINE", {"line": number}
        if not isinstance(row, dict) or not isinstance(row.get("id"), str) or not row["id"]:
            return "BAD_ROW_ID", {"line": number}
        if row["id"] in ids:
            return "DUPLICATE_ID", {"line": number}
        ids.add(row["id"])
        if row.get("__typename") == "Product":
            if "__parentId" in row:
                return "ROOT_HAS_PARENT", {"line": number}
            products.add(row["id"])
        elif row.get("__typename") == "ProductVariant":
            parent = row.get("__parentId")
            if not isinstance(parent, str) or not parent:
                return "MISSING_PARENT_ID", {"line": number}
            parents[row["id"]] = parent
        else:
            return "ROW_TYPE_OUTSIDE_QUERY", {"line": number}
    if any(parent not in products for parent in parents.values()):
        return "ORPHAN_VARIANT", None
    if len(products) != roots or len(ids) != objects:
        return "COUNT_MISMATCH", {"products": len(products), "objects": len(ids)}
    return "ACCEPT_ID_RELATIONS", {"product_ids": sorted(products),
                                    "variant_parent_ids": parents}

使用方式是将 payload["data"]["bulkOperation"]、保存的 ID、保存的查询及以 UTF-8 打开的文件传给 accept_export。接受路径会读到 EOF 后检查父对象、唯一 ID 和计数；遇到错误则提前拒绝并返回原因。成功值 ACCEPT_ID_RELATIONS 仅表示这个特定身份文件通过所列校验。

代码为小规模离线演示，虽然输入可以逐行读取，ID 集合和父子映射仍随对象数增长而占用内存。大文件应采用适当的存储与完整性校验实现；本文没有提供生产级大文件处理器，也没有验证查询范围、业务字段或持续变化数据的一致快照。

按拒绝原因定位下一步

本轮离线合成用例已执行状态分支、部分结果拒绝、ID 错配、类型缺失或 MUTATION 作业拒绝、孤立 Variant、重复行、截断 JSON、计数混用和 UInt64 上限检查。另有人工将 Variant 调到父行之前的压力用例，只检验此消费者按 ID 合并的行为，不能称为 Shopify 实测或官方顺序保证的反例。所有下载地址都是占位值，没有执行商家 Bulk 作业或真实下载。

留下作业 ID、原查询、实际版本、最终状态、两种计数、下载取得时间、文件哈希和验收结果，便于复核这次文件。需要进一步制定导出结果如何晋级、失败如何隔离及如何回滚的团队规则，可接着阅读GraphQL 数据处理与结果治理；本篇的接受结论止于指定作业的身份关系文件。