166 lines
6.0 KiB
Python
166 lines
6.0 KiB
Python
"""
|
|
Migrate SKU values to GTIN/UPC field for all LEGO items.
|
|
|
|
This script reads existing Square catalog items in the LEGO category,
|
|
and for each variation that has a SKU but no UPC, it moves the SKU
|
|
value to the UPC field (leaving SKU empty or optionally extracting
|
|
set number from item name).
|
|
"""
|
|
|
|
import asyncio
|
|
import json
|
|
import re
|
|
import sys
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
# Add parent directory to path for imports
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
from asquare import AsyncSquare, get_square_client
|
|
from get_sq_env import SQUARE_LEGO_CAT_ID
|
|
|
|
|
|
async def migrate_sku_to_gtin():
|
|
"""Migrate SKU values to GTIN for all LEGO items."""
|
|
client = await get_square_client()
|
|
asquare = AsyncSquare()
|
|
|
|
print(f"[@INFO] Searching for items in LEGO category {SQUARE_LEGO_CAT_ID}...")
|
|
|
|
# Search all items in LEGO category
|
|
cursor = None
|
|
page = 1
|
|
total_items = 0
|
|
updated_items = 0
|
|
|
|
while True:
|
|
print(f"[@INFO] Processing page {page}...")
|
|
|
|
result = await client.catalog.search(
|
|
body={
|
|
"object_types": ["ITEM"],
|
|
"query": {
|
|
"exact_query": {
|
|
"attribute_name": "category_id",
|
|
"attribute_value": SQUARE_LEGO_CAT_ID,
|
|
}
|
|
},
|
|
"limit": 100,
|
|
"cursor": cursor,
|
|
}
|
|
)
|
|
|
|
result_json = json.loads(result.model_dump_json())
|
|
items = result_json.get("objects", [])
|
|
cursor = result_json.get("cursor")
|
|
|
|
if not items:
|
|
break
|
|
|
|
# Process each item
|
|
for item in items:
|
|
item_id = item["id"]
|
|
item_data = item.get("item_data", {})
|
|
item_name = item_data.get("name", "")
|
|
variations = item_data.get("variations", [])
|
|
item_version = item.get("version")
|
|
|
|
total_items += 1
|
|
needs_update = False
|
|
updated_variations = []
|
|
|
|
# Extract set number from item name if it matches LEGO format
|
|
set_number = None
|
|
match = re.search(r"^(\d{5})\s*-", item_name)
|
|
if match:
|
|
set_number = match.group(1)
|
|
|
|
# Check each variation
|
|
for var in variations:
|
|
var_data = var.get("item_variation_data", {})
|
|
var_id = var.get("id")
|
|
var_version = var.get("version")
|
|
sku = var_data.get("sku")
|
|
upc = var_data.get("upc")
|
|
|
|
# If SKU exists but UPC doesn't, migrate
|
|
if sku and not upc:
|
|
print(f"[@INFO] Item {item_id} ({item_name})")
|
|
print(f"[@INFO] Variation {var_id}: Moving SKU '{sku}' to UPC")
|
|
|
|
# Build updated variation
|
|
updated_var = {
|
|
"type": "ITEM_VARIATION",
|
|
"id": var_id,
|
|
"version": var_version,
|
|
"item_variation_data": {
|
|
"item_id": item_id,
|
|
"name": var_data.get("name", "Regular"),
|
|
"pricing_type": var_data.get("pricing_type", "FIXED_PRICING"),
|
|
"upc": sku, # Move SKU value to UPC
|
|
"sku": set_number if set_number else None, # Set to LEGO set number if available
|
|
}
|
|
}
|
|
|
|
# Preserve pricing info
|
|
if "price_money" in var_data:
|
|
updated_var["item_variation_data"]["price_money"] = var_data["price_money"]
|
|
|
|
updated_variations.append(updated_var)
|
|
needs_update = True
|
|
else:
|
|
# Keep variation as-is
|
|
updated_variations.append(var)
|
|
|
|
# Update item if needed
|
|
if needs_update:
|
|
try:
|
|
# Get fresh item data to ensure we have latest version
|
|
fresh_result = await client.catalog.batch_get(
|
|
object_ids=[item_id],
|
|
include_related_objects=False,
|
|
)
|
|
fresh_json = json.loads(fresh_result.model_dump_json())
|
|
fresh_item = fresh_json["objects"][0]
|
|
fresh_item_data = fresh_item.get("item_data", {})
|
|
fresh_version = fresh_item.get("version")
|
|
|
|
# Update variations while preserving all other fields
|
|
fresh_item_data["variations"] = updated_variations
|
|
|
|
upsert_body = {
|
|
"idempotency_key": str(uuid.uuid4()),
|
|
"object": {
|
|
"type": "ITEM",
|
|
"id": item_id,
|
|
"version": fresh_version,
|
|
"item_data": fresh_item_data,
|
|
}
|
|
}
|
|
|
|
await client.catalog.object.upsert(body=upsert_body)
|
|
print(f"[@INFO] ✓ Successfully updated item {item_id}")
|
|
updated_items += 1
|
|
|
|
# Small delay to avoid rate limits
|
|
await asyncio.sleep(0.1)
|
|
|
|
except Exception as e:
|
|
print(f"[@ERROR] Failed to update item {item_id}: {e}")
|
|
|
|
page += 1
|
|
|
|
# If no cursor, we're done
|
|
if not cursor:
|
|
break
|
|
|
|
print(f"\n[@INFO] Migration complete!")
|
|
print(f"[@INFO] Total items processed: {total_items}")
|
|
print(f"[@INFO] Items updated: {updated_items}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
print("[@INFO] Starting SKU to GTIN migration...\n")
|
|
asyncio.run(migrate_sku_to_gtin())
|