68 Commits
Author SHA1 Message Date
C dcccdcd9ec fix pricing not being entered 2026-01-07 11:36:24 -08:00
C dbc1115249 fix hang 2026-01-07 11:25:45 -08:00
C dbe5cd6146 final commit before first prod test 2026-01-07 09:59:36 -08:00
C a6e2397046 fix prod hang 2026-01-05 16:41:25 -08:00
C 2da61539cc fix worker warning 2026-01-05 16:32:37 -08:00
C bf65965b26 fix banner 2026-01-05 16:26:19 -08:00
C bcf625d2b0 add stats, versioning 2026-01-05 16:23:44 -08:00
C dfd2c2f07d add caching, add to stats 2026-01-05 16:01:36 -08:00
C ab3297acfb fix env thing 2026-01-05 15:44:04 -08:00
C 90aca6cca4 hopefully fix workers not kloading the key 2026-01-05 15:42:54 -08:00
C 199995b947 fix key please 2026-01-05 15:37:50 -08:00
C 2f7769e2c3 imporerror 2026-01-05 15:27:34 -08:00
C 7ca181bdd6 Merge remote-tracking branch 'origin/rewrite' into rewrite 2026-01-05 15:24:55 -08:00
C 1e96dead07 fix imports and price 2026-01-05 15:24:50 -08:00
C 248950adb9 fix imports 2026-01-05 15:23:33 -08:00
C 832122e63c refactor and reorganize!! 2026-01-04 17:55:43 -08:00
C e7f10c4f59 stats yay 2026-01-04 16:00:25 -08:00
C 2de0843344 branding + image stuff if we can 2026-01-04 15:22:42 -08:00
C daf6f215fe ean vs barcode 2026-01-04 14:53:53 -08:00
C 7fbf415acf dfgds 2026-01-04 14:31:14 -08:00
C 80542de333 d 2026-01-04 14:29:05 -08:00
C a11aba2564 we messed up lole 2026-01-04 14:22:40 -08:00
C b40526c494 s 2026-01-04 14:12:30 -08:00
C 803a4412c1 dsfadsfsa 2026-01-04 14:11:21 -08:00
C 3dfd6a27b9 bbbbb
b
2026-01-04 14:10:21 -08:00
C ce32abf58c d 2026-01-04 14:09:31 -08:00
C 94af7d9d89 derp 2026-01-04 14:08:19 -08:00
C f652b108f4 whoops 2026-01-04 14:08:10 -08:00
C 4a88f1ec53 only one var 2026-01-04 14:03:43 -08:00
C 87ed932f95 square is so fucking annoying dude 2026-01-04 13:57:11 -08:00
C fce743d580 vfrs 2026-01-04 13:53:07 -08:00
C 1352c7bb81 vfd 2026-01-04 13:49:50 -08:00
C 7fe6ac2331 were going in circles again 2026-01-04 13:45:43 -08:00
C 931341751d omg bro 2026-01-04 13:43:09 -08:00
C 48a8a5ce54 man 2026-01-04 13:38:48 -08:00
C 7296927cea dfds 2026-01-04 13:36:32 -08:00
C a2746737ce NO LOCATION AAAAH 2026-01-04 13:34:51 -08:00
C 0f68b09a00 bfdafdsa 2026-01-04 13:31:08 -08:00
C 58618429db cmon man 2026-01-04 13:29:22 -08:00
C b988db11ec location location location! 2026-01-04 13:26:45 -08:00
C bf7ebdab05 bro all locations what 2026-01-04 13:25:48 -08:00
C d7ed797cda what the helly 2026-01-04 13:20:07 -08:00
C 7954b0b70d fix cat issue 2026-01-04 13:16:31 -08:00
C 97174b2080 mismatch 2026-01-04 12:43:39 -08:00
C 0089ce8556 fdsadsa 2026-01-04 12:41:28 -08:00
C bcb4e65585 horse race tests 2026-01-04 12:40:06 -08:00
C 734d43b817 s 2026-01-04 12:38:16 -08:00
C 53f221884a vrasdfas 2026-01-04 12:36:42 -08:00
C 5f4bbb2565 dssfgfds 2026-01-04 12:34:43 -08:00
C da382942d3 very confusing lol 2026-01-04 12:30:59 -08:00
C 16d0cd85c2 dfas 2026-01-04 12:27:18 -08:00
C e10c275cbc VERSION THESE NUTS 2026-01-04 12:23:45 -08:00
C 5cbbed1992 SQUARE THESE NUTS 2026-01-04 12:21:15 -08:00
C 8a1856448d RAAAA 2026-01-04 12:18:55 -08:00
C cff03d5eee WE ARE SO CLOSE 2026-01-04 12:17:19 -08:00
C 3e21b904a2 throttler fix 2026-01-04 12:12:09 -08:00
C 535826fce6 FIXXXXX 2026-01-04 12:02:55 -08:00
C 30b4b26c9e CATEGORY ID IS NONE?????? 2026-01-04 11:56:27 -08:00
C d07f66ca57 okay man so whats the category 2026-01-04 11:49:31 -08:00
C 424d7795bc pagination fix 2026-01-04 11:44:30 -08:00
C 6c0d15dec9 dsadgdfgdsgdsf 2026-01-04 11:41:26 -08:00
C 6d376c3653 i think were going in circles 2026-01-04 11:38:02 -08:00
C 8c11958d95 test 2026-01-04 11:36:07 -08:00
C 187bf1c111 grr 2026-01-04 11:33:48 -08:00
C 4f425bfca0 vro 2026-01-04 11:30:50 -08:00
C bb1c421df1 import errors? 2026-01-04 11:29:35 -08:00
C 8e21780fb4 whoops 2026-01-04 11:27:25 -08:00
C d341684567 test glue 2026-01-04 11:18:43 -08:00
23 changed files with 1993 additions and 112 deletions

No files matched your search

+10 -3
View File
@@ -6,17 +6,24 @@ BRICKSET_API_KEY = "your_brickset_api_key"
# Square environment - "sandbox" for testing, "production" for live store
SQUARE_ENV = "sandbox" # or "production"
SQUARE_WEBHOOK_URL = "https://yourdomain.com/hook"
# Public webhook URL that Square will POST events to
SQUARE_WEBHOOK_URL = "https://yourdomain.com/webhook"
# Sandbox credentials - test environment (fake data, safe for development)
SANDBOX_SQUARE_APP_ID = "your_sandbox_app_id"
SANDBOX_SQUARE_TOKEN = "your_sandbox_access_token"
SANDBOX_SQUARE_WEBHOOK_INVENTORY_ID = "your_sandbox_webhook_signature_key"
# Sandbox webhook subscription ID and signature key for verifying webhook requests
SANDBOX_SQUARE_WEBHOOK_SUB_ID = "your_sandbox_webhook_subscription_id"
SANDBOX_SQUARE_WEBHOOK_SIGNATURE_KEY = "your_sandbox_webhook_signature_key"
# Production credentials - live store (real inventory and sales)
PROD_SQUARE_APP_ID = "your_production_app_id"
PROD_SQUARE_TOKEN = "your_production_access_token"
PROD_SQUARE_WEBHOOK_INVENTORY_ID = "your_production_webhook_signature_key"
# Production webhook subscription ID and signature key for verifying webhook requests
PROD_SQUARE_WEBHOOK_SUB_ID = "your_production_webhook_subscription_id"
PROD_SQUARE_WEBHOOK_SIGNATURE_KEY = "your_production_webhook_signature_key"
# Category ID for LEGO items in Square catalog (find in Dashboard → Items → Categories)
SQUARE_LEGO_CAT_ID = "your_lego_category_id"
+1
View File
@@ -0,0 +1 @@
0.3.0
+37 -4
View File
@@ -1,5 +1,17 @@
import multiprocessing
import os
import git
import src.banner as banner
def get_git_commit():
"""Get first 6 characters of current git commit hash"""
try:
repo = git.Repo(search_parent_directories=True)
return repo.head.commit.hexsha[:6]
except Exception:
return "unknown"
# Server socket
bind = "127.0.0.1:7433" # nginx will proxy to this
@@ -52,6 +64,16 @@ def on_starting(server):
"""Called just before the master process is initialized."""
print("Studspider WEB starting...")
# Verify .env file exists (workers will load it via get_sq_env.py)
from pathlib import Path
env_path = Path(__file__).parent.parent / ".env"
print(f"[@DEBUG] Looking for .env at: {env_path.absolute()}")
if env_path.exists():
print(f"✓ .env file found (workers will load via get_sq_env.py)")
else:
print(f"⚠ No .env file found at {env_path.absolute()}")
def on_reload(server):
"""Called to recycle workers during a reload via SIGHUP."""
@@ -59,10 +81,21 @@ def on_reload(server):
def when_ready(server):
"""Called just after the server is started."""
print(f"STUDSPIDER HOOK v0.1 RUNNING ON {bind}")
print(f"WORKERS: {workers}")
print(f"WORKING CLASS: {worker_class}\n")
commit_hash = get_git_commit()
banner_text = banner.generate("studspider v0.3.0")
# Add commit hash to the last line of banner
banner_lines = banner_text.rstrip("\n").split("\n")
if banner_lines:
banner_lines[-1] += f" - {commit_hash}"
print("\n".join(banner_lines))
print("Built for PLAY-WELL TEKNOLOGIES by contact@studspider.sh\n\n")
print("PRODUCTION MODE\n" if os.getenv("SQUARE_ENV") == "production" else "SANDBOX MODE\n")
print(f"HOST:\t{bind}")
print(f"WORKERS:\t{workers}")
print(f"WORKER TYPE:\t{worker_class}")
print(f"{'=' * 60}\n")
def worker_int(worker):
+3 -1
View File
@@ -1,6 +1,6 @@
[project]
name = "studpider"
version = "0.1.0"
version = "0.3.0"
description = ""
authors = [{ name = "CF", email = "ph0sgene@pm.me" }]
requires-python = ">=3.13,<3.14"
@@ -21,6 +21,8 @@ dependencies = [
"fastapi>=0.128.0",
"uvicorn>=0.40.0",
"gunicorn>=23.0.0",
"async-lru>=2.0.4",
"gitpython>=3.1.0",
]
[build-system]
+1 -1
View File
@@ -2,7 +2,7 @@ from setuptools import setup
setup(
name="studspider",
version="0.1",
version="0.3.0",
description="Bricket + Square integration application",
author="CF",
author_email="chris-forst@pm.me",
+133 -15
View File
@@ -7,9 +7,13 @@ import asyncio
import os
import aiohttp
import json
import math
import dotenv
from typing import Dict, Any
import throttler
from typing import Dict, Any, Optional
from datetime import datetime, timedelta
from async_lru import alru_cache
from . import throttler
from . import get_sq_env
Throttler = throttler.GlobalThrottler
@@ -19,16 +23,115 @@ dotenv.load_dotenv()
class AsyncBricksetAPI:
"""Asynchronous client for the Brickset API"""
def __init__(self, debug: bool = False, logger=None):
def __init__(self, debug: bool = None, logger=None):
self.username = os.getenv("BRICKSET_USERNAME")
self.password = os.getenv("BRICKSET_PASS")
self.api_key = os.getenv("BRICKSET_API_KEY")
self.debug = debug
# Use DEBUG_MODE from env if debug not explicitly set
self.debug = debug if debug is not None else get_sq_env.DEBUG_MODE
self.headers = {"UserAgent": "studspider/0.1"}
self.throttler = Throttler(rate_limit=500, period=60)
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"
@@ -55,16 +158,27 @@ class AsyncBricksetAPI:
self.logger.info("Successfully authenticated with Brickset API")
return result
@alru_cache(maxsize=1000, ttl=604800) # Cache 1000 sets, 7 days TTL
async def get_set(self, set_number: int) -> Dict[str, Any]:
"""
Fetch LEGO set information asynchronously
Fetch LEGO set information asynchronously (cached for 7 days)
Args:
set_number (str): LEGO set number
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()
@@ -101,26 +215,30 @@ class AsyncBricksetAPI:
)
set_data = result["sets"][0]
# Try to 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")
# Get retail price before rounding
retail_price = set_data["LEGOCom"]["US"]["retailPrice"]
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": set_data["barcode"]["UPC"],
"set_retailprice": round(
set_data["LEGOCom"]["US"]["retailPrice"]
),
"set_barcode": upc_code,
"set_retailprice": round(retail_price),
}
# Calculate adjusted price (10% markup)
retail_price = our_set["set_retailprice"]
our_set["set_adjustredprice"] = retail_price + (
(10 / 100) * retail_price
)
# Calculate adjusted price (10% markup), then round up
adjusted_price = retail_price * 1.10
our_set["set_adjustredprice"] = math.ceil(adjusted_price)
# Set PW inventory name with theme
our_set["pw_name"] = (
f"{our_set['set_num']} - {our_set['set_theme'].upper()} {our_set['set_name']}"
f"{our_set['set_num']} - {our_set['set_theme'].upper()} - {our_set['set_name']}"
)
# Flag for manual price review if needed
+436 -7
View File
@@ -4,11 +4,15 @@ import json
import asyncio
import uuid
import re
import traceback
from time import time
import aiohttp
from io import BytesIO
import db
from square.core.api_error import ApiError
import get_sq_env
from . import db
from . import get_sq_env
def get_set_number(text: str) -> int:
@@ -51,10 +55,230 @@ async def get_itm(sku: str):
return {"name": name, "id": itm_id}
async def update_itm(lego_set: dict):
async def get_item_by_id(item_id: str) -> dict:
"""
Retrieve full item details by Square catalog item ID.
Uses batch_get API for direct ID lookup.
Args:
item_id: Square catalog item ID
Returns:
dict: Item details including name, category_id, variations, etc.
Raises:
ValueError: If item not found
Exception: If API error
"""
try:
# Use batch_get to retrieve item directly by ID
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, # Return the entire item_data object
"updated_at": obj.get("updated_at"),
"version": obj.get("version"),
}
except ValueError:
raise
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.
Since async Square API lacks direct object retrieval, we search all items
in the LEGO category and match variations.
Args:
variation_id: Square catalog item variation ID
Returns:
dict: Parent item details
Raises:
Exception: If variation/item not found or API error
"""
try:
# Search LEGO category items to find which one has this variation
print(f"[@DEBUG] Searching for variation {variation_id} in LEGO category")
cursor = None
page_count = 0
max_pages = 50 # Safety limit to prevent infinite loops
while page_count < max_pages:
page_count += 1
print(f"[@DEBUG] Searching page {page_count}...")
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(f"No items found in LEGO category")
else:
# No more results
print(f"[@DEBUG] No more results on page {page_count}")
break
# Search through items on this page to find which has this 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 the parent item!
print(
f"[@DEBUG] Found parent item: {obj['id']} on page {page_count}"
)
# Since we searched by category, we know it's in LEGO_CAT_ID
# But also try to get from item_data as fallback
category_id = item_data.get("category_id") or LEGO_CAT_ID
print(
f"[@DEBUG] Item category_id from data: {item_data.get('category_id')}"
)
print(f"[@DEBUG] Using category_id: {category_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"),
}
# Check if there are more pages
cursor = result_json.get("cursor")
if not cursor:
# No more pages
print(f"[@DEBUG] No more pages after page {page_count}")
break
# If we get here, variation wasn't found
if page_count >= max_pages:
raise ValueError(
f"Variation {variation_id} not found after searching {max_pages} pages (possible infinite loop prevented)"
)
raise ValueError(
f"Variation {variation_id} not found in any LEGO category item (searched {page_count} pages)"
)
except Exception as e:
print(f"[@ERROR] Failed to get item from variation {variation_id}: {e}")
raise
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 upload successful, False otherwise
"""
try:
print(f"[@DEBUG] Downloading image from {image_url}")
# Download image from URL
async with aiohttp.ClientSession() as session:
async with session.get(image_url) as resp:
if resp.status != 200:
print(f"[@ERROR] Failed to download image: HTTP {resp.status}")
return False
image_data = await resp.read()
# Determine content type from URL extension
content_type = "image/jpeg"
if image_url.lower().endswith(".png"):
content_type = "image/png"
elif image_url.lower().endswith(".gif"):
content_type = "image/gif"
print(
f"[@DEBUG] Uploading {len(image_data)} bytes as {content_type} to item {item_id}"
)
# 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": f"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())
if result_json.get("image"):
print(f"[@DEBUG] ✓ Successfully uploaded image for item {item_id}")
return True
else:
print(f"[@ERROR] Image upload failed: {result_json}")
return False
except Exception as e:
print(f"[@ERROR] Failed to upload image for item {item_id}: {e}")
return False
async def update_itm(item_id: str, lego_set: dict, item_data: dict = None):
"""
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
@@ -63,25 +287,230 @@ async def update_itm(lego_set: dict):
- "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
item_data (dict): Optional pre-fetched item data to avoid redundant API call
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"]
result = await client.catalog.search_items(text_filter=set_number, limit=1)
itm_id = json.loads(result.model_dump_json())["items"][0]["id"]
price_in_cents = int(set_adjustedprice * 100)
# Retry logic for version conflicts (race conditions with concurrent updates)
max_retries = 3
retry_delay = 0.5 # seconds
for attempt in range(max_retries):
try:
# ALWAYS fetch fresh item data right before update to get latest version
print(
f"[@DEBUG] Fetching fresh item data (attempt {attempt + 1}/{max_retries})..."
)
fresh_item = await get_item_by_id(item_id)
except ValueError as e:
# Item was deleted between webhook and enrichment
print(f"[@WARNING] {e} - skipping update (item may have been deleted)")
return
variations = fresh_item.get("item_data", {}).get("variations", [])
if not variations:
print(
f"[@WARNING] Item {item_id} has no variations, creating new variation"
)
variation_id = None
variation_version = None
present_at_location_ids = []
present_at_all_locations = False
else:
variation_id = variations[0]["id"]
variation_version = variations[0].get("version")
# Check if variation is at all locations
present_at_all_locations = variations[0].get(
"present_at_all_locations", False
)
# Extract location IDs from variation (only if not at all locations)
present_at_location_ids = variations[0].get(
"present_at_location_ids", []
)
# Get current version for version control
current_version = fresh_item.get("version")
# CRITICAL: batch_get NEVER returns category_id (always None)
# We MUST use the category from initial item_data (from search which includes category)
# or fallback to LEGO_CAT_ID since we know this is a LEGO item
if item_data and item_data.get("category_id"):
category_id = item_data.get("category_id")
print(f"[@DEBUG] Using category_id from item_data: {category_id}")
else:
# Fallback to LEGO_CAT_ID (we know it's a LEGO item if enrichment was triggered)
category_id = LEGO_CAT_ID
print(
f"[@WARNING] No category_id in item_data, using LEGO_CAT_ID: {LEGO_CAT_ID}"
)
print(f"[@DEBUG] Fresh item version: {current_version}")
print(f"[@DEBUG] Fresh variation version: {variation_version}")
print(f"[@DEBUG] Final category_id for upsert: {category_id}")
# Get the full existing item_data and modify only what we need
# This preserves ALL existing fields (location settings, tax, etc)
existing_item_data = fresh_item.get("item_data", {})
# Update item-level fields
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, keep all others unchanged
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 and only update what we need
updated_var = existing_var.copy()
updated_var_data = existing_var.get(
"item_variation_data", {}
).copy()
# Check if this variation has VARIABLE_PRICING
current_pricing_type = updated_var_data.get("pricing_type")
# Only set price if not using variable pricing
if current_pricing_type != "VARIABLE_PRICING":
updated_var_data["price_money"] = {
"amount": price_in_cents,
"currency": "USD",
}
print(f"[@DEBUG] Setting price: ${set_adjustedprice} for variation {variation_id}")
else:
print(f"[@WARNING] Skipping price update for {variation_id} - has VARIABLE_PRICING enabled")
updated_var_data["sku"] = set_number # LEGO set number
updated_var_data["upc"] = (
set_upc if set_upc else None
) # Barcode/GTIN
updated_var["item_variation_data"] = updated_var_data
# Update version (required for updates)
if variation_version is not None:
updated_var["version"] = variation_version
updated_variations.append(updated_var)
variation_found = True
else:
# Keeprice_money": {
"amount": price_in_cents,
"currency": "USD",
},
"p other variations as-is
updated_variations.append(existing_var)
# If variation wasn't found (shouldn't happen), create new one
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",
"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
item_obj = {
"type": "ITEM",
"id": item_id,
"version": current_version,
"item_data": existing_item_data,
}
upsert_body = {
"idempotency_key": str(uuid.uuid4()),
"object": item_obj,
}
# Remove None values
if not set_upc:
del variation_obj["item_variation_data"]["sku"]
print(f"[@DEBUG] Upserting item {item_id} with name: {pw_name}")
print(f"[@DEBUG] Upsert body: {json.dumps(upsert_body, indent=2)}")
# Perform upsert using catalog.object.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"):
print(f"[@DEBUG] ✓ Successfully updated item {item_id}")
# DEBUG: Check what Square actually returned
returned_item = result_json["catalog_object"]
returned_category = returned_item.get("item_data", {}).get(
"category_id"
)
print(f"[@DEBUG] Square returned category_id: {returned_category}")
# Upload image if available (DISABLED - concerns about storage costs)
# if set_image:
# print(f"[@DEBUG] Uploading image from {set_image}")
# await upload_item_image(item_id, set_image)
# await sq.catalog.object.upsert()
return True
else:
print(
f"[@ERROR] Failed to update item {item_id}: No catalog_object in response"
)
return False
except ApiError as e:
# Check if this is a VERSION_MISMATCH error
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:
print(
f"[@WARNING] Version conflict on attempt {attempt + 1}, retrying in {retry_delay}s..."
)
await asyncio.sleep(retry_delay)
retry_delay *= 2 # Exponential backoff
continue # Retry with fresh data
else:
print(
f"[@ERROR] Version conflict persists after {max_retries} attempts, giving up"
)
return False
# Re-raise other API errors
raise
except Exception as e:
print(f"[@ERROR] Failed to update item {item_id}: {e}")
traceback.print_exc()
return False
async def acat2id(cat_id: str):
+9 -2
View File
@@ -1,14 +1,21 @@
import os
import dotenv
from pathlib import Path
from square.environment import SquareEnvironment
dotenv.load_dotenv()
# Load .env from project root (parent of src/)
env_path = Path(__file__).parent.parent / ".env"
dotenv.load_dotenv(env_path)
LEGO_CAT_ID = os.getenv("SQUARE_LEGO_CAT_ID")
SQUARE_WEBHOOK_URL = os.getenv("SQUARE_WEBHOOK_URL")
if os.getenv("SQUARE_ENV") == "production":
# Check environment mode
IS_PRODUCTION = os.getenv("SQUARE_ENV") == "production"
DEBUG_MODE = not IS_PRODUCTION # Debug logging enabled in sandbox/dev
if IS_PRODUCTION:
SQUARE_APP_ID = os.getenv("PROD_SQUARE_APP_ID")
SQUARE_TOKEN = os.getenv("PROD_SQUARE_TOKEN")
SQUARE_WEBHOOK_SIGNATURE_KEY = os.getenv("PROD_SQUARE_WEBHOOK_SIGNATURE_KEY")
+199 -23
View File
@@ -9,33 +9,209 @@ from . import asquare
from . import db
from . import abricksetapi
from . import throttler
Throttler = throttler.GlobalThrottler
LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID
# big glue boy
async def do_review() -> bool:
result = await asquare.acat2id(LEGO_CAT_ID) # type: ignore
for item in result:
if not re.match(r".*\d{5}.*", item["name"]):
missing = []
print(f"{item} is missing its set number!")
#print(f"[@DEBUG] Upserting {item['id']} into REVIEW table")
item_sanitized = {
"sq_object": item["id"],
"set_name": item["name"],
"set_number": None,
"time_added": time
}
missing.append(item_sanitized)
for item in missing:
print(item)
# sq_item = await db.fetch(item["id"])
if item["id"] != sq_item["sq_object"]:
print(f"[@DEBUG] {item['id']} is missing)")
#await db.upsert("MISSING", item_sanitized, "sq_object")
# #print(f"[@DEBUG] Upserted {item['id']} into REVIEW table")
#print(f"[@DEBUG] {item['id']} already in REVIEW table")
async def enrich_single_item(item_id: str, item_data: dict = None, logger=None) -> bool:
r"""
Enrich a single Square catalog item with Brickset data.
Flow:
1. Get item details from Square (by item_id) - skip if item_data provided
2. Extract 5-digit set number from name (regex \d{5})
3. Fetch Brickset data (with throttling)
4. Update Square item with enriched data
5. Handle errors (no set number → MISSING table)
Args:
item_id: Square catalog item ID
item_data: Optional pre-fetched item data (avoids redundant API call)
logger: Optional logger instance for debug output
Returns:
True if successful, False otherwise
"""
try:
if logger:
logger.info(f"Starting enrichment for item {item_id}")
else:
print(f"[@DEBUG] Starting enrichment for item {item_id}")
# Get item details from Square (only if not already provided)
if item_data is None:
item = await asquare.get_item_by_id(item_id)
else:
item = item_data
item_name = item.get("name", "")
if logger:
logger.debug(f"Item name: {item_name}")
else:
print(f"[@DEBUG] Item name: {item_name}")
# Extract 5-digit set number
set_numbers = re.findall(r"\d{5}", item_name)
if not set_numbers:
# No set number found - log to MISSING table
if logger:
logger.warning(f"No set number found in item {item_id}: {item_name}")
else:
print(f"[@WARNING] No set number found in item {item_id}: {item_name}")
await db.upsert(
"MISSING",
{
"sq_object": item_id,
"time_added": int(time()),
"set_number": None,
"set_name": item_name,
},
"sq_object",
)
return False
set_number = set_numbers[0]
# Check if item is already enriched (format: "XXXXX - THEME Name")
# Pattern: 5 digits, space, dash, space, uppercase theme
if re.match(r"^\d{5}\s-\s[A-Z]", item_name):
if logger:
logger.info(f"Item {item_id} already enriched, skipping: {item_name}")
else:
print(
f"[@DEBUG] Item {item_id} already enriched, skipping: {item_name}"
)
return True
if logger:
logger.info(f"Extracted set number: {set_number}")
else:
print(f"[@DEBUG] Extracted set number: {set_number}")
# Fetch Brickset data (with throttling)
# debug flag will be auto-set based on SQUARE_ENV
brickset_client = abricksetapi.AsyncBricksetAPI(logger=logger)
lego_set = await brickset_client.throttled_get_set(set_number)
if logger:
logger.info(f"Retrieved Brickset data for {lego_set['pw_name']}")
else:
print(f"[@DEBUG] Retrieved Brickset data for {lego_set['pw_name']}")
# Note: Price updates disabled - enriching without price
# REVIEW table logic skipped
# Update Square item with enriched data (without price)
success = await asquare.update_itm(item_id, lego_set, item_data=item)
if success:
if logger:
logger.info(f"✓ Successfully enriched item {item_id}")
else:
print(f"[@DEBUG] ✓ Successfully enriched item {item_id}")
return True
else:
if logger:
logger.error(f"Failed to update Square item {item_id}")
else:
print(f"[@ERROR] Failed to update Square item {item_id}")
return False
except ValueError as e:
if logger:
logger.error(f"Value error enriching item {item_id}: {e}")
else:
print(f"[@ERROR] Value error enriching item {item_id}: {e}")
return False
except Exception as e:
if logger:
logger.error(f"Unexpected error enriching item {item_id}: {e}")
else:
print(f"[@ERROR] Unexpected error enriching item {item_id}: {e}")
import traceback
traceback.print_exc()
return False
async def enrich_bulk(logger=None) -> None:
"""
Enrich all items in the LEGO category (bulk operation).
This is the refactored RUNUPDATE command logic.
Iterates all items in LEGO category and enriches each one.
Args:
logger: Optional logger instance for debug output
"""
try:
if logger:
logger.info("Starting bulk enrichment for LEGO category")
else:
print(f"[@DEBUG] Starting bulk enrichment for LEGO category {LEGO_CAT_ID}")
# Get all items in LEGO category
items = await asquare.acat2id(LEGO_CAT_ID)
if logger:
logger.info(f"Found {len(items)} items in LEGO category")
else:
print(f"[@DEBUG] Found {len(items)} items in LEGO category")
# Enrich each item with concurrent batches
successful = 0
failed = 0
batch_size = 20 # Process 20 items concurrently
for i in range(0, len(items), batch_size):
batch = items[i : i + batch_size]
batch_num = (i // batch_size) + 1
total_batches = (len(items) + batch_size - 1) // batch_size
if logger:
logger.info(
f"Processing batch {batch_num}/{total_batches} ({len(batch)} items)"
)
else:
print(
f"[@DEBUG] Processing batch {batch_num}/{total_batches} ({len(batch)} items)"
)
# Process batch concurrently
tasks = [enrich_single_item(item["id"], logger=logger) for item in batch]
results = await asyncio.gather(*tasks, return_exceptions=True)
# Count results
for result in results:
if isinstance(result, Exception):
failed += 1
elif result:
successful += 1
else:
failed += 1
if logger:
logger.info(
f"Bulk enrichment complete: {successful} successful, {failed} failed"
)
else:
print(
f"[@DEBUG] Bulk enrichment complete: {successful} successful, {failed} failed"
)
except Exception as e:
if logger:
logger.error(f"Error during bulk enrichment: {e}")
else:
print(f"[@ERROR] Error during bulk enrichment: {e}")
import traceback
traceback.print_exc()
File renamed without changes.
File renamed without changes.
File renamed without changes.
+277
View File
@@ -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)
+426
View File
@@ -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
+89
View File
@@ -0,0 +1,89 @@
"""
List all Square categories with their IDs to help identify the correct LEGO category ID
"""
import asyncio
import json
from .. import asquare
from .. import get_sq_env
async def list_categories():
"""List all categories in Square catalog"""
try:
print("\n" + "=" * 60)
print("SQUARE CATALOG CATEGORIES")
print("=" * 60 + "\n")
cursor = None
page_count = 0
total_categories = 0
while True:
page_count += 1
print(f"Fetching page {page_count}...")
search_params = {
"object_types": ["CATEGORY"],
"limit": 1000,
}
if cursor:
search_params["cursor"] = cursor
result = await asquare.client.catalog.search(**search_params)
result_json = json.loads(result.model_dump_json())
if not result_json.get("objects"):
if page_count == 1:
print("No categories found!")
break
for obj in result_json["objects"]:
category_data = obj.get("category_data", {})
category_id = obj.get("id")
category_name = category_data.get("name", "Unnamed")
print(f"\nCategory: {category_name}")
print(f" ID: {category_id}")
# Count items in this category
items_result = await asquare.client.catalog.search(
object_types=["ITEM"],
query={
"exact_query": {
"attribute_name": "category_id",
"attribute_value": category_id,
}
},
limit=1,
)
items_json = json.loads(items_result.model_dump_json())
item_count = len(items_json.get("objects", []))
print(f" Items: {item_count}+ items")
total_categories += 1
cursor = result_json.get("cursor")
if not cursor:
break
print("\n" + "=" * 60)
print(f"Total categories found: {total_categories}")
print("=" * 60 + "\n")
# Show current configured category
print(f"Currently configured LEGO_CAT_ID: {get_sq_env.LEGO_CAT_ID}")
print("\nUpdate .env file with the correct category ID if needed:")
print('SQUARE_LEGO_CAT_ID = "YOUR_CATEGORY_ID_HERE"')
print()
except Exception as e:
print(f"Error listing categories: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
asyncio.run(list_categories())
+162
View File
@@ -0,0 +1,162 @@
"""
Migrate SKU values to GTIN/UPC field for all LEGO items.
This script reads existing Square catalog items in the LEGO category,
and for each variation that has a SKU but no UPC, it moves the SKU
value to the UPC field (leaving SKU empty or optionally extracting
set number from item name).
"""
import asyncio
import json
import re
import sys
import uuid
from pathlib import Path
# Add parent directory to path for imports
sys.path.insert(0, str(Path(__file__).parent.parent))
import get_sq_env
from square import AsyncSquare
SQUARE_LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID
async def migrate_sku_to_gtin():
"""Migrate SKU values to GTIN for all LEGO items."""
client = AsyncSquare(
token=get_sq_env.SQUARE_TOKEN,
environment=get_sq_env.environment,
)
print(f"[@INFO] Searching for items in LEGO category {SQUARE_LEGO_CAT_ID}...")
# Search all items in LEGO category
cursor = None
page = 1
total_items = 0
updated_items = 0
while True:
print(f"[@INFO] Processing page {page}...")
search_params = {
"object_types": ["ITEM"],
"query": {
"exact_query": {
"attribute_name": "category_id",
"attribute_value": SQUARE_LEGO_CAT_ID,
}
},
"limit": 100,
}
if cursor:
search_params["cursor"] = cursor
result = await client.catalog.search(**search_params)
result_json = json.loads(result.model_dump_json())
items = result_json.get("objects", [])
cursor = result_json.get("cursor")
if not items:
break
# Process each item
for item in items:
item_id = item["id"]
item_data = item.get("item_data", {})
item_name = item_data.get("name", "")
variations = item_data.get("variations", [])
item_version = item.get("version")
total_items += 1
needs_update = False
updated_variations = []
# Extract set number from item name if it matches LEGO format
set_number = None
match = re.search(r"^(\d{5})\s*-", item_name)
if match:
set_number = match.group(1)
# Check each variation
for var in variations:
var_data = var.get("item_variation_data", {})
var_id = var.get("id")
var_version = var.get("version")
sku = var_data.get("sku")
upc = var_data.get("upc")
# If SKU exists but UPC doesn't, migrate
if sku and not upc:
print(f"[@INFO] Item {item_id} ({item_name})")
print(f"[@INFO] Variation {var_id}: Moving SKU '{sku}' to UPC")
# 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
# 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:
# Keep variation as-is
updated_variations.append(var)
# Update item if needed
if needs_update:
try:
# Get fresh item data to ensure we have latest version
fresh_result = await client.catalog.batch_get(
object_ids=[item_id],
include_related_objects=False,
)
fresh_json = json.loads(fresh_result.model_dump_json())
fresh_item = fresh_json["objects"][0]
fresh_item_data = fresh_item.get("item_data", {})
fresh_version = fresh_item.get("version")
# Update variations while preserving all other fields
fresh_item_data["variations"] = updated_variations
await client.catalog.object.upsert(
idempotency_key=str(uuid.uuid4()),
object={
"type": "ITEM",
"id": item_id,
"version": fresh_version,
"item_data": fresh_item_data,
},
)
print(f"[@INFO] ✓ Successfully updated item {item_id}")
updated_items += 1
# Small delay to avoid rate limits
await asyncio.sleep(0.1)
except Exception as e:
print(f"[@ERROR] Failed to update item {item_id}: {e}")
page += 1
# If no cursor, we're done
if not cursor:
break
print(f"\n[@INFO] Migration complete!")
print(f"[@INFO] Total items processed: {total_items}")
print(f"[@INFO] Items updated: {updated_items}")
if __name__ == "__main__":
print("[@INFO] Starting SKU to GTIN migration...\n")
asyncio.run(migrate_sku_to_gtin())
@@ -38,12 +38,17 @@ def get_set(set_number: int) -> dict:
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": data["sets"][0]["barcode"]["UPC"],
"set_barcode": upc_code,
"set_retailprice": round(data["sets"][0]["LEGOCom"]["US"]["retailPrice"]),
}
# add PW 10% b4 tax
@@ -51,7 +56,7 @@ def get_set(set_number: int) -> dict:
{"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']}"
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
File renamed without changes.
+70 -9
View File
@@ -4,15 +4,24 @@ Queries Square catalog to collect metrics about LEGO inventory quality.
"""
import json
import git
from typing import Dict, Any
from square import AsyncSquare
import get_sq_env
from . import get_sq_env
from . import abricksetapi
def get_git_commit() -> str:
"""Get first 6 characters of current git commit hash"""
try:
repo = git.Repo(search_parent_directories=True)
return repo.head.commit.hexsha[:6]
except Exception:
return "unknown"
# Initialize Square client
client = AsyncSquare(
token=get_sq_env.SQUARE_TOKEN,
environment=get_sq_env.environment
)
client = AsyncSquare(token=get_sq_env.SQUARE_TOKEN, environment=get_sq_env.environment)
LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID
@@ -104,12 +113,62 @@ async def get_daily_stats() -> Dict[str, Any]:
}
async def get_total_items() -> int:
async def get_brickset_cache_stats() -> Dict[str, Any]:
"""
Count total items in LEGO category.
Get Brickset API cache statistics.
Returns:
int: Total number of items
dict: {
"cache_enabled": bool,
"cache_hits": int,
"cache_misses": int,
"cache_size": int,
"cache_maxsize": int,
"hit_rate": float (0-100%)
}
"""
# Create a temporary client instance to access cache info
temp_client = abricksetapi.AsyncBricksetAPI()
# Check if get_set has cache_info (it should with @alru_cache)
if hasattr(temp_client.get_set, "cache_info"):
info = await temp_client.get_set.cache_info()
hits = info.hits
misses = info.misses
total = hits + misses
hit_rate = (hits / total * 100) if total > 0 else 0.0
return {
"cache_enabled": True,
"cache_hits": hits,
"cache_misses": misses,
"cache_size": info.currsize,
"cache_maxsize": info.maxsize,
"hit_rate": round(hit_rate, 2),
}
else:
return {
"cache_enabled": False,
"cache_hits": 0,
"cache_misses": 0,
"cache_size": 0,
"cache_maxsize": 0,
"hit_rate": 0.0,
}
async def get_daily_stats() -> Dict[str, Any]:
"""
Gather all metrics for daily_stats table in a single pass.
Returns:
dict: {
"total_items": int,
"enriched_items": int,
"missing_upcs": int,
"missing_images": int,
"total_catalog_value": float
}
"""
result = await client.catalog.search(
object_types=["ITEM"],
@@ -293,6 +352,8 @@ if __name__ == "__main__":
print(f"Missing UPCs: {stats['missing_upcs']}")
print(f"Missing Images: {stats['missing_images']}")
print(f"Total Catalog Value: ${stats['total_catalog_value']:,.2f}")
print(f"Completion Rate: {(stats['enriched_items'] / stats['total_items'] * 100):.1f}%")
print(
f"Completion Rate: {(stats['enriched_items'] / stats['total_items'] * 100):.1f}%"
)
asyncio.run(main())
+11 -7
View File
@@ -1,19 +1,24 @@
import asyncio
from asyncio_throttle import Throttler as AsyncThrottler
class GlobalThrottler:
_instance = None
_lock = asyncio.Lock()
def __init__(self, rate_limit: int = 500, period: int = 60) -> None:
if not hasattr(self, '_initialized'):
if not hasattr(self, "_initialized"):
self.rate_limit = rate_limit
self.period = period
self.throttler = AsyncThrottler(rate_limit=self.rate_limit, period=self.period)
self.throttler = AsyncThrottler(
rate_limit=self.rate_limit, period=self.period
)
self._initialized = True
@classmethod
async def get_instance(cls, rate_limit: int = 500, period: int = 60) -> 'GlobalThrottler':
async def get_instance(
cls, rate_limit: int = 500, period: int = 60
) -> "GlobalThrottler":
async with cls._lock:
if not cls._instance:
cls._instance = cls(rate_limit, period)
@@ -22,12 +27,11 @@ class GlobalThrottler:
async def acquire(self):
await self.throttler.acquire()
def release(self):
self.throttler.release()
async def __aenter__(self):
await self.acquire()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
self.release()
# asyncio_throttle.Throttler doesn't have a release() method
# It automatically manages rate limiting via acquire()
pass
+113 -15
View File
@@ -5,6 +5,7 @@ from typing import Dict, Any
from fastapi import FastAPI, Request, Response, status
from fastapi.responses import JSONResponse
import src.get_sq_env as sq_env
import src.stats as stats
# Square webhook validation
from square.utils.webhooks_helper import verify_signature
@@ -12,17 +13,7 @@ from square.utils.webhooks_helper import verify_signature
# Configuration
SQUARE_WEBHOOK_SIGNATURE_KEY = sq_env.SQUARE_WEBHOOK_SIGNATURE_KEY
SQUARE_WEBHOOK_URL = sq_env.SQUARE_WEBHOOK_URL
# Debug: Print configuration (redact key)
if SQUARE_WEBHOOK_SIGNATURE_KEY:
print(f"[CONFIG] Signature key loaded: {SQUARE_WEBHOOK_SIGNATURE_KEY[:10]}...")
else:
print("[CONFIG] WARNING: SQUARE_WEBHOOK_SIGNATURE_KEY not set!")
if SQUARE_WEBHOOK_URL:
print(f"[CONFIG] Webhook URL: {SQUARE_WEBHOOK_URL}")
else:
print("[CONFIG] WARNING: SQUARE_WEBHOOK_URL not set!")
SQUARE_LEGO_CAT_ID = sq_env.LEGO_CAT_ID
# In-memory store for processed event IDs (idempotency)
processed_event_ids = set()
@@ -38,7 +29,11 @@ async def root():
return {
"service": "Studspider",
"status": "running",
"endpoints": {"webhook": "/hook (POST)", "health": "/health (GET)"},
"endpoints": {
"webhook": "/hook (POST)",
"health": "/health (GET)",
"stats": "/stats (GET)",
},
}
@@ -53,6 +48,40 @@ async def health_check():
}
@app.get("/stats")
async def get_stats():
"""
Get daily statistics for LEGO inventory.
Returns metrics about enrichment progress, missing data, catalog value, and cache stats.
"""
try:
daily_stats = await stats.get_daily_stats()
cache_stats = await stats.get_brickset_cache_stats()
# Calculate completion percentage
if daily_stats["total_items"] > 0:
completion_rate = (
daily_stats["enriched_items"] / daily_stats["total_items"] * 100
)
else:
completion_rate = 0.0
return {
"status": "success",
"version": "0.3.0",
"git_commit": stats.get_git_commit(),
"stats": daily_stats,
"completion_rate": round(completion_rate, 1),
"brickset_cache": cache_stats,
}
except Exception as e:
return JSONResponse(
status_code=500,
content={"status": "error", "message": str(e)},
)
@app.post("/hook")
async def webhook_handler(request: Request):
"""
@@ -233,17 +262,83 @@ async def process_webhook(
inventory_counts = (
data.get("data", {}).get("object", {}).get("inventory_counts", [])
)
# Import here to avoid circular imports
import src.asquare as asquare
import src.glue as glue
for count in inventory_counts:
catalog_id = count.get("catalog_object_id")
variation_id = count.get("catalog_object_id")
quantity = count.get("quantity")
state = count.get("state")
print(
f"[PROCESS] Catalog: {catalog_id}, Qty: {quantity}, State: {state}"
f"[PROCESS] Catalog: {variation_id}, Qty: {quantity}, State: {state}"
)
try:
# Get parent item from variation ID
item = await asquare.get_item_by_variation_id(variation_id)
item_id = item.get("id")
item_name = item.get("name", "")
category_id = item.get("category_id")
print(f"[PROCESS] Item: {item_id} - {item_name}")
print(f"[PROCESS] Category: {category_id}")
# Only process LEGO category items
if category_id == SQUARE_LEGO_CAT_ID:
print(
f"[PROCESS] ✓ LEGO item detected - triggering enrichment"
)
# Trigger enrichment in background with item data to avoid redundant lookup
asyncio.create_task(
glue.enrich_single_item(item_id, item_data=item)
)
else:
print(f"[PROCESS] ✗ Not a LEGO item - skipping enrichment")
except Exception as e:
print(f"[ERROR] Failed to process variation {variation_id}: {e}")
elif webhook_type == "catalog.version.updated":
print(f"[PROCESS] Processing catalog version update...")
# Add your catalog processing logic here
# Import here to avoid circular imports
import src.asquare as asquare
import src.glue as glue
catalog_object = data.get("data", {}).get("object", {})
object_type = catalog_object.get("type")
if object_type == "ITEM":
item_id = catalog_object.get("id")
print(f"[PROCESS] Item updated: {item_id}")
try:
# Get item details to check category
item = await asquare.get_item_by_id(item_id)
item_name = item.get("name", "")
category_id = item.get("category_id")
print(f"[PROCESS] Item: {item_id} - {item_name}")
print(f"[PROCESS] Category: {category_id}")
# Only process LEGO category items
if category_id == SQUARE_LEGO_CAT_ID:
print(
f"[PROCESS] ✓ LEGO item detected - triggering enrichment"
)
# Trigger enrichment in background with item data to avoid redundant lookup
asyncio.create_task(
glue.enrich_single_item(item_id, item_data=item)
)
else:
print(f"[PROCESS] ✗ Not a LEGO item - skipping enrichment")
except Exception as e:
print(f"[ERROR] Failed to process item {item_id}: {e}")
else:
print(f"[PROCESS] Object type '{object_type}' - skipping")
else:
print(f"[PROCESS] Webhook type '{webhook_type}' - no handler configured")
@@ -252,6 +347,9 @@ async def process_webhook(
except Exception as e:
print(f"[ERROR] Failed to process webhook {event_id}: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
+2 -2
View File
@@ -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")
+7 -21
View File
@@ -54,29 +54,15 @@ UA = re.compile(r"^kopling\/?", re.IGNORECASE)
async def run_update():
set_list = await sq.acat2id(SQUARE_LEGO_CAT_ID) # type: ignore
# 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] Found {l_setinfo["pw_name"]}")
except Exception as e:
print(f"Error fetching set\t: {l_set}: {str(e)}")
continue
"""
DEPRECATED: Use glue.enrich_bulk() instead
# print(f"[@DEBUG]{l_setinfo}")
# await sq.update_itm(l_set, l_setinfo)
"""for l_set in set_list:
print(f"[@DEBUG]{l_set}")
try:
l_setinfo = await lego.throttled_get_set(l_set["number"])
except Exception as e:
print(f"Error fetching set\t: {l_set}: {str(e)}")
continue
Legacy RUNUPDATE command handler - kept for backwards compatibility
but now delegates to the new glue.enrich_bulk() function
"""
from src import glue
print(f"[@DEBUG]{l_setinfo}")"""
# await sq.update_itm(l_set, l_setinfo)
await glue.enrich_bulk()
def valid_uuid(val):