add stuff todo for fixing database

.gitignore

@@ -24,3 +24,5 @@ doc/pw-inventory/SQUARE_inventory_forautomation_2025-06-18-0139.csv
 doc/pw-inventory/square_inventory_export_2025-06-18-0137.xlsx
 
 square-src.code-workspace
+
+doc/pw-inventory/

TODO.md

@@ -2,3 +2,10 @@
 - test sq
 - test inserting brickowl into sq
 - test kopling
+
+
+square inv needs lego set category changed
+
+key lights
+pastel wrist things
+some sets

src/asquare.py

@@ -103,7 +103,11 @@ async def acat2id(cat_id: str):
         {
             "id": obj["id"],
             "name": obj.get("item_data", {}).get("name", "Unknown"),
-            "number": get_set_number(obj.get("item_data", {}).get("name", ""))
+            "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", [])
     ]

src/asquareclass.py

@@ -0,0 +1,218 @@
+import os
+import json
+import asyncio
+import re
+from typing import List, Dict, Any, Optional
+from square import AsyncSquare
+import get_sq_env
+
+
+class AsyncSquareClient:
+    """Asynchronous client for the Square API with focus on catalog operations"""
+
+    def __init__(
+        self,
+        token: Optional[str] = None,
+        environment: Optional[str] = None,
+        logger=None,
+    ):
+        """
+        Initialize the AsyncSquareClient.
+
+        Args:
+            token (str, optional): Square API token. If not provided, fetches from environment.
+            environment (str, optional): Square environment (sandbox/production). If not provided, fetches from environment.
+            logger: Optional logger object for logging operations.
+        """
+        self.token = token or get_sq_env.SQUARE_TOKEN
+        self.environment = environment or get_sq_env.environment
+        self.lego_cat_id = get_sq_env.LEGO_CAT_ID
+        self.client = AsyncSquare(token=self.token, environment=self.environment)  # type: ignore
+        self.logger = logger
+
+    @staticmethod
+    def get_set_number(text: str) -> int:
+        """
+        Extract a five-digit LEGO set number from a string.
+
+        Args:
+            text (str): The text to search for a set number.
+
+        Returns:
+            int: The extracted set number.
+
+        Raises:
+            ValueError: If no five-digit number is found.
+        """
+        numbers = re.findall(r"\d{5}", text)
+        if not numbers:
+            raise ValueError(f"No five-digit number found in string {text}")
+        return int(numbers[0])
+
+    async def get_item(self, sku: str) -> Dict[str, 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.
+        """
+        if self.logger:
+            self.logger.debug(f"Searching for item with SKU: {sku}")
+
+        result = await self.client.catalog.search_items(text_filter=sku, limit=1)
+        result_json = json.loads(result.model_dump_json())
+
+        try:
+            name = result_json["items"][0]["item_data"]["name"]
+            itm_id = result_json["items"][0]["id"]
+
+            if self.logger:
+                self.logger.info(f"Found item: {name} with ID: {itm_id}")
+
+            return {"name": name, "id": itm_id}
+        except (KeyError, IndexError) as e:
+            if self.logger:
+                self.logger.error(f"Error retrieving item with SKU {sku}: {str(e)}")
+            raise
+
+    async def update_item(self, lego_set: Dict[str, Any]) -> bool:
+        """
+        Asynchronously updates an item in the Square catalog based on provided LEGO set information.
+
+        Args:
+            lego_set (dict): A dictionary containing LEGO set details including:
+                - sets: List containing set information
+                - pw_name: PriceWise system name
+                - set_adjustredprice: Adjusted retail price
+
+        Returns:
+            bool: True if update was successful.
+
+        Raises:
+            KeyError: If required keys are missing from input dictionary.
+            Exception: For Square API communication errors.
+        """
+        if self.logger:
+            self.logger.debug(
+                f"Updating item with set info: {lego_set['sets'][0]['number']}"
+            )
+
+        try:
+            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 self.client.catalog.search_items(
+                text_filter=set_number, limit=1
+            )
+            itm_id = json.loads(result.model_dump_json())["items"][0]["id"]
+
+            if self.logger:
+                self.logger.info(f"Successfully updated item {set_number}")
+
+            # TODO: Implement actual update logic here
+            # await self.client.catalog.object.upsert()
+            return True
+
+        except Exception as e:
+            if self.logger:
+                self.logger.error(f"Error updating item: {str(e)}")
+            raise
+
+    async def get_category_items(
+        self, cat_id: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """
+        Return a list of items within the specified category ID.
+
+        Args:
+            cat_id (str, optional): The category ID to search for items.
+                                  If not provided, uses the default LEGO category ID.
+
+        Returns:
+            list[dict]: A list of dictionaries, each containing:
+                - 'id': The item's ID
+                - 'name': The item's name
+                - 'number': The LEGO set number
+        """
+        category_id = cat_id or self.lego_cat_id
+
+        if self.logger:
+            self.logger.debug(f"Fetching items for category: {category_id}")
+
+        try:
+            # Create a properly structured query for the Square API
+            search_request = {
+                "object_types": ["ITEM"],
+                "query": {
+                    "exact_query": {
+                        "attribute_name": "category_id",
+                        "attribute_value": category_id,
+                    }
+                },
+            }
+
+            result = await self.client.catalog.search(**search_request)
+            result_json = json.loads(result.model_dump_json())
+
+            items = [
+                {
+                    "id": obj["id"],
+                    "name": obj.get("item_data", {}).get("name", "Unknown"),
+                    "number": self.get_set_number(
+                        obj.get("item_data", {}).get("name", "")
+                    ),
+                }
+                for obj in result_json.get("objects", [])
+            ]
+
+            if self.logger:
+                self.logger.info(f"Found {len(items)} items in category {category_id}")
+
+            return items
+
+        except Exception as e:
+            if self.logger:
+                self.logger.error(f"Error fetching category items: {str(e)}")
+            raise
+
+
+# Provide acat2id as a module-level function for backward compatibility
+async def acat2id(cat_id: str) -> List[Dict[str, Any]]:
+    """
+    Backward compatibility function for getting category items.
+
+    Args:
+        cat_id (str): The category ID to search for items.
+
+    Returns:
+        list[dict]: A list of dictionaries containing item information.
+    """
+    client = AsyncSquareClient()
+    return await client.get_category_items(cat_id)
+
+
+async def test() -> None:
+    """Test function to demonstrate class usage"""
+    client = AsyncSquareClient()
+    print(f"@[DEBUG] ENV:\t{client.environment}")
+    print(f"@[DEBUG] CATID:\t{client.lego_cat_id}")
+
+    result = await client.get_category_items()
+    print("\nOBJECTS IN LEGO CATEGORY:")
+    for item in result:
+        print(f"ID: {item['id']}, Name: {item['name']}")
+
+
+if __name__ == "__main__":
+    asyncio.run(test())

src/test_server.py

@@ -11,42 +11,52 @@ import re
 import logging
 from logging.handlers import RotatingFileHandler
 from pathlib import Path
-
+print("Creating log directory if it doesn't exist...")
 Path("./logs/").mkdir(parents=True, exist_ok=True)
 
 import banner
 import get_sq_env
+
 import asquare as sq
+
+# from asquareclass import AsyncSquareClient  # type: ignore
 from abricksetapi import AsyncBricksetAPI
 
-
+print("Loading environment variables...")
 dotenv.load_dotenv()
 SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
 
-
 from asyncio_throttle.throttler import Throttler
 
 # set up logger
+print("Setting up logger...")
 logger = logging.getLogger("edderkop")
 logger.setLevel(logging.DEBUG)
-handler = RotatingFileHandler("./logs/", maxBytes=1024, backupCount=5)
+handler = RotatingFileHandler("./logs/edderkop.log", maxBytes=125000, backupCount=5)
 logger.addHandler(handler)
 
 lego = AsyncBricksetAPI(debug=True, logger=logger)
-
+# sq = AsyncSquareClient(logger=logger)
 # logging.basicConfig(
 #    filename="./logs/edderkop.log", encoding="utf-8", level=logging.DEBUG
 # )
-
+print("Setting up rate limiter...")
+print("Doing RE things...")
 UA = re.compile(r"^kopling\/?", re.IGNORECASE)
 
 
 async def run_update():
-    set_list = await sq.acat2id(SQUARE_LEGO_CAT_ID)  # type: ignore
-    for l_set in set_list:
+    set_list = await sq.acat2id(SQUARE_LEGO_CAT_ID)
+    print(f"[@DEBUG]\t {set_list}")  # type: ignore
+    """for l_set in set_list:
         print(f"[@DEBUG]{l_set}")
+        try:
             l_setinfo = await lego.throttled_get_set(l_set["number"])
-        print(f"[@DEBUG]{l_setinfo}")
+        except Exception as e:
+            print(f"Error fetching set\t: {l_set}: {str(e)}")
+            continue
+
+        print(f"[@DEBUG]{l_setinfo}")"""
     #    await sq.update_itm(l_set, l_setinfo)
 
 
@@ -101,11 +111,9 @@ async def request_handler(request: web.Request) -> web.Response:
 
             # Process the command
             if decoded_cmd == "RUNUPDATE":
-                # Add your command handling logic here
-                await run_update()
                 response_data = {"status": "success", "message": "Update started"}
+                await run_update()
                 logging.info(f"{time()}\tUpdate command processed successfully")
-
                 return web.json_response(response_data)
             else:
                 return web.Response(status=400, text=f"Unknown command: {decoded_cmd}")
@@ -118,7 +126,8 @@ async def request_handler(request: web.Request) -> web.Response:
         print(f"Request handler error: {str(e)}")
         return web.Response(status=500, text="Internal server error")
 
-
+print("Starting web server...")
+os.system('cls' if os.name == 'nt' else 'clear')
 app = web.Application()
 app.router.add_post("/", request_handler)