diff --git a/src/abricksetapi.py b/src/abricksetapi.py index 482ab19..d4d18f2 100644 --- a/src/abricksetapi.py +++ b/src/abricksetapi.py @@ -8,7 +8,8 @@ import os import aiohttp import json import dotenv -from typing import Dict, Any +from typing import Dict, Any, Optional +from datetime import datetime, timedelta from . import throttler Throttler = throttler.GlobalThrottler @@ -29,6 +30,104 @@ class AsyncBricksetAPI: self.user_hash = None self.logger = logger + # API usage tracking + self.daily_requests_used: Optional[int] = None + self.daily_requests_limit = 1000 + self.usage_last_checked: Optional[datetime] = None + self.usage_check_interval = timedelta(hours=1) # Check usage every hour + + async def get_key_usage_stats(self) -> Dict[str, Any]: + """ + Fetch API key usage statistics from Brickset. + + Returns: + dict: Usage stats including dateStamp and apiCalls + """ + url = "https://brickset.com/api/v3.asmx/getKeyUsageStats" + params = {"apiKey": self.api_key} + + async with aiohttp.ClientSession() as session: + async with session.get( + url, params=params, headers=self.headers + ) as response: + if response.status == 200: + data = await response.read() + result = json.loads(data.decode("utf-8")) + + if self.debug: + print(f"[@DEBUG] API usage stats: {result}") + if self.logger: + self.logger.debug(f"API usage stats: {result}") + + return result + else: + error_msg = f"Failed to get usage stats: HTTP {response.status}" + if self.logger: + self.logger.error(error_msg) + raise Exception(error_msg) + + async def check_and_update_usage(self) -> bool: + """ + Check current API usage and update internal counter. + Automatically called before each request if needed. + + Returns: + bool: True if under limit, False if at/over limit + """ + now = datetime.now() + + # Check usage if we haven't checked yet or if it's been over an hour + should_check = ( + self.daily_requests_used is None + or self.usage_last_checked is None + or (now - self.usage_last_checked) > self.usage_check_interval + ) + + if should_check: + try: + stats = await self.get_key_usage_stats() + # Brickset returns {"status": "success", "apiKeyUsage": [...]} + usage_array = stats.get("apiKeyUsage", []) + if usage_array and len(usage_array) > 0: + today_stats = usage_array[0] # Most recent day + self.daily_requests_used = today_stats.get("count", 0) + self.usage_last_checked = now + + if self.logger: + self.logger.info( + f"API usage: {self.daily_requests_used}/{self.daily_requests_limit}" + ) + + if self.debug: + print( + f"[@DEBUG] API usage: {self.daily_requests_used}/{self.daily_requests_limit}" + ) + except Exception as e: + if self.logger: + self.logger.warning(f"Failed to check API usage: {e}") + # Don't block requests if usage check fails + return True + + # Check if we're over the limit + if self.daily_requests_used is not None: + if self.daily_requests_used >= self.daily_requests_limit: + error_msg = f"Daily API limit reached: {self.daily_requests_used}/{self.daily_requests_limit}" + if self.logger: + self.logger.error(error_msg) + if self.debug: + print(f"[@ERROR] {error_msg}") + return False + + # Warn when approaching limit + if self.daily_requests_used >= self.daily_requests_limit * 0.9: + warning_msg = f"Approaching daily API limit: {self.daily_requests_used}/{self.daily_requests_limit}" + if self.logger: + self.logger.warning(warning_msg) + if self.debug: + print(f"[@WARNING] {warning_msg}") + + return True + async def login(self) -> Dict[str, Any]: """Asynchronously authenticate with Brickset API""" url = "https://brickset.com/api/v3.asmx/login" @@ -64,7 +163,17 @@ class AsyncBricksetAPI: Returns: Dict containing set information + + Raises: + Exception: If daily API limit exceeded or API error """ + # Check API usage before making request + if not await self.check_and_update_usage(): + raise Exception( + f"Brickset API daily limit reached ({self.daily_requests_limit} requests/day). " + "Please try again tomorrow." + ) + if not self.user_hash: await self.login() diff --git a/src/abrickowlapi.py b/src/old/abrickowlapi.py similarity index 97% rename from src/abrickowlapi.py rename to src/old/abrickowlapi.py index 49ca97d..562db7a 100644 --- a/src/abrickowlapi.py +++ b/src/old/abrickowlapi.py @@ -1,120 +1,120 @@ -import aiohttp -import json -import dotenv -import os -import importlib.metadata - -dotenv.load_dotenv() - -api_key = os.getenv("BRICKOWL") -abrickowl_version = importlib.metadata.version("abrickowlapi") - - -class ABrickowlAPI: - - def __init__(self, value): - self.user_agent = f"abrickowlapi/{abrickowl_version}" - self.headers = {"content-type": "application/x-www-form-urlencoded"} - self.base_url = "https://api.brickowl.com/v1/" - self.session = aiohttp.ClientSession() - return self - - def set_valid(self, set_id: int): - """ - Verify if a LEGO set ID is valid. - """ - if len(set_id) == 5: - return True - else: - return False - - @classmethod - async def catalog_list(self): - """ - Fetch the list of LEGO sets from the Brickowl API. - """ - async with self.session.get( - f"{self.base_url}catalog/list/view?key={api_key}", - headers=self.headers, - user_agent=self.user_agent, - ) as response: - if response.status == 200: - return await response.json() - else: - raise Exception(f"Error fetching list, {response.status}") - - @classmethod - async def catalog_lookup(self, set_id: int): - """ - Fetch details from the brickowl API for a specific LEGO set. - - set_id: The LEGO set ID to look up. - - returns JSON response containing set details. - """ - async with self.session.get( - f"{self.base_url}catalog/lookup/{set_id}view?key={api_key}", - headers=self.headers, - user_agent=self.user_agent, - ) as response: - if response.status == 200: - return await response.json() - else: - raise Exception(f"Error fetching set {set_id}, {response.status}") - - @classmethod - async def catalog_availability(self, set_id: int): - """ - Fetch availability information for a specific LEGO set. - """ - async with self.session.get( - f"{self.base_url}catalog/availability/{set_id}view?key={api_key}", - headers=self.headers, - user_agent=self.user_agent, - ) as response: - if response.status == 200: - return await response.json() - else: - raise Exception( - f"Error fetching fetching availability for {set_id}, {response.status}" - ) - - @classmethod - async def catalog_id_lookup(self, set_id: int): - async with self.session.get( - f"{self.base_url}catalog/id_lookup/{set_id}view?key={api_key}", - headers=self.headers, - user_agent=self.user_agent, - ) as response: - if response.status == 200: - return await response.json() - else: - raise Exception(f"ID lookup failed for set {set_id}, {response.status}") - - @classmethod - async def catalog_bulk(self): - async with self.session.get( - f"{self.base_url}catalog/bulk/view?key={api_key}", - headers=self.headers, - user_agent=self.user_agent, - ) as response: - if response.status == 200: - return await response.json() - else: - raise Exception(f"Error fetching bulk, {response.status}") - - @classmethod - async def catalog_search(self, query: str): - async with self.session.get( - f"{self.base_url}catalog/search/{query}view?key={api_key}", - headers=self.headers, - user_agent=self.user_agent, - ) as response: - if response.status == 200: - return await response.json() - else: - raise Exception(f"Search term {set_id} failed, {response.status}") - - -# Usage: -# obj = await ABrickowlAPI.catalog_lookup(value) +import aiohttp +import json +import dotenv +import os +import importlib.metadata + +dotenv.load_dotenv() + +api_key = os.getenv("BRICKOWL") +abrickowl_version = importlib.metadata.version("abrickowlapi") + + +class ABrickowlAPI: + + def __init__(self, value): + self.user_agent = f"abrickowlapi/{abrickowl_version}" + self.headers = {"content-type": "application/x-www-form-urlencoded"} + self.base_url = "https://api.brickowl.com/v1/" + self.session = aiohttp.ClientSession() + return self + + def set_valid(self, set_id: int): + """ + Verify if a LEGO set ID is valid. + """ + if len(set_id) == 5: + return True + else: + return False + + @classmethod + async def catalog_list(self): + """ + Fetch the list of LEGO sets from the Brickowl API. + """ + async with self.session.get( + f"{self.base_url}catalog/list/view?key={api_key}", + headers=self.headers, + user_agent=self.user_agent, + ) as response: + if response.status == 200: + return await response.json() + else: + raise Exception(f"Error fetching list, {response.status}") + + @classmethod + async def catalog_lookup(self, set_id: int): + """ + Fetch details from the brickowl API for a specific LEGO set. + + set_id: The LEGO set ID to look up. + + returns JSON response containing set details. + """ + async with self.session.get( + f"{self.base_url}catalog/lookup/{set_id}view?key={api_key}", + headers=self.headers, + user_agent=self.user_agent, + ) as response: + if response.status == 200: + return await response.json() + else: + raise Exception(f"Error fetching set {set_id}, {response.status}") + + @classmethod + async def catalog_availability(self, set_id: int): + """ + Fetch availability information for a specific LEGO set. + """ + async with self.session.get( + f"{self.base_url}catalog/availability/{set_id}view?key={api_key}", + headers=self.headers, + user_agent=self.user_agent, + ) as response: + if response.status == 200: + return await response.json() + else: + raise Exception( + f"Error fetching fetching availability for {set_id}, {response.status}" + ) + + @classmethod + async def catalog_id_lookup(self, set_id: int): + async with self.session.get( + f"{self.base_url}catalog/id_lookup/{set_id}view?key={api_key}", + headers=self.headers, + user_agent=self.user_agent, + ) as response: + if response.status == 200: + return await response.json() + else: + raise Exception(f"ID lookup failed for set {set_id}, {response.status}") + + @classmethod + async def catalog_bulk(self): + async with self.session.get( + f"{self.base_url}catalog/bulk/view?key={api_key}", + headers=self.headers, + user_agent=self.user_agent, + ) as response: + if response.status == 200: + return await response.json() + else: + raise Exception(f"Error fetching bulk, {response.status}") + + @classmethod + async def catalog_search(self, query: str): + async with self.session.get( + f"{self.base_url}catalog/search/{query}view?key={api_key}", + headers=self.headers, + user_agent=self.user_agent, + ) as response: + if response.status == 200: + return await response.json() + else: + raise Exception(f"Search term {set_id} failed, {response.status}") + + +# Usage: +# obj = await ABrickowlAPI.catalog_lookup(value) diff --git a/src/asquareclass.py b/src/old/asquareclass.py similarity index 100% rename from src/asquareclass.py rename to src/old/asquareclass.py diff --git a/src/scheduler.py b/src/old/scheduler.py similarity index 100% rename from src/scheduler.py rename to src/old/scheduler.py diff --git a/src/prod/abricksetapi_clean.py b/src/prod/abricksetapi_clean.py new file mode 100644 index 0000000..e0e7283 --- /dev/null +++ b/src/prod/abricksetapi_clean.py @@ -0,0 +1,277 @@ +""" +Clean refactored version of abricksetapi.py without debug logging. + +This module provides async functions for interacting with Brickset API +to retrieve LEGO set information. +""" + +import os +import aiohttp +import json +import dotenv +from typing import Dict, Any, Optional +from datetime import datetime, timedelta + +from . import throttler + +dotenv.load_dotenv() + + +class AsyncBricksetAPI: + """Asynchronous client for the Brickset API.""" + + def __init__(self, logger=None): + """ + Initialize Brickset API client. + + Args: + logger: Optional logger instance for structured logging + """ + self.username = os.getenv("BRICKSET_USERNAME") + self.password = os.getenv("BRICKSET_PASS") + self.api_key = os.getenv("BRICKSET_API_KEY") + self.headers = {"UserAgent": "studspider/0.1"} + self.throttler = throttler.GlobalThrottler(rate_limit=500, period=60) + self.user_hash: Optional[str] = None + self.logger = logger + + # API usage tracking (1000 requests per day limit) + self.daily_requests_used: Optional[int] = None + self.daily_requests_limit = 1000 + self.usage_last_checked: Optional[datetime] = None + self.usage_check_interval = timedelta(hours=1) + + async def login(self) -> Dict[str, Any]: + """ + Authenticate with Brickset API. + + Returns: + dict: Authentication response containing user hash + + Raises: + Exception: If authentication fails + """ + url = "https://brickset.com/api/v3.asmx/login" + params = { + "username": self.username, + "password": self.password, + "apiKey": self.api_key, + } + + async with aiohttp.ClientSession() as session: + async with session.get( + url, params=params, headers=self.headers + ) as response: + 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_key_usage_stats(self) -> Dict[str, Any]: + """ + Fetch API key usage statistics from Brickset. + + Returns: + dict: Usage stats with apiKeyUsage array containing dateStamp and count + """ + url = "https://brickset.com/api/v3.asmx/getKeyUsageStats" + params = {"apiKey": self.api_key} + + async with aiohttp.ClientSession() as session: + async with session.get( + url, params=params, headers=self.headers + ) as response: + if response.status == 200: + data = await response.read() + return json.loads(data.decode("utf-8")) + else: + raise Exception( + f"Failed to get usage stats: HTTP {response.status}" + ) + + async def check_and_update_usage(self) -> bool: + """ + Check current API usage and update internal counter. + Automatically called before each request if needed. + + Returns: + bool: True if under limit, False if at/over limit + """ + now = datetime.now() + + # Check usage if we haven't checked yet or if it's been over an hour + should_check = ( + self.daily_requests_used is None + or self.usage_last_checked is None + or (now - self.usage_last_checked) > self.usage_check_interval + ) + + if should_check: + try: + stats = await self.get_key_usage_stats() + usage_array = stats.get("apiKeyUsage", []) + if usage_array and len(usage_array) > 0: + today_stats = usage_array[0] # Most recent day + self.daily_requests_used = today_stats.get("count", 0) + self.usage_last_checked = now + + if self.logger: + self.logger.info( + f"API usage: {self.daily_requests_used}/{self.daily_requests_limit}" + ) + except Exception as e: + if self.logger: + self.logger.warning(f"Failed to check API usage: {e}") + return True # Don't block requests if usage check fails + + # Check if we're over the limit + if self.daily_requests_used is not None: + if self.daily_requests_used >= self.daily_requests_limit: + if self.logger: + self.logger.error( + f"Daily API limit reached: {self.daily_requests_used}/{self.daily_requests_limit}" + ) + return False + + # Warn when approaching limit (90%) + if self.daily_requests_used >= self.daily_requests_limit * 0.9: + if self.logger: + self.logger.warning( + f"Approaching daily API limit: {self.daily_requests_used}/{self.daily_requests_limit}" + ) + + return True + + async def get_set(self, set_number: int) -> Dict[str, Any]: + """ + Fetch LEGO set information from Brickset API. + + Args: + set_number: LEGO set number (e.g., 10338) + + Returns: + dict: Set information including: + - set_num: LEGO set number + - set_name: Official set name + - set_theme: Theme (e.g., "Icons") + - set_image: Thumbnail URL + - set_barcode: UPC or EAN barcode + - set_retailprice: US retail price + - set_adjustredprice: Retail price + 10% markup + - pw_name: Formatted name for Square (e.g., "10338 - ICONS - Bumblebee") + - requires_manual_price: True if price >= $110 + + Raises: + ValueError: If set not found + Exception: For API errors or if daily limit exceeded + """ + # Check API usage before making request + if not await self.check_and_update_usage(): + raise Exception( + f"Brickset API daily limit reached ({self.daily_requests_limit} requests/day). " + "Please try again tomorrow." + ) + + if not self.user_hash: + await self.login() + + url = "https://brickset.com/api/v3.asmx/getSets" + params = { + "apiKey": self.api_key, + "userHash": self.user_hash, + "params": json.dumps({"setNumber": f"{set_number}-1"}), + } + + async with aiohttp.ClientSession() as session: + async with session.get( + url, params=params, headers=self.headers + ) as response: + 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"): + 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] + + # Get UPC barcode, fall back to EAN if not available + barcode_data = set_data.get("barcode", {}) + upc_code = barcode_data.get("UPC") or barcode_data.get("EAN") + + # Extract set information + our_set = { + "set_num": set_data["number"], + "set_name": set_data["name"], + "set_theme": set_data["theme"], + "set_image": set_data["image"]["thumbnailURL"], + "set_barcode": upc_code, + "set_retailprice": round( + set_data["LEGOCom"]["US"]["retailPrice"] + ), + } + + # Calculate adjusted price (10% markup) + retail_price = our_set["set_retailprice"] + our_set["set_adjustredprice"] = retail_price + ( + (10 / 100) * retail_price + ) + + # Format name for Square inventory + our_set["pw_name"] = ( + f"{our_set['set_num']} - {our_set['set_theme'].upper()} - {our_set['set_name']}" + ) + + # Flag for manual price review if needed + if our_set["set_retailprice"] >= 110: + our_set["requires_manual_price"] = True + + return our_set + + elif response.status == 500: + error_msg = ( + "Internal Server Error: Please check API key and parameters." + ) + if self.logger: + self.logger.error(error_msg) + raise Exception(error_msg) + else: + 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: int) -> Dict[str, Any]: + """ + Get set information with rate limiting (500 requests per 60 seconds). + + Args: + set_number: LEGO set number + + Returns: + dict: Set information + + Raises: + ValueError: If set not found + Exception: For API errors + """ + async with self.throttler: + return await self.get_set(set_number) diff --git a/src/prod/asquare_clean.py b/src/prod/asquare_clean.py new file mode 100644 index 0000000..89dfcf9 --- /dev/null +++ b/src/prod/asquare_clean.py @@ -0,0 +1,426 @@ +""" +This module provides async functions for interacting with Square Catalog API +to manage LEGO inventory items. +""" + +import json +import asyncio +import uuid +import re +import aiohttp + +from square.core.api_error import ApiError +from square import AsyncSquare + +from . import get_sq_env + + +def get_set_number(text: str) -> int: + """Extract 5-digit LEGO set number from text.""" + numbers = re.findall(r"\d{5}", text) + if not numbers: + raise ValueError("No five-digit number found in string") + return int(numbers[0]) + + +# Configuration +SQUARE_TOKEN = get_sq_env.SQUARE_TOKEN +LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID +ENVIRONMENT = get_sq_env.environment + +# Initialize Square client +client = AsyncSquare(token=SQUARE_TOKEN, environment=ENVIRONMENT) + + +async def get_itm(sku: str) -> dict: + """ + Retrieve item details from Square Catalog using a SKU. + + Args: + sku: The SKU or text filter to search for + + Returns: + dict: Item name and ID + + Raises: + IndexError: If no items found + Exception: For API errors + """ + 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, version + + Raises: + ValueError: If item not found + Exception: For API errors + """ + 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"], + "item_data": item_data, + "updated_at": obj.get("updated_at"), + "version": obj.get("version"), + } + + +async def get_item_by_variation_id(variation_id: str) -> dict: + """ + Retrieve parent item details from a variation ID. + + Webhooks send variation IDs, but we need the parent item. + Searches all items in LEGO category to find the matching variation. + + Args: + variation_id: Square catalog item variation ID + + Returns: + dict: Parent item details + + Raises: + ValueError: If variation/item not found + Exception: For API errors + """ + cursor = None + page_count = 0 + + while True: + page_count += 1 + + 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("No items found in LEGO category") + break + + # Search items on this page for matching 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 parent item + category_id = item_data.get("category_id") or LEGO_CAT_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"), + } + + cursor = result_json.get("cursor") + if not cursor: + break + + raise ValueError( + f"Variation {variation_id} not found in LEGO category (searched {page_count} pages)" + ) + + +async def upload_item_image(item_id: str, image_url: str) -> bool: + """ + Download image from URL and upload to Square catalog item. + + Args: + item_id: Square catalog item ID + image_url: URL to download image from (e.g., Brickset thumbnail) + + Returns: + bool: True if successful + """ + try: + # Download image + async with aiohttp.ClientSession() as session: + async with session.get(image_url) as resp: + if resp.status != 200: + return False + image_data = await resp.read() + + # Determine content type + content_type = "image/jpeg" + if image_url.lower().endswith(".png"): + content_type = "image/png" + elif image_url.lower().endswith(".gif"): + content_type = "image/gif" + + # Upload to Square + result = await client.catalog.images.create( + image_file=image_data, + request={ + "idempotency_key": str(uuid.uuid4()), + "image": { + "type": "IMAGE", + "id": f"#IMAGE_{item_id}", + "image_data": {"name": "LEGO Set Image"}, + }, + "object_id": item_id, + }, + image_file_name=f"lego_{item_id}.jpg", + content_type=content_type, + ) + + result_json = json.loads(result.model_dump_json()) + return result_json.get("image") is not None + + except Exception: + return False + + +async def update_itm(item_id: str, lego_set: dict, item_data: dict = None) -> bool: + """ + Update Square catalog item with Brickset LEGO set data. + + Updates name, description, price, SKU, UPC, and category while preserving + all other item fields (location settings, tax, etc). + + Args: + item_id: Square catalog item ID to update + lego_set: Dict containing LEGO set details from Brickset: + - 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 (e.g., "75192 - STAR WARS - Millennium Falcon") + - set_adjustredprice: Retail price with 10% markup + item_data: Optional pre-fetched item data with category_id + + Returns: + bool: True if successful + + Raises: + KeyError: If expected keys missing from lego_set + Exception: For API errors + """ + try: + # Extract LEGO set data + 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"] + + price_in_cents = int(set_adjustedprice * 100) + + # Retry logic for version conflicts + max_retries = 3 + retry_delay = 0.5 + + for attempt in range(max_retries): + try: + # Fetch fresh item data for latest version + fresh_item = await get_item_by_id(item_id) + except ValueError: + # Item was deleted + return False + + variations = fresh_item.get("item_data", {}).get("variations", []) + current_version = fresh_item.get("version") + + # Get variation details + if variations: + variation_id = variations[0]["id"] + variation_version = variations[0].get("version") + else: + variation_id = None + variation_version = None + + # Determine category ID + # batch_get doesn't return category_id, so use from item_data or fallback + if item_data and item_data.get("category_id"): + category_id = item_data.get("category_id") + else: + category_id = LEGO_CAT_ID + + # Preserve all existing item_data and update only what we need + existing_item_data = fresh_item.get("item_data", {}) + existing_item_data["name"] = pw_name + existing_item_data["description"] = f"{set_name} - {set_theme}" + existing_item_data["category_id"] = category_id + + # Update only the specific variation + existing_variations = existing_item_data.get("variations", []) + updated_variations = [] + variation_found = False + + for existing_var in existing_variations: + if existing_var.get("id") == variation_id: + # Preserve all variation fields, update only specific ones + updated_var = existing_var.copy() + updated_var_data = existing_var.get( + "item_variation_data", {} + ).copy() + + updated_var_data["price_money"] = { + "amount": price_in_cents, + "currency": "USD", + } + updated_var_data["sku"] = set_number + updated_var_data["upc"] = set_upc if set_upc else None + + updated_var["item_variation_data"] = updated_var_data + + if variation_version is not None: + updated_var["version"] = variation_version + + updated_variations.append(updated_var) + variation_found = True + else: + updated_variations.append(existing_var) + + # Create new variation if not found + if not variation_found: + new_variation = { + "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_number, + "upc": set_upc if set_upc else None, + }, + } + if variation_version is not None: + new_variation["version"] = variation_version + updated_variations.append(new_variation) + + existing_item_data["variations"] = updated_variations + + # Build upsert request + upsert_body = { + "idempotency_key": str(uuid.uuid4()), + "object": { + "type": "ITEM", + "id": item_id, + "version": current_version, + "item_data": existing_item_data, + }, + } + + # Perform 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"): + # Upload image if available (currently disabled) + # if set_image: + # await upload_item_image(item_id, set_image) + return True + else: + return False + + except ApiError as e: + # Handle version conflicts with retry + 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: + await asyncio.sleep(retry_delay) + retry_delay *= 2 + continue + else: + return False + raise + + except Exception: + return False + + +async def acat2id(cat_id: str) -> list[dict]: + """ + List all items in a category. + + Args: + cat_id: Category ID to search + + Returns: + list[dict]: List of items with id, name, set number, updated_at, version + """ + 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 = [] + + for obj in result_json.get("objects", []): + item_name = obj.get("item_data", {}).get("name", "") + set_number = None + + if re.findall(r"\d{5}", item_name): + try: + set_number = get_set_number(item_name) + except ValueError: + pass + + items.append( + { + "id": obj["id"], + "name": item_name or "Unknown", + "number": set_number, + "updated_at": obj.get("updated_at"), + "version": obj.get("version"), + } + ) + + return items diff --git a/src/list_categories.py b/src/scripts/list_categories.py similarity index 98% rename from src/list_categories.py rename to src/scripts/list_categories.py index 22e25e1..79c244c 100644 --- a/src/list_categories.py +++ b/src/scripts/list_categories.py @@ -4,8 +4,8 @@ List all Square categories with their IDs to help identify the correct LEGO cate import asyncio import json -from . import asquare -from . import get_sq_env +from .. import asquare +from .. import get_sq_env async def list_categories(): diff --git a/src/scripts/migrate_sku_to_gtin.py b/src/scripts/migrate_sku_to_gtin.py index 6179e12..06ed78f 100644 --- a/src/scripts/migrate_sku_to_gtin.py +++ b/src/scripts/migrate_sku_to_gtin.py @@ -95,30 +95,17 @@ async def migrate_sku_to_gtin(): print(f"[@INFO] Item {item_id} ({item_name})") print(f"[@INFO] Variation {var_id}: Moving SKU '{sku}' to UPC") - # Build updated variation - updated_var = { - "type": "ITEM_VARIATION", - "id": var_id, - "version": var_version, - "item_variation_data": { - "item_id": item_id, - "name": var_data.get("name", "Regular"), - "pricing_type": var_data.get( - "pricing_type", "FIXED_PRICING" - ), - "upc": sku, # Move SKU value to UPC - "sku": ( - set_number if set_number else None - ), # Set to LEGO set number if available - }, - } + # Preserve all existing variation data and only modify what we need + updated_var = var.copy() # Shallow copy the entire variation + updated_var_data = var_data.copy() # Copy the variation_data - # Preserve pricing info - if "price_money" in var_data: - updated_var["item_variation_data"]["price_money"] = var_data[ - "price_money" - ] + # Update only UPC and SKU fields + updated_var_data["upc"] = sku # Move SKU value to UPC + updated_var_data["sku"] = ( + set_number if set_number else None + ) # Set to LEGO set number + updated_var["item_variation_data"] = updated_var_data updated_variations.append(updated_var) needs_update = True else: diff --git a/src/bricksetapi.py b/src/scripts/syncro/bricksetapi.py similarity index 96% rename from src/bricksetapi.py rename to src/scripts/syncro/bricksetapi.py index 42fb7c2..f06b9d2 100644 --- a/src/bricksetapi.py +++ b/src/scripts/syncro/bricksetapi.py @@ -1,70 +1,70 @@ -import os -import requests -import json -import dotenv - -dotenv.load_dotenv() - -USERNAME = os.getenv("BRICKSET_USERNAME") -PASS = os.getenv("BRICKSET_PASS") -BRICKSET_API_KEY = os.getenv("BRICKSET_API_KEY") - -debug = False # Set to True to enable debug output - - -def login(username, password): - url = "https://brickset.com/api/v3.asmx/login" - headers = {"UserAgent": "studspider/0.1"} - params = { - "UserAgent": "studspider/0.1", - "username": username, - "password": password, - "apiKey": BRICKSET_API_KEY, - } - response = requests.get(url, params=params, headers=headers) - if debug: - print(f"[@DEBUG] Login response: {response.status_code} - {response.text}\n") - return response.json() - - -def get_set(set_number: int) -> dict: - login_response = login(USERNAME, PASS) - url = f"https://brickset.com/api/v3.asmx/getSets?apiKey={BRICKSET_API_KEY}&userHash={login_response["hash"]}¶ms={{'setNumber' : '{set_number}-1'}}" - response = requests.get(url) - if response.status_code == 200: - data = response.json() - if not data["sets"] and data["message"] == "API limit exceeded": - raise Exception("API limit exceeded. Please try again later.") - return False - - print(f"[@DEBUG]\t{data}") - - # Try to get UPC barcode, fall back to EAN if not available - barcode_data = data["sets"][0].get("barcode", {}) - upc_code = barcode_data.get("UPC") or barcode_data.get("EAN") - - our_set = { - "set_num": data["sets"][0]["number"], - "set_name": data["sets"][0]["name"], - "set_theme": data["sets"][0]["theme"], - "set_image": data["sets"][0]["image"]["thumbnailURL"], - "set_barcode": upc_code, - "set_retailprice": round(data["sets"][0]["LEGOCom"]["US"]["retailPrice"]), - } - # add PW 10% b4 tax - our_set.update( - {"set_adjustredprice": our_set + ((10 / 100) * our_set["set_retailprice"])} - ) - our_set["pw_name"] = ( - f"{our_set['set_num']} - {our_set["set_theme"].upper()} {our_set['set_name']}" - ) - if our_set["set_retailprice"] >= 110: - our_set["requires_manual_price"] = True - return our_set - - elif response.status_code == 500: - raise Exception( - "Internal Server Error: Please check the API key and parameters." - ) - else: - raise Exception("Unexpected response from the API.") +import os +import requests +import json +import dotenv + +dotenv.load_dotenv() + +USERNAME = os.getenv("BRICKSET_USERNAME") +PASS = os.getenv("BRICKSET_PASS") +BRICKSET_API_KEY = os.getenv("BRICKSET_API_KEY") + +debug = False # Set to True to enable debug output + + +def login(username, password): + url = "https://brickset.com/api/v3.asmx/login" + headers = {"UserAgent": "studspider/0.1"} + params = { + "UserAgent": "studspider/0.1", + "username": username, + "password": password, + "apiKey": BRICKSET_API_KEY, + } + response = requests.get(url, params=params, headers=headers) + if debug: + print(f"[@DEBUG] Login response: {response.status_code} - {response.text}\n") + return response.json() + + +def get_set(set_number: int) -> dict: + login_response = login(USERNAME, PASS) + url = f"https://brickset.com/api/v3.asmx/getSets?apiKey={BRICKSET_API_KEY}&userHash={login_response["hash"]}¶ms={{'setNumber' : '{set_number}-1'}}" + response = requests.get(url) + if response.status_code == 200: + data = response.json() + if not data["sets"] and data["message"] == "API limit exceeded": + raise Exception("API limit exceeded. Please try again later.") + return False + + print(f"[@DEBUG]\t{data}") + + # Try to get UPC barcode, fall back to EAN if not available + barcode_data = data["sets"][0].get("barcode", {}) + upc_code = barcode_data.get("UPC") or barcode_data.get("EAN") + + our_set = { + "set_num": data["sets"][0]["number"], + "set_name": data["sets"][0]["name"], + "set_theme": data["sets"][0]["theme"], + "set_image": data["sets"][0]["image"]["thumbnailURL"], + "set_barcode": upc_code, + "set_retailprice": round(data["sets"][0]["LEGOCom"]["US"]["retailPrice"]), + } + # add PW 10% b4 tax + our_set.update( + {"set_adjustredprice": our_set + ((10 / 100) * our_set["set_retailprice"])} + ) + our_set["pw_name"] = ( + f"{our_set['set_num']} - {our_set["set_theme"].upper()} - {our_set['set_name']}" + ) + if our_set["set_retailprice"] >= 110: + our_set["requires_manual_price"] = True + return our_set + + elif response.status_code == 500: + raise Exception( + "Internal Server Error: Please check the API key and parameters." + ) + else: + raise Exception("Unexpected response from the API.") diff --git a/src/libsquare.py b/src/scripts/syncro/libsquare.py similarity index 96% rename from src/libsquare.py rename to src/scripts/syncro/libsquare.py index e87bab6..2d89a05 100644 --- a/src/libsquare.py +++ b/src/scripts/syncro/libsquare.py @@ -1,66 +1,66 @@ -import json - -import get_sq_env - -SQUARE_TOKEN = get_sq_env.SQUARE_TOKEN -LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID -ENVIRONMENT = get_sq_env.environment - -from square import Square -from square.environment import SquareEnvironment - - -def get_square_client(): - """Initialize and return a Square client.""" - if not SQUARE_TOKEN: - raise ValueError("SQUARE_TOKEN environment variable is not set.") - square = Square(token=SQUARE_TOKEN, environment=ENVIRONMENT) # type: ignore - return square - - -client = get_square_client() - - -def get_itm(item_id: str): - """Fetch an item from Square by its ID.""" - result = client.catalog.search_items(text_filter=item_id, limit=1) - result_json = json.loads(result.model_dump_json()) - ids = { - "item_id": result_json["items"][0]["id"], - "vari_id": result_json["items"][0]["item_data"]["variations"][0][ - "item_variation_data" - ]["item_id"], - } - item = { - "set_name": result_json["items"][0]["item_data"]["name"], - "set_price": result_json["items"][0]["item_data"]["variations"][0][ - "item_variation_data" - ]["price_money"], - "set_barcode": result_json["items"][0]["item_data"]["variations"][0][ - "item_variation_data" - ]["sku"], - } - return [ids, item] - - -def cat2id(cat_id: str): - """Return a list of item IDs within the specified category ID.""" - result = 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()) - item_ids = [obj["id"] for obj in result_json.get("objects", [])] - return item_ids - - -def test() -> None: - # all_lego_sets = cat2id(SQUARE_LEGO_CAT_ID) # type: ignore - # result = get_itm("71433") - # print(result) - pass +import json + +import get_sq_env + +SQUARE_TOKEN = get_sq_env.SQUARE_TOKEN +LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID +ENVIRONMENT = get_sq_env.environment + +from square import Square +from square.environment import SquareEnvironment + + +def get_square_client(): + """Initialize and return a Square client.""" + if not SQUARE_TOKEN: + raise ValueError("SQUARE_TOKEN environment variable is not set.") + square = Square(token=SQUARE_TOKEN, environment=ENVIRONMENT) # type: ignore + return square + + +client = get_square_client() + + +def get_itm(item_id: str): + """Fetch an item from Square by its ID.""" + result = client.catalog.search_items(text_filter=item_id, limit=1) + result_json = json.loads(result.model_dump_json()) + ids = { + "item_id": result_json["items"][0]["id"], + "vari_id": result_json["items"][0]["item_data"]["variations"][0][ + "item_variation_data" + ]["item_id"], + } + item = { + "set_name": result_json["items"][0]["item_data"]["name"], + "set_price": result_json["items"][0]["item_data"]["variations"][0][ + "item_variation_data" + ]["price_money"], + "set_barcode": result_json["items"][0]["item_data"]["variations"][0][ + "item_variation_data" + ]["sku"], + } + return [ids, item] + + +def cat2id(cat_id: str): + """Return a list of item IDs within the specified category ID.""" + result = 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()) + item_ids = [obj["id"] for obj in result_json.get("objects", [])] + return item_ids + + +def test() -> None: + # all_lego_sets = cat2id(SQUARE_LEGO_CAT_ID) # type: ignore + # result = get_itm("71433") + # print(result) + pass diff --git a/tests/test.py b/tests/test.py index d4d7c87..33bee63 100644 --- a/tests/test.py +++ b/tests/test.py @@ -1,6 +1,6 @@ import src.libsquare as sq import src.bricksetapi as lego -oursq = sq.get_itm("71417") -ourlego = lego.get_set(71417) +oursq = sq.get_itm("10291") +ourlego = lego.get_set(10291) print(f"{oursq}\n\n{ourlego}\n\n")