59 Commits
Author SHA1 Message Date
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
C 03f6e84965 name changes 2026-01-04 10:41:18 -08:00
C bb6704f3aa fix asquare to match brickset 2026-01-04 10:41:10 -08:00
C 63a8c29105 reporting stats 2026-01-04 10:40:43 -08:00
C 84cb6becf9 update sq env 2026-01-04 10:40:29 -08:00
C 48835cd857 im actually gonna crash out rn dude 2026-01-03 17:24:49 -08:00
C c4bca1e4d0 bro my grammar was so ass. was i drunk when i wrote this? 2026-01-03 16:45:00 -08:00
C 42b4575cc4 update env 2026-01-03 16:34:19 -08:00
C a51c19ea53 complete project restructure 2026-01-03 16:08:48 -08:00
21 changed files with 1620 additions and 470 deletions

No files matched your search

+17 -78
View File
@@ -1,83 +1,22 @@
# studspider Environment Configuration
# Brickset.com credentials - LEGO database for pricing, images, UPC codes
BRICKSET_USERNAME = "your_brickset_username"
BRICKSET_PASS = "your_brickset_password"
BRICKSET_API_KEY = "your_brickset_api_key"
# ============================================================================
# Square API Configuration
# ============================================================================
# Square environment - "sandbox" for testing, "production" for live store
SQUARE_ENV = "sandbox" # or "production"
# Environment: "production" or "sandbox"
SQUARE_ENV=sandbox
SQUARE_WEBHOOK_URL = "https://yourdomain.com/hook"
# Sandbox credentials
SANDBOX_SQUARE_APP_ID=your_sandbox_app_id
SANDBOX_SQUARE_TOKEN=your_sandbox_access_token
# 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"
# Production credentials
PROD_SQUARE_APP_ID=your_production_app_id
PROD_SQUARE_TOKEN=your_production_access_token
# 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"
# Square LEGO category ID (find this in Square Dashboard)
SQUARE_LEGO_CAT_ID=your_lego_category_id
# ============================================================================
# Brickset API Configuration
# ============================================================================
BRICKSET_USERNAME=your_brickset_username
BRICKSET_PASS=your_brickset_password
BRICKSET_API_KEY=your_brickset_api_key
# ============================================================================
# SSL Certificate Configuration (Optional)
# ============================================================================
# NOTE: Only needed if running HTTPS directly in Python (not recommended)
# For production, use nginx reverse proxy with Let's Encrypt instead
# nginx handles SSL/TLS, your Python app runs HTTP internally
# Path to SSL certificate in PEM format (if running HTTPS directly)
# SSL_CERT_PATH=/path/to/cert.pem
# Path to SSL private key in PEM format (if running HTTPS directly)
# SSL_KEY_PATH=/path/to/key.pem
# ============================================================================
# Square Webhook Signature Verification
# ============================================================================
# CRITICAL: Get these from Square Developer Dashboard → Webhooks
# 1. Go to https://developer.squareup.com/apps
# 2. Select your application
# 3. Click "Webhooks" section
# 4. Create webhook subscription (Notification URL + Events)
# 5. Copy the Signature Key that Square displays
# Signature key from Square Dashboard (a string like "asdf1234")
SQUARE_WEBHOOK_SIGNATURE_KEY=your_webhook_signature_key_from_square
# Your webhook notification URL (must match EXACTLY what's in Square Dashboard)
# This should be your public HTTPS URL (through nginx)
# Example: https://yourdomain.com/hook
SQUARE_WEBHOOK_URL=https://yourdomain.com/hook
# ============================================================================
# Server Configuration (Optional)
# ============================================================================
# Default host and port (can override with command line args)
# SERVER_HOST=0.0.0.0
# SERVER_PORT=7432
# SERVER_HOOK_PORT=8080
# ============================================================================
# Usage Examples
# ============================================================================
# Basic HTTP server (recommended with nginx):
# python src/test_server.py
#
# With SSL from .env:
# Set SSL_CERT_PATH and SSL_KEY_PATH above, then:
# python src/test_server.py --hook-port 8443 --hook-ssl-only
#
# Override SSL from command line:
# python src/test_server.py --ssl-cert /custom/cert.pem --ssl-key /custom/key.pem
# Category ID for LEGO items in Square catalog (find in Dashboard → Items → Categories)
SQUARE_LEGO_CAT_ID = "your_lego_category_id"
+5 -4
View File
@@ -5,14 +5,14 @@
is a tool used to talk to brickset.com and a square inventory to pull lego set information for entering in the square db
is a tool used to talk to brickset.com, pull lego set information and automatically enter it into the square inventory!
## Run me
Clone the project
```bash
git clone http://git.traps.moe/ph0sgene/studspider.git
git checkout uv
git checkout branch
```
Go to the project directory
@@ -30,14 +30,15 @@ Install dependencies
Start the server
```bash
uvx run studspider.py
uv run studspider.py
```
## Tech
- gunicorn
- uvicorn
- aiohttp
- squareup
- apscheduler
- aiosqlite
@@ -1,10 +1,7 @@
"""
Gunicorn configuration for StudSpider webhook server
Production-ready settings for running with nginx reverse proxy
"""
import multiprocessing
import os
import src.banner as banner
# Server socket
bind = "127.0.0.1:7433" # nginx will proxy to this
@@ -28,11 +25,11 @@ loglevel = "info"
access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'
# Process naming
proc_name = "studspider-webhook"
proc_name = "studspider"
# Server mechanics
daemon = False # Run in foreground (systemd will handle daemonization)
pidfile = "./logs/webhook.pid"
pidfile = "./logs/studspider.pid"
umask = 0
user = None # Run as current user (systemd will set user)
group = None
@@ -57,6 +54,29 @@ def on_starting(server):
"""Called just before the master process is initialized."""
print("Studspider WEB starting...")
# Load .env file before workers start
import dotenv
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():
dotenv.load_dotenv(env_path)
print(f"✓ Loaded environment from {env_path}")
# Verify key was loaded
import os
key = os.getenv("PROD_SQUARE_WEBHOOK_SIGNATURE_KEY") or os.getenv(
"SANDBOX_SQUARE_WEBHOOK_SIGNATURE_KEY"
)
if key:
print(f"✓ SQUARE_WEBHOOK_SIGNATURE_KEY loaded: {key[:10]}...")
else:
print(f"⚠ SQUARE_WEBHOOK_SIGNATURE_KEY not found in .env")
else:
print(f"⚠ No .env file found at {env_path.absolute()}")
def on_reload(server):
"""Called to recycle workers during a reload via SIGHUP."""
@@ -64,7 +84,8 @@ def on_reload(server):
def when_ready(server):
"""Called just after the server is started."""
print(banner.generate("studspider v0.1"))
print("Built for PLAY-WELL TEKNOLOGIES by contact@studspider.sh\n\n")
print(f"STUDSPIDER HOOK v0.1 RUNNING ON {bind}")
print(f"WORKERS: {workers}")
print(f"WORKING CLASS: {worker_class}\n")
-11
View File
@@ -1,11 +0,0 @@
{
"folders": [
{
"path": "."
},
{
"path": "../kopling"
}
],
"settings": {}
}
+8 -3
View File
@@ -9,7 +9,7 @@ import aiohttp
import json
import dotenv
from typing import Dict, Any
import throttler
from . import throttler
Throttler = throttler.GlobalThrottler
@@ -101,12 +101,17 @@ 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")
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_barcode": upc_code,
"set_retailprice": round(
set_data["LEGOCom"]["US"]["retailPrice"]
),
@@ -120,7 +125,7 @@ class AsyncBricksetAPI:
# 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
+435 -17
View File
@@ -5,10 +5,13 @@ import asyncio
import uuid
import re
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,33 +54,448 @@ 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
while True:
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
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
break
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:
lego_set (dict): A dictionary containing LEGO set details. Expected keys include:
- "sets": List of set dictionaries, each with "name", "number", "barcode" (with "UPC"), and "image" (with "thumbnailURL").
- "pw_name": The name associated with the LEGO set in another system.
- "set_adjustredprice": The adjusted price for the set.
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
- "set_theme": LEGO theme (e.g., "Star Wars")
- "set_barcode": UPC barcode
- "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"]
set_name = lego_set["sets"][0]["name"]
set_number = lego_set["sets"][0]["number"]
set_upc = lego_set["sets"][0]["barcode"]["UPC"]
set_image = lego_set["sets"][0]["image"]["thumbnailURL"]
pw_name = lego_set["pw_name"]
set_adjustedprice = lego_set["set_adjustredprice"]
# Convert price to cents for Square API
price_in_cents = int(set_adjustedprice * 100)
result = await client.catalog.search_items(text_filter=set_number, limit=1)
itm_id = json.loads(result.model_dump_json())["items"][0]["id"]
# Retry logic for version conflicts (race conditions with concurrent updates)
max_retries = 3
retry_delay = 0.5 # seconds
# await sq.catalog.object.upsert()
return True
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()
# Update only specific fields
updated_var_data["price_money"] = {
"amount": price_in_cents,
"currency": "USD",
}
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:
# Keep 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",
"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
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)
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}")
import traceback
traceback.print_exc()
return False
async def acat2id(cat_id: str):
+6 -1
View File
@@ -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
+4 -7
View File
@@ -6,20 +6,17 @@ dotenv.load_dotenv()
LEGO_CAT_ID = os.getenv("SQUARE_LEGO_CAT_ID")
# SSL Certificate paths (optional - for direct HTTPS, not needed with nginx)
SSL_CERT_PATH = os.getenv("SSL_CERT_PATH")
SSL_KEY_PATH = os.getenv("SSL_KEY_PATH")
# Square Webhook Configuration
# Get signature key from Square Developer Dashboard → Webhooks
SQUARE_WEBHOOK_SIGNATURE_KEY = os.getenv("SQUARE_WEBHOOK_SIGNATURE_KEY")
SQUARE_WEBHOOK_URL = os.getenv("SQUARE_WEBHOOK_URL")
if os.getenv("SQUARE_ENV") == "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")
SQUARE_WEBHOOK_SUB_ID = os.getenv("PROD_SQUARE_WEBHOOK_SUB_ID")
environment = SquareEnvironment.PRODUCTION
else:
SQUARE_APP_ID = os.getenv("SANDBOX_SQUARE_APP_ID")
SQUARE_TOKEN = os.getenv("SANDBOX_SQUARE_TOKEN")
SQUARE_WEBHOOK_SIGNATURE_KEY = os.getenv("SANDBOX_SQUARE_WEBHOOK_SIGNATURE_KEY")
SQUARE_WEBHOOK_SUB_ID = os.getenv("SANDBOX_SQUARE_WEBHOOK_SUB_ID")
environment = SquareEnvironment.SANDBOX
+194 -24
View File
@@ -9,33 +9,203 @@ 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")
return True
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)
brickset_client = abricksetapi.AsyncBricksetAPI(debug=True, 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']}")
# Check if manual price review needed
if lego_set.get("requires_manual_price"):
if logger:
logger.info(f"Item {item_id} requires manual price review (≥$110)")
else:
print(f"[@DEBUG] Item {item_id} requires manual price review")
await db.upsert(
"REVIEW",
{
"sq_object": item_id,
"price": lego_set["set_adjustredprice"],
"date_modified": int(time()),
},
"sq_object",
)
# Update Square item with enriched data
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
successful = 0
failed = 0
for item in items:
item_id = item["id"]
result = await enrich_single_item(item_id, logger=logger)
if 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()
+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())
File renamed without changes.
View File
File renamed without changes.
+175
View File
@@ -0,0 +1,175 @@
"""
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")
# 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 pricing info
if "price_money" in var_data:
updated_var["item_variation_data"]["price_money"] = var_data[
"price_money"
]
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())
+298
View File
@@ -0,0 +1,298 @@
"""
Statistics gathering functions for daily_stats table.
Queries Square catalog to collect metrics about LEGO inventory quality.
"""
import json
from typing import Dict, Any
from square import AsyncSquare
import get_sq_env
# Initialize Square client
client = AsyncSquare(
token=get_sq_env.SQUARE_TOKEN,
environment=get_sq_env.environment
)
LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID
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
}
"""
# Get all items in LEGO category
result = await client.catalog.search(
object_types=["ITEM"],
query={
"exact_query": {
"attribute_name": "category_id",
"attribute_value": LEGO_CAT_ID,
}
},
)
result_json = json.loads(result.model_dump_json())
objects = result_json.get("objects", [])
# Initialize counters
total_items = 0
enriched_items = 0
missing_upcs = 0
missing_images = 0
total_catalog_value = 0.0
for obj in objects:
total_items += 1
item_data = obj.get("item_data", {})
# Check if item has all enrichment data
has_upc = False
has_image = False
has_proper_name = False
item_price = 0.0
# Check variations for UPC and price
variations = item_data.get("variations", [])
if variations:
var_data = variations[0].get("item_variation_data", {})
# Check UPC
if var_data.get("upc"):
has_upc = True
else:
missing_upcs += 1
# Get price (stored in cents)
price_money = var_data.get("price_money", {})
if price_money and price_money.get("amount"):
item_price = price_money["amount"] / 100.0 # Convert cents to dollars
total_catalog_value += item_price
else:
missing_upcs += 1
# Check for image
if item_data.get("image_ids"):
has_image = True
else:
missing_images += 1
# Check if name follows format: "12345 - THEME Set Name"
name = item_data.get("name", "")
if " - " in name and name.split()[0].isdigit() and len(name.split()[0]) == 5:
has_proper_name = True
# Item is "enriched" if it has UPC, image, and proper name format
if has_upc and has_image and has_proper_name:
enriched_items += 1
return {
"total_items": total_items,
"enriched_items": enriched_items,
"missing_upcs": missing_upcs,
"missing_images": missing_images,
"total_catalog_value": round(total_catalog_value, 2),
}
async def get_total_items() -> int:
"""
Count total items in LEGO category.
Returns:
int: Total number of items
"""
result = await client.catalog.search(
object_types=["ITEM"],
query={
"exact_query": {
"attribute_name": "category_id",
"attribute_value": LEGO_CAT_ID,
}
},
)
result_json = json.loads(result.model_dump_json())
return len(result_json.get("objects", []))
async def get_enriched_count() -> int:
"""
Count items with complete enrichment (UPC + image + proper name format).
Returns:
int: Number of fully enriched items
"""
result = await client.catalog.search(
object_types=["ITEM"],
query={
"exact_query": {
"attribute_name": "category_id",
"attribute_value": LEGO_CAT_ID,
}
},
)
result_json = json.loads(result.model_dump_json())
objects = result_json.get("objects", [])
enriched_count = 0
for obj in objects:
item_data = obj.get("item_data", {})
# Check UPC
has_upc = False
variations = item_data.get("variations", [])
if variations:
var_data = variations[0].get("item_variation_data", {})
has_upc = bool(var_data.get("upc"))
# Check image
has_image = bool(item_data.get("image_ids"))
# Check name format: "12345 - THEME Set Name"
name = item_data.get("name", "")
has_proper_name = False
if " - " in name:
first_part = name.split()[0]
if first_part.isdigit() and len(first_part) == 5:
has_proper_name = True
if has_upc and has_image and has_proper_name:
enriched_count += 1
return enriched_count
async def get_missing_upcs() -> int:
"""
Count items without UPC barcodes.
Returns:
int: Number of items missing UPCs
"""
result = await client.catalog.search(
object_types=["ITEM"],
query={
"exact_query": {
"attribute_name": "category_id",
"attribute_value": LEGO_CAT_ID,
}
},
)
result_json = json.loads(result.model_dump_json())
objects = result_json.get("objects", [])
missing_count = 0
for obj in objects:
item_data = obj.get("item_data", {})
variations = item_data.get("variations", [])
if not variations:
missing_count += 1
continue
var_data = variations[0].get("item_variation_data", {})
if not var_data.get("upc"):
missing_count += 1
return missing_count
async def get_missing_images() -> int:
"""
Count items without images.
Returns:
int: Number of items missing images
"""
result = await client.catalog.search(
object_types=["ITEM"],
query={
"exact_query": {
"attribute_name": "category_id",
"attribute_value": LEGO_CAT_ID,
}
},
)
result_json = json.loads(result.model_dump_json())
objects = result_json.get("objects", [])
missing_count = 0
for obj in objects:
item_data = obj.get("item_data", {})
if not item_data.get("image_ids"):
missing_count += 1
return missing_count
async def get_total_catalog_value() -> float:
"""
Calculate total dollar value of all items in LEGO category.
Returns:
float: Total catalog value in dollars
"""
result = await client.catalog.search(
object_types=["ITEM"],
query={
"exact_query": {
"attribute_name": "category_id",
"attribute_value": LEGO_CAT_ID,
}
},
)
result_json = json.loads(result.model_dump_json())
objects = result_json.get("objects", [])
total_value = 0.0
for obj in objects:
item_data = obj.get("item_data", {})
variations = item_data.get("variations", [])
if variations:
var_data = variations[0].get("item_variation_data", {})
price_money = var_data.get("price_money", {})
if price_money and price_money.get("amount"):
# Price is stored in cents, convert to dollars
total_value += price_money["amount"] / 100.0
return round(total_value, 2)
# Example usage
if __name__ == "__main__":
import asyncio
async def main():
print("Gathering daily statistics for LEGO catalog...")
stats = await get_daily_stats()
print(f"\n=== Daily Stats ===")
print(f"Total Items: {stats['total_items']}")
print(f"Enriched Items: {stats['enriched_items']}")
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}%")
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
-278
View File
@@ -1,278 +0,0 @@
"""
Production webhook server using FastAPI + Gunicorn/Uvicorn
Receives Square webhooks with signature verification
"""
import os
import json
import asyncio
from typing import Dict, Any
from fastapi import FastAPI, Request, Response, status
from fastapi.responses import JSONResponse
import dotenv
# Load environment variables
dotenv.load_dotenv()
# Square webhook validation
from square.utils.webhooks_helper import verify_signature
# Configuration
SQUARE_WEBHOOK_SIGNATURE_KEY = os.getenv("SANDBOX_SQUARE_WEBHOOK_INVENTORY_ID")
SQUARE_WEBHOOK_URL = os.getenv("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!")
# In-memory store for processed event IDs (idempotency)
processed_event_ids = set()
MAX_STORED_EVENT_IDS = 10000
# Create FastAPI app
app = FastAPI(title="Studspider")
@app.get("/")
async def root():
"""Landing page"""
return {
"service": "Studspider",
"status": "running",
"endpoints": {"webhook": "/hook (POST)", "health": "/health (GET)"},
}
@app.get("/health")
async def health_check():
"""Health check endpoint for monitoring"""
return {
"status": "healthy",
"signature_verification": (
"enabled" if SQUARE_WEBHOOK_SIGNATURE_KEY else "disabled"
),
}
@app.post("/hook")
async def webhook_handler(request: Request):
"""
Handle Square webhook events - MUST respond within 2 seconds!
Square sends POST requests with:
- x-square-hmacsha256-signature header
- JSON body with event data
- event_id for idempotency
"""
try:
# Log incoming request
print(f"\n[WEBHOOK] Incoming request from {request.client.host}")
print(f"[WEBHOOK] Method: {request.method}")
print(f"[WEBHOOK] URL: {request.url}")
print(f"[WEBHOOK] Headers: {dict(request.headers)}")
# Read raw body for signature verification
body = await request.body()
body_str = body.decode("utf-8")
print(f"[WEBHOOK] Body length: {len(body_str)} bytes")
# Get signature header
signature_header = request.headers.get("x-square-hmacsha256-signature", "")
# Verify Square webhook signature (if configured)
if SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL:
if not signature_header:
print(
f"[ERROR] Missing x-square-hmacsha256-signature header from {request.client.host}"
)
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"error": "Missing signature header"},
)
# Debug signature verification inputs
print(f"[DEBUG] Signature header: {signature_header[:20]}...")
print(f"[DEBUG] Signature key (first 10 chars): {SQUARE_WEBHOOK_SIGNATURE_KEY[:10]}...")
print(f"[DEBUG] Notification URL: {SQUARE_WEBHOOK_URL}")
print(f"[DEBUG] Body (first 100 chars): {body_str[:100]}...")
try:
is_valid = verify_signature(
request_body=body_str,
signature_header=signature_header,
signature_key=SQUARE_WEBHOOK_SIGNATURE_KEY,
notification_url=SQUARE_WEBHOOK_URL,
)
if not is_valid:
print(
f"[ERROR] Invalid webhook signature from {request.client.host}"
)
print(f"[ERROR] This usually means:")
print(f"[ERROR] 1. SQUARE_WEBHOOK_URL doesn't match Square Dashboard exactly")
print(f"[ERROR] 2. SQUARE_WEBHOOK_SIGNATURE_KEY is incorrect")
print(f"[ERROR] 3. Request body was modified in transit")
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"error": "Invalid signature"},
)
print(f"[INFO] ✓ Webhook signature verified from {request.client.host}")
except Exception as e:
print(f"[ERROR] Signature verification exception: {e}")
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"error": f"Signature verification failed: {str(e)}"},
)
else:
print(
f"[WARN] Webhook signature verification DISABLED - configure SQUARE_WEBHOOK_SIGNATURE_KEY"
)
# Parse JSON data
try:
data = json.loads(body_str)
except json.JSONDecodeError as e:
print(f"[ERROR] Invalid JSON from {request.client.host}: {e}")
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={"error": "Invalid JSON"},
)
# Check idempotency (event_id)
event_id = data.get("event_id")
if event_id:
if event_id in processed_event_ids:
print(
f"[INFO] Duplicate event {event_id} - already processed (idempotency)"
)
return JSONResponse(
status_code=status.HTTP_200_OK,
content={
"status": "success",
"message": "Event already processed",
"event_id": event_id,
},
)
# Add to processed set
processed_event_ids.add(event_id)
# Prevent memory growth
if len(processed_event_ids) > MAX_STORED_EVENT_IDS:
processed_event_ids.pop()
# Get event info for quick response
webhook_type = data.get("type", "unknown")
event_id_val = data.get("event_id", "none")
merchant_id = data.get("merchant_id", "unknown")
# RESPOND IMMEDIATELY - Square requires fast response (< 2 seconds)
response_data = {
"status": "success",
"message": "Webhook received",
"event_id": event_id_val,
}
# Process webhook in background (async task)
asyncio.create_task(
process_webhook(
webhook_type, event_id_val, merchant_id, data, request.client.host
)
)
# Return immediate response
print(f"[INFO] 200 OK response sent for event {event_id_val}")
return JSONResponse(status_code=status.HTTP_200_OK, content=response_data)
except Exception as e:
# Catch-all for any unexpected errors - still respond quickly
print(f"[ERROR] Unexpected error in webhook handler: {e}")
import traceback
traceback.print_exc()
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={"error": "Internal server error"},
)
# Return immediate response
return JSONResponse(status_code=status.HTTP_200_OK, content=response_data)
async def process_webhook(
webhook_type: str,
event_id: str,
merchant_id: str,
data: Dict[str, Any],
client_ip: str,
):
"""
Process webhook data asynchronously after responding to Square
This runs in the background and doesn't block the HTTP response
"""
try:
print("\n" + "=" * 60)
print(f"[WEBHOOK] Received at {asyncio.get_event_loop().time()}")
print(f"[WEBHOOK] From: {client_ip}")
print(f"[WEBHOOK] Type: {webhook_type}")
print(f"[WEBHOOK] Event ID: {event_id}")
print(f"[WEBHOOK] Merchant: {merchant_id}")
print(f"[WEBHOOK] Full payload:")
print(json.dumps(data, indent=2))
print("=" * 60 + "\n")
# Process based on webhook type
if webhook_type == "inventory.count.updated":
print(f"[PROCESS] Processing inventory count update...")
inventory_counts = (
data.get("data", {}).get("object", {}).get("inventory_counts", [])
)
for count in inventory_counts:
catalog_id = count.get("catalog_object_id")
quantity = count.get("quantity")
state = count.get("state")
print(
f"[PROCESS] Catalog: {catalog_id}, Qty: {quantity}, State: {state}"
)
elif webhook_type == "catalog.version.updated":
print(f"[PROCESS] Processing catalog version update...")
# Add your catalog processing logic here
else:
print(f"[PROCESS] Webhook type '{webhook_type}' - no handler configured")
print(f"[PROCESS] ✓ Webhook processing complete for {event_id}")
except Exception as e:
print(f"[ERROR] Failed to process webhook {event_id}: {e}")
if __name__ == "__main__":
# For development/testing - use uvicorn directly
import uvicorn
print("\n" + "=" * 60)
print("Studspider webhook boot dev")
print("=" * 60)
print(
f"Signature verification: {'ENABLED' if SQUARE_WEBHOOK_SIGNATURE_KEY else 'DISABLED'}"
)
print(f"Webhook URL: {SQUARE_WEBHOOK_URL or 'NOT CONFIGURED'}")
print("\nFor production, use Gunicorn:")
print(
" gunicorn webhook_server:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:7433"
)
print("=" * 60 + "\n")
uvicorn.run(app, host="0.0.0.0", port=7433, log_level="info")
+342 -11
View File
@@ -1,14 +1,345 @@
import src.test_server
import os
import json
import asyncio
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
# Square webhook validation
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
SQUARE_LEGO_CAT_ID = sq_env.LEGO_CAT_ID
class Studspider:
def __init__(self):
self.name = "Studspider"
self.version = "0.0.1"
self.description = """
studspider is an application designed to interact with the Brickset API and directly with Square.
It allows users to search for items, manage collections, and perform various operations related to bricksets.
"""
# 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!")
def run(self):
print(f"{self.name} v{self.version} is running.")
if SQUARE_WEBHOOK_URL:
print(f"[CONFIG] Webhook URL: {SQUARE_WEBHOOK_URL}")
else:
print("[CONFIG] WARNING: SQUARE_WEBHOOK_URL not set!")
# In-memory store for processed event IDs (idempotency)
processed_event_ids = set()
MAX_STORED_EVENT_IDS = 10000
# Create FastAPI app
app = FastAPI(title="Studspider")
@app.get("/")
async def root():
"""Landing page"""
return {
"service": "Studspider",
"status": "running",
"endpoints": {"webhook": "/hook (POST)", "health": "/health (GET)"},
}
@app.get("/health")
async def health_check():
"""Health check endpoint for monitoring"""
return {
"status": "healthy",
"signature_verification": (
"enabled" if SQUARE_WEBHOOK_SIGNATURE_KEY else "disabled"
),
}
@app.post("/hook")
async def webhook_handler(request: Request):
"""
Handle Square webhook events - MUST respond within 2 seconds!
Square sends POST requests with:
- x-square-hmacsha256-signature header
- JSON body with event data
- event_id for idempotency
"""
try:
# Log incoming request
print(f"\n[WEBHOOK] Incoming request from {request.client.host}")
print(f"[WEBHOOK] Method: {request.method}")
print(f"[WEBHOOK] URL: {request.url}")
print(f"[WEBHOOK] Headers: {dict(request.headers)}")
# Read raw body for signature verification
body = await request.body()
body_str = body.decode("utf-8")
print(f"[WEBHOOK] Body length: {len(body_str)} bytes")
# Get signature header
signature_header = request.headers.get("x-square-hmacsha256-signature", "")
# Verify Square webhook signature (if configured)
if SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL:
if not signature_header:
print(
f"[ERROR] Missing x-square-hmacsha256-signature header from {request.client.host}"
)
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"error": "Missing signature header"},
)
# Debug signature verification inputs
print(f"[DEBUG] Signature header: {signature_header[:20]}...")
print(
f"[DEBUG] Signature key (first 10 chars): {SQUARE_WEBHOOK_SIGNATURE_KEY[:10]}..."
)
print(f"[DEBUG] Notification URL: {SQUARE_WEBHOOK_URL}")
print(f"[DEBUG] Body (first 100 chars): {body_str[:100]}...")
try:
is_valid = verify_signature(
request_body=body_str,
signature_header=signature_header,
signature_key=SQUARE_WEBHOOK_SIGNATURE_KEY,
notification_url=SQUARE_WEBHOOK_URL,
)
if not is_valid:
print(
f"[ERROR] Invalid webhook signature from {request.client.host}"
)
print(f"[ERROR] This usually means:")
print(
f"[ERROR] 1. SQUARE_WEBHOOK_URL doesn't match Square Dashboard exactly"
)
print(f"[ERROR] 2. SQUARE_WEBHOOK_SIGNATURE_KEY is incorrect")
print(f"[ERROR] 3. Request body was modified in transit")
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"error": "Invalid signature"},
)
print(f"[INFO] ✓ Webhook signature verified from {request.client.host}")
except Exception as e:
print(f"[ERROR] Signature verification exception: {e}")
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"error": f"Signature verification failed: {str(e)}"},
)
else:
print(
f"[WARN] Webhook signature verification DISABLED - configure SQUARE_WEBHOOK_SIGNATURE_KEY"
)
# Parse JSON data
try:
data = json.loads(body_str)
except json.JSONDecodeError as e:
print(f"[ERROR] Invalid JSON from {request.client.host}: {e}")
return JSONResponse(
status_code=status.HTTP_400_BAD_REQUEST,
content={"error": "Invalid JSON"},
)
# Check idempotency (event_id)
event_id = data.get("event_id")
if event_id:
if event_id in processed_event_ids:
print(
f"[INFO] Duplicate event {event_id} - already processed (idempotency)"
)
return JSONResponse(
status_code=status.HTTP_200_OK,
content={
"status": "success",
"message": "Event already processed",
"event_id": event_id,
},
)
# Add to processed set
processed_event_ids.add(event_id)
# Prevent memory growth
if len(processed_event_ids) > MAX_STORED_EVENT_IDS:
processed_event_ids.pop()
# Get event info for quick response
webhook_type = data.get("type", "unknown")
event_id_val = data.get("event_id", "none")
merchant_id = data.get("merchant_id", "unknown")
# RESPOND IMMEDIATELY - Square requires fast response (< 2 seconds)
response_data = {
"status": "success",
"message": "Webhook received",
"event_id": event_id_val,
}
# Process webhook in background (async task)
asyncio.create_task(
process_webhook(
webhook_type, event_id_val, merchant_id, data, request.client.host
)
)
# Return immediate response
print(f"[INFO] 200 OK response sent for event {event_id_val}")
return JSONResponse(status_code=status.HTTP_200_OK, content=response_data)
except Exception as e:
# Catch-all for any unexpected errors - still respond quickly
print(f"[ERROR] Unexpected error in webhook handler: {e}")
import traceback
traceback.print_exc()
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={"error": "Internal server error"},
)
# Return immediate response
return JSONResponse(status_code=status.HTTP_200_OK, content=response_data)
async def process_webhook(
webhook_type: str,
event_id: str,
merchant_id: str,
data: Dict[str, Any],
client_ip: str,
):
"""
Process webhook data asynchronously after responding to Square
This runs in the background and doesn't block the HTTP response
"""
try:
print("\n" + "=" * 60)
print(f"[WEBHOOK] Received at {asyncio.get_event_loop().time()}")
print(f"[WEBHOOK] From: {client_ip}")
print(f"[WEBHOOK] Type: {webhook_type}")
print(f"[WEBHOOK] Event ID: {event_id}")
print(f"[WEBHOOK] Merchant: {merchant_id}")
print(f"[WEBHOOK] Full payload:")
print(json.dumps(data, indent=2))
print("=" * 60 + "\n")
# Process based on webhook type
if webhook_type == "inventory.count.updated":
print(f"[PROCESS] Processing inventory count update...")
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:
variation_id = count.get("catalog_object_id")
quantity = count.get("quantity")
state = count.get("state")
print(
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...")
# 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")
print(f"[PROCESS] ✓ Webhook processing complete for {event_id}")
except Exception as e:
print(f"[ERROR] Failed to process webhook {event_id}: {e}")
import traceback
traceback.print_exc()
if __name__ == "__main__":
# For development/testing - use uvicorn directly
import uvicorn
print("\n" + "=" * 60)
print("Studspider webhook boot dev")
print("=" * 60)
print(
f"Signature verification: {'ENABLED' if SQUARE_WEBHOOK_SIGNATURE_KEY else 'DISABLED'}"
)
print(f"Webhook URL: {SQUARE_WEBHOOK_URL or 'NOT CONFIGURED'}")
print("\nFor production, use Gunicorn:")
print(
" gunicorn conf/studspider:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:7433"
)
print("=" * 60 + "\n")
uvicorn.run(app, host="0.0.0.0", port=7433, log_level="info")
File renamed without changes.
+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):
File renamed without changes.
File renamed without changes.