import os import dotenv import json import asyncio import uuid import re from time import time 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. Args: item_id: Square catalog item ID Returns: dict: Item details including name, category_id, variations, etc. Raises: Exception: If item not found or API error """ try: result = await client.catalog.batch_retrieve_catalog_objects( body={"object_ids": [item_id], "include_related_objects": True} ) result_json = json.loads(result.model_dump_json()) if not result_json.get("objects"): raise ValueError(f"Item {item_id} not found") 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 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. Args: variation_id: Square catalog item variation ID Returns: dict: Parent item details Raises: Exception: If variation/item not found or API error """ try: # Get variation object using batch_retrieve result = await client.catalog.batch_retrieve_catalog_objects( body={"object_ids": [variation_id], "include_related_objects": True} ) result_json = json.loads(result.model_dump_json()) if not result_json.get("objects"): raise ValueError(f"Variation {variation_id} not found") variation_obj = result_json["objects"][0] # Get parent item ID from variation item_id = variation_obj.get("item_variation_data", {}).get("item_id") if not item_id: raise ValueError(f"No parent item_id found for variation {variation_id}") # Retrieve parent item return await get_item_by_id(item_id) 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): """ 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 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) # Get current item to retrieve variation ID item = await get_item_by_id(item_id) variations = item.get("variations", []) if not variations: print( f"[@WARNING] Item {item_id} has no variations, creating new variation" ) variation_id = None else: variation_id = variations[0]["id"] # Build upsert body upsert_body = { "idempotency_key": str(uuid.uuid4()), "object": { "type": "ITEM", "id": item_id, "item_data": { "name": pw_name, "description": f"{set_name} - {set_theme}", "variations": [ { "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, }, } ], }, }, } # Remove None values if not set_upc: del upsert_body["object"]["item_data"]["variations"][0][ "item_variation_data" ]["sku"] print(f"[@DEBUG] Upserting item {item_id} with name: {pw_name}") # Perform upsert result = await client.catalog.upsert_catalog_object(body=upsert_body) 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 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())