163 lines
5.4 KiB
Python
163 lines
5.4 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))
|
|
|
|
import get_sq_env
|
|
from square import AsyncSquare
|
|
|
|
SQUARE_LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID
|
|
|
|
|
|
async def migrate_sku_to_gtin():
|
|
"""Migrate SKU values to GTIN for all LEGO items."""
|
|
client = AsyncSquare(
|
|
token=get_sq_env.SQUARE_TOKEN,
|
|
environment=get_sq_env.environment,
|
|
)
|
|
|
|
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}...")
|
|
|
|
search_params = {
|
|
"object_types": ["ITEM"],
|
|
"query": {
|
|
"exact_query": {
|
|
"attribute_name": "category_id",
|
|
"attribute_value": SQUARE_LEGO_CAT_ID,
|
|
}
|
|
},
|
|
"limit": 100,
|
|
}
|
|
|
|
if cursor:
|
|
search_params["cursor"] = cursor
|
|
|
|
result = await client.catalog.search(**search_params)
|
|
|
|
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")
|
|
|
|
# Preserve all existing variation data and only modify what we need
|
|
updated_var = var.copy() # Shallow copy the entire variation
|
|
updated_var_data = var_data.copy() # Copy the variation_data
|
|
|
|
# Update only UPC and SKU fields
|
|
updated_var_data["upc"] = sku # Move SKU value to UPC
|
|
updated_var_data["sku"] = (
|
|
set_number if set_number else None
|
|
) # Set to LEGO set number
|
|
|
|
updated_var["item_variation_data"] = updated_var_data
|
|
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
|
|
|
|
await client.catalog.object.upsert(
|
|
idempotency_key=str(uuid.uuid4()),
|
|
object={
|
|
"type": "ITEM",
|
|
"id": item_id,
|
|
"version": fresh_version,
|
|
"item_data": fresh_item_data,
|
|
},
|
|
)
|
|
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())
|