{
  "identifier": { "handle": "winter-hat" },
  "input": { "handle": "winter-hat", "title": "Winter hat" }
}

import csv, json, sys, collections
 
LOCATION_ID = sys.argv[2]  # gid://shopify/Location/...
rows = collections.defaultdict(list)
with open(sys.argv[1], newline="", encoding="utf-8") as f:
    for r in csv.DictReader(f):
        rows[r["handle"]].append(r)
 
for handle, rs in rows.items():
    head = rs[0]
    names = [head[k] for k in ("opt1_name", "opt2_name") if head[k]]
    if not names:                        # オプションなし商品は別扱いにする
        print(f"skip (no options): {handle}", file=sys.stderr)
        continue
    values = {n: [] for n in names}      # 出現順を保つ
    variants, files, seen = [], [], set()
    for r in rs:
        ov = []
        for i, n in enumerate(names, 1):
            v = r[f"opt{i}_value"]
            if v not in values[n]:
                values[n].append(v)
            ov.append({"optionName": n, "name": v})
        variants.append({
            "sku": r["sku"],
            "price": r["price"],
            "optionValues": ov,
            "inventoryQuantities": [{
                "locationId": LOCATION_ID, "name": "available",
                "quantity": int(r["qty"])}],
        })
        if r["image_url"] and r["image_url"] not in seen:
            seen.add(r["image_url"])
            files.append({"originalSource": r["image_url"],
                          "alt": head["title"], "contentType": "IMAGE"})
    if len(variants) > 250:              # 入力配列の上限
        print(f"skip (>250 variants): {handle}", file=sys.stderr)
        continue
    options = [{"name": n, "values": [{"name": v} for v in values[n]]}
               for n in names]
    print(json.dumps({
        "identifier": {"handle": handle},
        "input": {"handle": handle, "title": head["title"],
                  "vendor": head["vendor"], "status": "DRAFT",
                  "productOptions": options, "variants": variants,
                  "files": files}}, ensure_ascii=False))

skip (no options): bike-frame

{
  "permissions": {
    "deny": ["Read(./.env)", "Bash(curl *)"]
  }
}

## Shopify商品移行
- トークンは読まない・表示しない。`SHOPIFY_TOKEN` は push.py だけが使う
- productSet は variants を全置換する。CSV は handle ごとに全行そろえて変換する
- 変換後は out.jsonl の行数と標準エラーのスキップ数を CSV の handle 数と突き合わせて報告する

import json, os, sys, urllib.request
 
URL = (f"https://{os.environ['SHOP']}.myshopify.com"
       f"/admin/api/2026-10/graphql.json")
MUTATION = """
mutation($input: ProductSetInput!, $identifier: ProductSetIdentifiers) {
  productSet(input: $input, identifier: $identifier) {
    product { handle variants(first: 250) { nodes { sku } } }
    userErrors { field message }
  }
}"""
 
for line in open(sys.argv[1], encoding="utf-8"):
    body = json.loads(line)
    req = urllib.request.Request(
        URL, method="POST",
        headers={"Content-Type": "application/json",
                 "X-Shopify-Access-Token": os.environ["SHOPIFY_TOKEN"]},
        data=json.dumps({"query": MUTATION, "variables": body}).encode())
    res = json.load(urllib.request.urlopen(req))
    print(json.dumps(res, ensure_ascii=False))

import csv, json, sys, collections
 
want = collections.defaultdict(set)          # handle -> CSV の SKU 集合
for r in csv.DictReader(open(sys.argv[1], newline="", encoding="utf-8")):
    want[r["handle"]].add(r["sku"])
 
got = {}                                     # handle -> Shopify 側の SKU 集合
for line in open(sys.argv[2], encoding="utf-8"):
    p = json.loads(line)["data"]["productSet"]
    if p["userErrors"] or not p["product"]:
        print("ERROR", p["userErrors"]); continue
    pr = p["product"]
    got[pr["handle"]] = {v["sku"] for v in pr["variants"]["nodes"]}
 
print(f"products: csv={len(want)} shopify={len(got)}")
for h in sorted(want):
    miss, extra = want[h] - got.get(h, set()), got.get(h, set()) - want[h]
    if miss or extra:
        print(h, "missing:", sorted(miss), "extra:", sorted(extra))

products: csv=2 shopify=2
winter-hat missing: ['HAT-BLK-M']