add stuff todo for fixing database

.gitignore

@@ -9,6 +9,7 @@ __pycache__/
 *.sqlite3
 *.log
 poetry.lock
+uv.lock
 bin/build/
 .zip
 snippet.py
@@ -23,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/

README.md

@@ -1,6 +1,7 @@
 # Edderkop
 
 environment: poetry
+
 code style: black
 
 built with

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

pyproject.toml

@@ -2,12 +2,10 @@
 name = "edderkop"
 version = "0.1.0"
 description = ""
-authors = [
+authors = [{ name = "CF", email = "ph0sgene@pm.me" }]
-    {name = "CF", email = "chris-forst@pm.me"}
-]
-license = {text = "MIT"}
-readme = "README.md"
 requires-python = ">=3.13,<3.14"
+readme = "README.md"
+license = { text = "MIT" }
 dependencies = [
     "aiohttp (>=3.12.13,<4.0.0)",
     "dotenv (>=0.9.9,<0.10.0)",
@@ -20,13 +18,12 @@ dependencies = [
     "aiosqlite (>=0.21.0,<0.22.0)",
 ]
 
-
 [build-system]
-requires = ["poetry-core>=2.0.0,<3.0.0"]
+requires = ["hatchling"]
-build-backend = "poetry.core.masonry.api"
+build-backend = "hatchling.build"
 
-[tool.poetry]
+[tool.uv]
-package-mode = false
+package = false
 
 [tool.bricksetapi]
 version = "0.1.0"

src/abricksetapi.py

@@ -17,7 +17,7 @@ dotenv.load_dotenv()
 class AsyncBricksetAPI:
     """Asynchronous client for the Brickset API"""
 
-    def __init__(self, debug: bool = False):
+    def __init__(self, debug: bool = False, logger=None):
         self.username = os.getenv("BRICKSET_USERNAME")
         self.password = os.getenv("BRICKSET_PASS")
         self.api_key = os.getenv("BRICKSET_API_KEY")
@@ -25,6 +25,7 @@ class AsyncBricksetAPI:
         self.headers = {"UserAgent": "Edderkop/0.1"}
         self.throttler = Throttler(rate_limit=500, period=60)
         self.user_hash = None
+        self.logger = logger
 
     async def login(self) -> Dict[str, Any]:
         """Asynchronously authenticate with Brickset API"""
@@ -41,10 +42,15 @@ class AsyncBricksetAPI:
             ) as response:
                 if self.debug:
                     print(f"[@DEBUG] Login response status: {response.status}")
+                if self.logger:
+                    self.logger.debug(f"Login response status: {response.status}")
 
                 data = await response.read()
                 result = json.loads(data.decode("utf-8"))
                 self.user_hash = result.get("hash")
+
+                if self.logger:
+                    self.logger.info("Successfully authenticated with Brickset API")
                 return result
 
     async def get_set(self, set_number: str) -> Dict[str, Any]:
@@ -73,13 +79,23 @@ class AsyncBricksetAPI:
             ) as response:
                 if self.debug:
                     print(f"[@DEBUG] Get set response status: {response.status}")
+                if self.logger:
+                    self.logger.debug(f"Get set response status: {response.status}")
 
                 if response.status == 200:
                     data = await response.read()
                     result = json.loads(data.decode("utf-8"))
 
                     if not result.get("sets"):
-                        raise ValueError(f"No data found for set {set_number}")
+                        error_msg = f"No data found for set {set_number}"
+                        if self.logger:
+                            self.logger.error(error_msg)
+                        raise ValueError(error_msg)
+
+                    if self.logger:
+                        self.logger.info(
+                            f"Successfully retrieved data for set {set_number}"
+                        )
 
                     set_data = result["sets"][0]
                     our_set = {
@@ -111,11 +127,17 @@ class AsyncBricksetAPI:
                     return our_set
 
                 elif response.status == 500:
-                    raise Exception(
+                    error_msg = (
                         "Internal Server Error: Please check API key and parameters."
                     )
+                    if self.logger:
+                        self.logger.error(error_msg)
+                    raise Exception(error_msg)
                 else:
-                    raise Exception(f"Unexpected response: {response.status}")
+                    error_msg = f"Unexpected response: {response.status}"
+                    if self.logger:
+                        self.logger.error(error_msg)
+                    raise Exception(error_msg)
 
     async def throttled_get_set(self, set_number: str) -> Dict[str, Any]:
         """Get set information with rate limiting"""

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

@@ -9,37 +9,54 @@ import dotenv
 import os
 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
 
-lego = AsyncBricksetAPI(debug=True)
+print("Loading environment variables...")
-
 dotenv.load_dotenv()
 SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
 
-
 from asyncio_throttle.throttler import Throttler
 
-logger = logging.getLogger("aiohttp.access")
+# set up logger
-logging.basicConfig(
+print("Setting up logger...")
-    filename="./logs/edderkop.log", encoding="utf-8", level=logging.DEBUG
+logger = logging.getLogger("edderkop")
-)
+logger.setLevel(logging.DEBUG)
+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)
-    for l_set in set_list:
+    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)
 
 
@@ -94,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}")
@@ -111,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)