get-shopify-metafields.py

# Usage:
#    python3 check-metafields.py # everything
#    python3 check-metafields.py products # just products
#    python3 check-metafields.py collections # just collections
#    python3 check-metafields.py {PRODUCT_ID} # a specific product ID
#
import http.client
import json
import sys

# base URL for your target store
BASE_URL = ""

# create a header for the basic auth:
# https://shopify.dev/tutorials/authenticate-a-private-app-with-shopify-admin#basic-authentication
# echo -n '{username}:{password}' | openssl base64
AUTH = "Basic xxxx"

# hide index-out-of-range errors but keep ternary
arg1 = (sys.argv, [0, False])[len(sys.argv) == 1][1]

try:
    # how many products or collections to loop over
    limit = int(arg1) if arg1 and int(arg1) else 250
    get_product_id = int(arg1) if arg1 and int(arg1) > 1000 else False
except:
    limit = 250
    get_product_id = False

only_collections = True if arg1 == "collections" and get_product_id == False else False
only_products = True if arg1 == "products" and get_product_id == False else False

results = {
    'limit': limit,
    'products': [],
    'collections': [],
    'warnings': []
}

filename = "product-and-collection-metafields-%s.json" % limit
output = open(filename, "w")


def closeOut():
    output.write(json.dumps(results, sort_keys=False, indent=2))
    output.close()
    print("write out: " + filename)
    exit()

def request(url: str):
    conn = http.client.HTTPSConnection(BASE_URL)

    payload = ''

    headers = {
        'Authorization': AUTH,
        'Content-type': 'application/json'
    }

    conn.request("GET", "/admin/api/2020-01" + url, payload, headers)

    res = conn.getresponse()

    data = res.read().decode("utf-8")

    return json.loads(data)


def getMetafields(meta):
    fields = {}
    if "metafields" in meta:
        for field in meta["metafields"]:
            key = "%s:%s" % (field["namespace"], field["key"])
            # ignore judgeme and yotpo fields
            if "judge" not in key and "yotpo" not in key:
                fields[key] = field["value"]

    return fields


def getVariantMetafields(product, variants):
    fields = {}

    for variant in variants:
        product_variants = request("/products/%s/variants/%s/metafields.json" % (product, variant["id"]))
        print("Fetching product %s metafields for variant: %s" % (product, variant["id"]))

        fields[variant["id"]] = {}
        fields[variant["id"]]["options"] = "%s, %s, %s" % (variant["option1"], variant["option2"], variant["option3"])
        fields[variant["id"]]["sku"] = variant["sku"]

        if "metafields" in product_variants and len(product_variants["metafields"]) < 1:
            warning = "NO METAFIELDS FOUND FOR: product %s variant %s" % (str(product), str(variant["id"]))
            print(warning)
            warning2 = "https://%s/admin/products/%s.json" % (BASE_URL, str(product))
            warning3 = "https://%s/admin/products/%s/variants/%s/metafields.json" % (BASE_URL, str(product), str(variant["id"]))
            results["warnings"].append([warning, warning2, warning3])
            continue

        if "metafields" in product_variants:
            for field in product_variants["metafields"]:
                key = "%s:%s" % (field["namespace"], field["key"])
                # ignore judgeme and yotpo fields
                if "judge" not in key and "yotpo" not in key:
                    fields[variant["id"]][key] = field["value"]

    return fields


def getProduct(product):
    print("Fetching metafields for product: " + str(product["id"]))
    product_meta = request("/products/%s/metafields.json" % product["id"])

    fields = getMetafields(product_meta)

    varaiant_fields = getVariantMetafields(str(product["id"]), product["variants"])

    results["products"].append({
        'id': product["id"],
        'handle': product["handle"],
        'tags': product["tags"],
        'fields': fields,
        'variantFields': varaiant_fields
        })

if get_product_id:
    product = request("/products/%s.json" % get_product_id)
    getProduct(product["product"])
    closeOut()

if only_collections == True:
    print("Skipping products...")
else:
    products = request("/products.json?limit=250")
    print("Fetching products...")

    for index, product in enumerate(products["products"]):
        if index == limit:
            break

        getProduct(product)

    if only_products == True:
        closeOut()

collections = request("/custom_collections.json?limit=250")
print("Fetching collections...")

for index, collection in enumerate(collections["custom_collections"]):
    if index == limit:
        break

    print("Fetching metafields for collection:" + str(collection["id"]))
    collection_meta = request("/collections/%s/metafields.json" % collection["id"])

    fields = getMetafields(collection_meta)

    results["collections"].append({
        'id': collection["id"],
        'handle': collection["handle"],
        'image': collection["image"]["src"] if "image" in collection else False,
        'fields': fields
        })

print("count for products:" + str(len(results["products"])))
print("count for collections:" + str(len(results["collections"])))
print("count for warnings:" + str(len(results["warnings"])))

output.write(json.dumps(results, sort_keys=False, indent=2))
output.close()
print("write out: " + filename)