140 lines
4.4 KiB
Python
140 lines
4.4 KiB
Python
import os
|
|
import dotenv
|
|
import json
|
|
import asyncio
|
|
import uuid
|
|
import re
|
|
|
|
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 update_itm(lego_set: dict):
|
|
"""
|
|
Asynchronously updates an item in the Square catalog based on the provided LEGO set information.
|
|
Args:
|
|
lego_set (dict): A dictionary containing LEGO set details. Expected keys include:
|
|
- "sets": List of set dictionaries, each with "name", "number", "barcode" (with "UPC"), and "image" (with "thumbnailURL").
|
|
- "pw_name": The name associated with the LEGO set in another system.
|
|
- "set_adjustredprice": The adjusted price for the set.
|
|
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.
|
|
"""
|
|
|
|
set_name = lego_set["sets"][0]["name"]
|
|
set_number = lego_set["sets"][0]["number"]
|
|
set_upc = lego_set["sets"][0]["barcode"]["UPC"]
|
|
set_image = lego_set["sets"][0]["image"]["thumbnailURL"]
|
|
pw_name = lego_set["pw_name"]
|
|
set_adjustedprice = lego_set["set_adjustredprice"]
|
|
|
|
result = await client.catalog.search_items(text_filter=set_number, limit=1)
|
|
itm_id = json.loads(result.model_dump_json())["items"][0]["id"]
|
|
|
|
# await sq.catalog.object.upsert()
|
|
return True
|
|
|
|
|
|
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
|
|
)
|
|
}
|
|
for obj in result_json.get("objects", [])
|
|
]
|
|
|
|
return items
|
|
|
|
|
|
async def test() -> None:
|
|
task = asyncio.create_task(acat2id(LEGO_CAT_ID)) # type: ignore
|
|
await asyncio.sleep(1) # Give the task some time to complete
|
|
result = await task
|
|
print("OBJECTS IN LEGO CATEGORY:")
|
|
for item in result:
|
|
print(f"ID: {item['id']}, Name: {item['name']}")
|
|
# eturn
|
|
# pass
|
|
|
|
|
|
"""
|
|
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())
|