import os import dotenv import json import asyncio import uuid import re from time import time from square.core.api_error import ApiError from . import db from . import get_sq_env def get_set_number(text: str) -> int: # Remove all letters, keeping only digits and spaces numbers = re.findall(r"\d{5}", text) if not numbers: raise ValueError("No five-digit number found in string") return int(numbers[0]) SQUARE_TOKEN = get_sq_env.SQUARE_TOKEN LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID ENVIRONMENT = get_sq_env.environment from square import AsyncSquare from square.environment import SquareEnvironment client = AsyncSquare(token=SQUARE_TOKEN, environment=ENVIRONMENT) # type: ignore async def get_itm(sku: str): """ Asynchronously retrieves item details from Square Catalog using a SKU. Args: sku (str): The SKU or text filter to search for the item. Returns: dict: A dictionary containing the item's name and ID. Raises: IndexError: If no items are found matching the SKU. Exception: If there is an error communicating with the Square API. """ result = await client.catalog.search_items(text_filter=sku, limit=1) result_json = json.loads(result.model_dump_json()) name = result_json["items"][0]["item_data"]["name"] itm_id = result_json["items"][0]["id"] return {"name": name, "id": itm_id} async def get_item_by_id(item_id: str) -> dict: """ Retrieve full item details by Square catalog item ID. Uses batch_get API for direct ID lookup. Args: item_id: Square catalog item ID Returns: dict: Item details including name, category_id, variations, etc. Raises: ValueError: If item not found Exception: If API error """ try: # Use batch_get to retrieve item directly by ID result = await client.catalog.batch_get( object_ids=[item_id], include_related_objects=False, ) result_json = json.loads(result.model_dump_json()) if not result_json.get("objects"): raise ValueError(f"Item {item_id} not found (may have been deleted)") obj = result_json["objects"][0] item_data = obj.get("item_data", {}) return { "id": obj["id"], "name": item_data.get("name", ""), "description": item_data.get("description", ""), "category_id": item_data.get("category_id"), "variations": item_data.get("variations", []), "updated_at": obj.get("updated_at"), "version": obj.get("version"), } except ValueError: raise except Exception as e: print(f"[@ERROR] Failed to get item {item_id}: {e}") raise async def get_item_by_variation_id(variation_id: str) -> dict: """ Retrieve parent item details from a variation ID. Webhooks send variation IDs (inventory.count.updated), but we need the parent item to extract the set number from the name. Since async Square API lacks direct object retrieval, we search all items in the LEGO category and match variations. Args: variation_id: Square catalog item variation ID Returns: dict: Parent item details Raises: Exception: If variation/item not found or API error """ try: # Search LEGO category items to find which one has this variation print(f"[@DEBUG] Searching for variation {variation_id} in LEGO category") cursor = None page_count = 0 while True: page_count += 1 print(f"[@DEBUG] Searching page {page_count}...") search_params = { "object_types": ["ITEM"], "query": { "exact_query": { "attribute_name": "category_id", "attribute_value": LEGO_CAT_ID, } }, "limit": 1000, } if cursor: search_params["cursor"] = cursor result = await client.catalog.search(**search_params) result_json = json.loads(result.model_dump_json()) if not result_json.get("objects"): if page_count == 1: raise ValueError(f"No items found in LEGO category") else: # No more results break # Search through items on this page to find which has this variation for obj in result_json["objects"]: item_data = obj.get("item_data", {}) variations = item_data.get("variations", []) for variation in variations: if variation.get("id") == variation_id: # Found the parent item! print( f"[@DEBUG] Found parent item: {obj['id']} on page {page_count}" ) # Since we searched by category, we know it's in LEGO_CAT_ID # But also try to get from item_data as fallback category_id = item_data.get("category_id") or LEGO_CAT_ID print( f"[@DEBUG] Item category_id from data: {item_data.get('category_id')}" ) print(f"[@DEBUG] Using category_id: {category_id}") return { "id": obj["id"], "name": item_data.get("name", ""), "description": item_data.get("description", ""), "category_id": category_id, "variations": variations, "updated_at": obj.get("updated_at"), "version": obj.get("version"), } # Check if there are more pages cursor = result_json.get("cursor") if not cursor: # No more pages break raise ValueError( f"Variation {variation_id} not found in any LEGO category item (searched {page_count} pages)" ) except Exception as e: print(f"[@ERROR] Failed to get item from variation {variation_id}: {e}") raise async def update_itm(item_id: str, lego_set: dict, item_data: dict = None): """ Asynchronously updates an item in the Square catalog based on the provided LEGO set information. Args: item_id (str): Square catalog item ID to update lego_set (dict): A dictionary containing LEGO set details from abricksetapi. Expected keys include: - "set_num": LEGO set number (e.g., "75192") - "set_name": Official LEGO set name - "set_theme": LEGO theme (e.g., "Star Wars") - "set_barcode": UPC barcode - "set_image": Thumbnail URL - "pw_name": Formatted name for Square inventory (e.g., "75192 - STAR WARS UCS Millennium Falcon") - "set_adjustredprice": Retail price with 10% markup applied item_data (dict): Optional pre-fetched item data to avoid redundant API call Returns: bool: True if the update operation was successful. Raises: KeyError: If expected keys are missing from the input dictionary. Exception: For errors during Square API interaction. """ try: set_number = lego_set["set_num"] set_name = lego_set["set_name"] set_theme = lego_set["set_theme"] set_upc = lego_set["set_barcode"] set_image = lego_set["set_image"] pw_name = lego_set["pw_name"] set_adjustedprice = lego_set["set_adjustredprice"] # Convert price to cents for Square API price_in_cents = int(set_adjustedprice * 100) # Retry logic for version conflicts (race conditions with concurrent updates) max_retries = 3 retry_delay = 0.5 # seconds for attempt in range(max_retries): try: # ALWAYS fetch fresh item data right before update to get latest version print( f"[@DEBUG] Fetching fresh item data (attempt {attempt + 1}/{max_retries})..." ) fresh_item = await get_item_by_id(item_id) except ValueError as e: # Item was deleted between webhook and enrichment print(f"[@WARNING] {e} - skipping update (item may have been deleted)") return variations = fresh_item.get("variations", []) if not variations: print( f"[@WARNING] Item {item_id} has no variations, creating new variation" ) variation_id = None variation_version = None else: variation_id = variations[0]["id"] variation_version = variations[0].get("version") # Get current version for version control current_version = fresh_item.get("version") category_id = fresh_item.get("category_id") print(f"[@DEBUG] Fresh item data: {fresh_item}") print(f"[@DEBUG] Fresh item version: {current_version}") print(f"[@DEBUG] Fresh variation version: {variation_version}") print(f"[@DEBUG] Category ID from fresh_item: {category_id}") # Ensure we have a category_id if not category_id: print( f"[@WARNING] No category_id in fresh_item, using LEGO_CAT_ID: {LEGO_CAT_ID}" ) category_id = LEGO_CAT_ID # Build upsert body variation_obj = { "type": "ITEM_VARIATION", "id": variation_id if variation_id else "#TEMP_VARIATION_ID", "item_variation_data": { "item_id": item_id, "name": "Regular", "pricing_type": "FIXED_PRICING", "price_money": { "amount": price_in_cents, "currency": "USD", }, "sku": set_upc if set_upc else None, }, } # Include variation version if it exists (required for updates) if variation_version is not None: variation_obj["version"] = variation_version upsert_body = { "idempotency_key": str(uuid.uuid4()), "object": { "type": "ITEM", "id": item_id, "version": current_version, # Include version for conflict prevention "present_at_all_locations": False, # Required when using category_id "item_data": { "name": pw_name, "description": f"{set_name} - {set_theme}", "category_id": category_id, # Preserve category assignment "variations": [variation_obj], }, }, } # Remove None values if not set_upc: del variation_obj["item_variation_data"]["sku"] print(f"[@DEBUG] Upserting item {item_id} with name: {pw_name}") print(f"[@DEBUG] Upsert body: {json.dumps(upsert_body, indent=2)}") # Perform upsert using catalog.object.upsert try: result = await client.catalog.object.upsert( idempotency_key=upsert_body["idempotency_key"], object=upsert_body["object"], ) result_json = json.loads(result.model_dump_json()) if result_json.get("catalog_object"): print(f"[@DEBUG] ✓ Successfully updated item {item_id}") # TODO: Handle image upload separately (Square Images API) # This requires multipart/form-data upload return True else: print( f"[@ERROR] Failed to update item {item_id}: No catalog_object in response" ) return False except ApiError as e: # Check if this is a VERSION_MISMATCH error if e.status_code == 400 and e.body and isinstance(e.body, dict): errors = e.body.get("errors", []) if any(err.get("code") == "VERSION_MISMATCH" for err in errors): if attempt < max_retries - 1: print( f"[@WARNING] Version conflict on attempt {attempt + 1}, retrying in {retry_delay}s..." ) await asyncio.sleep(retry_delay) retry_delay *= 2 # Exponential backoff continue # Retry with fresh data else: print( f"[@ERROR] Version conflict persists after {max_retries} attempts, giving up" ) return False # Re-raise other API errors raise except Exception as e: print(f"[@ERROR] Failed to update item {item_id}: {e}") import traceback traceback.print_exc() return False async def acat2id(cat_id: str): """ Return a list of dictionaries containing item IDs and names within the specified category ID. Args: cat_id (str): The category ID to search for items. Returns: list[dict]: A list of dictionaries, each containing: - 'id': The item's ID - 'name': The item's name """ result = await client.catalog.search( object_types=["ITEM"], query={ "exact_query": { "attribute_name": "category_id", "attribute_value": cat_id, } }, ) result_json = json.loads(result.model_dump_json()) items = [ { "id": obj["id"], "name": obj.get("item_data", {}).get("name", "Unknown"), "number": ( None if not obj.get("item_data", {}).get("name", "") else ( get_set_number(obj.get("item_data", {}).get("name", "")) if re.findall(r"\d{5}", obj.get("item_data", {}).get("name", "")) else None ) ), "updated_at": obj.get("updated_at"), "version": obj.get("version"), } for obj in result_json.get("objects", []) ] return items async def test() -> None: result = await acat2id(LEGO_CAT_ID) # type: ignore print("OBJECTS IN LEGO CATEGORY:") for item in result: print(f"ID: {item['id']}, Name: {item['name']}, Updated: {item['updated_at']}") """ result = await sq.catalog.search_items(text_filter="71433", limit=1) result_json = json.loads(result.model_dump_json()) name = result_json["items"][0]["item_data"]["name"] print(f"Item name: {name}")""" if __name__ == "__main__": print(f"@[DEBUG] ENV:\t{ENVIRONMENT}") print(f"@[DEBUG] CATID:\t{LEGO_CAT_ID}") asyncio.run(test())