From 94af7d9d89a4286529d11b179893c5204769fd54 Mon Sep 17 00:00:00 2001 From: C <0xicl33n@gmail.com> Date: Sun, 4 Jan 2026 14:08:19 -0800 Subject: [PATCH] derp --- src/scripts/migrate_sku_to_gtin.py | 165 +++++++++++++++++++++++++++++ 1 file changed, 165 insertions(+) create mode 100644 src/scripts/migrate_sku_to_gtin.py diff --git a/src/scripts/migrate_sku_to_gtin.py b/src/scripts/migrate_sku_to_gtin.py new file mode 100644 index 0000000..5b06121 --- /dev/null +++ b/src/scripts/migrate_sku_to_gtin.py @@ -0,0 +1,165 @@ +""" +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())