64 Commits
Author SHA1 Message Date
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
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
28 changed files with 3422 additions and 1439 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 environment - "sandbox" for testing, "production" for live store
# Square API Configuration SQUARE_ENV = "sandbox" # or "production"
# ============================================================================
# Environment: "production" or "sandbox" SQUARE_WEBHOOK_URL = "https://yourdomain.com/hook"
SQUARE_ENV=sandbox
# Sandbox credentials # Sandbox credentials - test environment (fake data, safe for development)
SANDBOX_SQUARE_APP_ID=your_sandbox_app_id SANDBOX_SQUARE_APP_ID = "your_sandbox_app_id"
SANDBOX_SQUARE_TOKEN=your_sandbox_access_token SANDBOX_SQUARE_TOKEN = "your_sandbox_access_token"
SANDBOX_SQUARE_WEBHOOK_INVENTORY_ID = "your_sandbox_webhook_signature_key"
# Production credentials # Production credentials - live store (real inventory and sales)
PROD_SQUARE_APP_ID=your_production_app_id PROD_SQUARE_APP_ID = "your_production_app_id"
PROD_SQUARE_TOKEN=your_production_access_token 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) # Category ID for LEGO items in Square catalog (find in Dashboard → Items → Categories)
SQUARE_LEGO_CAT_ID=your_lego_category_id 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
+6 -5
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 ## Run me
Clone the project Clone the project
```bash ```bash
git clone http://git.traps.moe/ph0sgene/studspider.git git clone http://git.traps.moe/ph0sgene/studspider.git
git checkout uv git checkout branch
``` ```
Go to the project directory Go to the project directory
@@ -30,14 +30,15 @@ Install dependencies
Start the server Start the server
```bash ```bash
uvx run studspider.py uv run studspider.py
``` ```
## Tech ## Tech
- gunicorn
- uvicorn
- aiohttp - aiohttp
- squareup - squareup
- apscheduler - apscheduler
- aiosqlite - aiosqlite
@@ -1,10 +1,7 @@
"""
Gunicorn configuration for StudSpider webhook server
Production-ready settings for running with nginx reverse proxy
"""
import multiprocessing import multiprocessing
import os import os
import src.banner as banner
# Server socket # Server socket
bind = "127.0.0.1:7433" # nginx will proxy to this 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' 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 # Process naming
proc_name = "studspider-webhook" proc_name = "studspider"
# Server mechanics # Server mechanics
daemon = False # Run in foreground (systemd will handle daemonization) daemon = False # Run in foreground (systemd will handle daemonization)
pidfile = "./logs/webhook.pid" pidfile = "./logs/studspider.pid"
umask = 0 umask = 0
user = None # Run as current user (systemd will set user) user = None # Run as current user (systemd will set user)
group = None group = None
@@ -57,6 +54,29 @@ def on_starting(server):
"""Called just before the master process is initialized.""" """Called just before the master process is initialized."""
print("Studspider WEB starting...") 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): def on_reload(server):
"""Called to recycle workers during a reload via SIGHUP.""" """Called to recycle workers during a reload via SIGHUP."""
@@ -64,7 +84,8 @@ def on_reload(server):
def when_ready(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"STUDSPIDER HOOK v0.1 RUNNING ON {bind}")
print(f"WORKERS: {workers}") print(f"WORKERS: {workers}")
print(f"WORKING CLASS: {worker_class}\n") print(f"WORKING CLASS: {worker_class}\n")
-11
View File
@@ -1,11 +0,0 @@
{
"folders": [
{
"path": "."
},
{
"path": "../kopling"
}
],
"settings": {}
}
+126 -12
View File
@@ -7,9 +7,11 @@ import asyncio
import os import os
import aiohttp import aiohttp
import json import json
import math
import dotenv import dotenv
from typing import Dict, Any from typing import Dict, Any, Optional
import throttler from datetime import datetime, timedelta
from . import throttler
Throttler = throttler.GlobalThrottler Throttler = throttler.GlobalThrottler
@@ -29,6 +31,104 @@ class AsyncBricksetAPI:
self.user_hash = None self.user_hash = None
self.logger = logger 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]: async def login(self) -> Dict[str, Any]:
"""Asynchronously authenticate with Brickset API""" """Asynchronously authenticate with Brickset API"""
url = "https://brickset.com/api/v3.asmx/login" url = "https://brickset.com/api/v3.asmx/login"
@@ -64,7 +164,17 @@ class AsyncBricksetAPI:
Returns: Returns:
Dict containing set information 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: if not self.user_hash:
await self.login() await self.login()
@@ -101,26 +211,30 @@ class AsyncBricksetAPI:
) )
set_data = result["sets"][0] 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 = { our_set = {
"set_num": set_data["number"], "set_num": set_data["number"],
"set_name": set_data["name"], "set_name": set_data["name"],
"set_theme": set_data["theme"], "set_theme": set_data["theme"],
"set_image": set_data["image"]["thumbnailURL"], "set_image": set_data["image"]["thumbnailURL"],
"set_barcode": set_data["barcode"]["UPC"], "set_barcode": upc_code,
"set_retailprice": round( "set_retailprice": round(retail_price),
set_data["LEGOCom"]["US"]["retailPrice"]
),
} }
# Calculate adjusted price (10% markup) # Calculate adjusted price (10% markup), then round up
retail_price = our_set["set_retailprice"] adjusted_price = retail_price * 1.10
our_set["set_adjustredprice"] = retail_price + ( our_set["set_adjustredprice"] = math.ceil(adjusted_price)
(10 / 100) * retail_price
)
# Set PW inventory name with theme # Set PW inventory name with theme
our_set["pw_name"] = ( 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 # Flag for manual price review if needed
+434 -17
View File
@@ -4,11 +4,15 @@ import json
import asyncio import asyncio
import uuid import uuid
import re import re
import traceback
from time import time 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: def get_set_number(text: str) -> int:
@@ -51,33 +55,446 @@ async def get_itm(sku: str):
return {"name": name, "id": itm_id} 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. Asynchronously updates an item in the Square catalog based on the provided LEGO set information.
Args: Args:
lego_set (dict): A dictionary containing LEGO set details. Expected keys include: item_id (str): Square catalog item ID to update
- "sets": List of set dictionaries, each with "name", "number", "barcode" (with "UPC"), and "image" (with "thumbnailURL"). lego_set (dict): A dictionary containing LEGO set details from abricksetapi. Expected keys include:
- "pw_name": The name associated with the LEGO set in another system. - "set_num": LEGO set number (e.g., "75192")
- "set_adjustredprice": The adjusted price for the set. - "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: Returns:
bool: True if the update operation was successful. bool: True if the update operation was successful.
Raises: Raises:
KeyError: If expected keys are missing from the input dictionary. KeyError: If expected keys are missing from the input dictionary.
Exception: For errors during Square API interaction. 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"] # Convert price to cents for Square API
set_number = lego_set["sets"][0]["number"] price_in_cents = int(set_adjustedprice * 100)
set_upc = lego_set["sets"][0]["barcode"]["UPC"]
set_image = lego_set["sets"][0]["image"]["thumbnailURL"]
pw_name = lego_set["pw_name"]
set_adjustedprice = lego_set["set_adjustredprice"]
result = await client.catalog.search_items(text_filter=set_number, limit=1) # Retry logic for version conflicts (race conditions with concurrent updates)
itm_id = json.loads(result.model_dump_json())["items"][0]["id"] max_retries = 3
retry_delay = 0.5 # seconds
# await sq.catalog.object.upsert() for attempt in range(max_retries):
return True 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}")
traceback.print_exc()
return False
async def acat2id(cat_id: str): async def acat2id(cat_id: str):
+5 -8
View File
@@ -6,20 +6,17 @@ dotenv.load_dotenv()
LEGO_CAT_ID = os.getenv("SQUARE_LEGO_CAT_ID") 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") SQUARE_WEBHOOK_URL = os.getenv("SQUARE_WEBHOOK_URL")
if os.getenv("SQUARE_ENV") == "production": if os.getenv("SQUARE_ENV") == "production":
SQUARE_APP_ID = os.getenv("PROD_SQUARE_APP_ID") SQUARE_APP_ID = os.getenv("PROD_SQUARE_APP_ID")
SQUARE_TOKEN = os.getenv("PROD_SQUARE_TOKEN") 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 environment = SquareEnvironment.PRODUCTION
else: else:
SQUARE_APP_ID = os.getenv("SANDBOX_SQUARE_APP_ID") SQUARE_APP_ID = os.getenv("SANDBOX_SQUARE_APP_ID")
SQUARE_TOKEN = os.getenv("SANDBOX_SQUARE_TOKEN") SQUARE_TOKEN = os.getenv("SANDBOX_SQUARE_TOKEN")
environment = SquareEnvironment.SANDBOX 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
+196 -26
View File
@@ -8,34 +8,204 @@ from time import time
from . import asquare from . import asquare
from . import db from . import db
from . import abricksetapi from . import abricksetapi
from . import throttler from . import throttler
Throttler = throttler.GlobalThrottler Throttler = throttler.GlobalThrottler
LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID
# big glue boy
async def do_review() -> bool: async def enrich_single_item(item_id: str, item_data: dict = None, logger=None) -> bool:
result = await asquare.acat2id(LEGO_CAT_ID) # type: ignore r"""
for item in result: Enrich a single Square catalog item with Brickset data.
if not re.match(r".*\d{5}.*", item["name"]):
missing = [] Flow:
print(f"{item} is missing its set number!") 1. Get item details from Square (by item_id) - skip if item_data provided
#print(f"[@DEBUG] Upserting {item['id']} into REVIEW table") 2. Extract 5-digit set number from name (regex \d{5})
item_sanitized = { 3. Fetch Brickset data (with throttling)
"sq_object": item["id"], 4. Update Square item with enriched data
"set_name": item["name"], 5. Handle errors (no set number → MISSING table)
"set_number": None,
"time_added": time Args:
} item_id: Square catalog item ID
missing.append(item_sanitized) item_data: Optional pre-fetched item data (avoids redundant API call)
logger: Optional logger instance for debug output
for item in missing:
print(item) Returns:
# sq_item = await db.fetch(item["id"]) True if successful, False otherwise
if item["id"] != sq_item["sq_object"]: """
print(f"[@DEBUG] {item['id']} is missing)") try:
#await db.upsert("MISSING", item_sanitized, "sq_object") if logger:
# #print(f"[@DEBUG] Upserted {item['id']} into REVIEW table") logger.info(f"Starting enrichment for item {item_id}")
#print(f"[@DEBUG] {item['id']} already in REVIEW table") else:
return True 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()
+120 -120
View File
@@ -1,120 +1,120 @@
import aiohttp import aiohttp
import json import json
import dotenv import dotenv
import os import os
import importlib.metadata import importlib.metadata
dotenv.load_dotenv() dotenv.load_dotenv()
api_key = os.getenv("BRICKOWL") api_key = os.getenv("BRICKOWL")
abrickowl_version = importlib.metadata.version("abrickowlapi") abrickowl_version = importlib.metadata.version("abrickowlapi")
class ABrickowlAPI: class ABrickowlAPI:
def __init__(self, value): def __init__(self, value):
self.user_agent = f"abrickowlapi/{abrickowl_version}" self.user_agent = f"abrickowlapi/{abrickowl_version}"
self.headers = {"content-type": "application/x-www-form-urlencoded"} self.headers = {"content-type": "application/x-www-form-urlencoded"}
self.base_url = "https://api.brickowl.com/v1/" self.base_url = "https://api.brickowl.com/v1/"
self.session = aiohttp.ClientSession() self.session = aiohttp.ClientSession()
return self return self
def set_valid(self, set_id: int): def set_valid(self, set_id: int):
""" """
Verify if a LEGO set ID is valid. Verify if a LEGO set ID is valid.
""" """
if len(set_id) == 5: if len(set_id) == 5:
return True return True
else: else:
return False return False
@classmethod @classmethod
async def catalog_list(self): async def catalog_list(self):
""" """
Fetch the list of LEGO sets from the Brickowl API. Fetch the list of LEGO sets from the Brickowl API.
""" """
async with self.session.get( async with self.session.get(
f"{self.base_url}catalog/list/view?key={api_key}", f"{self.base_url}catalog/list/view?key={api_key}",
headers=self.headers, headers=self.headers,
user_agent=self.user_agent, user_agent=self.user_agent,
) as response: ) as response:
if response.status == 200: if response.status == 200:
return await response.json() return await response.json()
else: else:
raise Exception(f"Error fetching list, {response.status}") raise Exception(f"Error fetching list, {response.status}")
@classmethod @classmethod
async def catalog_lookup(self, set_id: int): async def catalog_lookup(self, set_id: int):
""" """
Fetch details from the brickowl API for a specific LEGO set. Fetch details from the brickowl API for a specific LEGO set.
set_id: The LEGO set ID to look up. set_id: The LEGO set ID to look up.
returns JSON response containing set details. returns JSON response containing set details.
""" """
async with self.session.get( async with self.session.get(
f"{self.base_url}catalog/lookup/{set_id}view?key={api_key}", f"{self.base_url}catalog/lookup/{set_id}view?key={api_key}",
headers=self.headers, headers=self.headers,
user_agent=self.user_agent, user_agent=self.user_agent,
) as response: ) as response:
if response.status == 200: if response.status == 200:
return await response.json() return await response.json()
else: else:
raise Exception(f"Error fetching set {set_id}, {response.status}") raise Exception(f"Error fetching set {set_id}, {response.status}")
@classmethod @classmethod
async def catalog_availability(self, set_id: int): async def catalog_availability(self, set_id: int):
""" """
Fetch availability information for a specific LEGO set. Fetch availability information for a specific LEGO set.
""" """
async with self.session.get( async with self.session.get(
f"{self.base_url}catalog/availability/{set_id}view?key={api_key}", f"{self.base_url}catalog/availability/{set_id}view?key={api_key}",
headers=self.headers, headers=self.headers,
user_agent=self.user_agent, user_agent=self.user_agent,
) as response: ) as response:
if response.status == 200: if response.status == 200:
return await response.json() return await response.json()
else: else:
raise Exception( raise Exception(
f"Error fetching fetching availability for {set_id}, {response.status}" f"Error fetching fetching availability for {set_id}, {response.status}"
) )
@classmethod @classmethod
async def catalog_id_lookup(self, set_id: int): async def catalog_id_lookup(self, set_id: int):
async with self.session.get( async with self.session.get(
f"{self.base_url}catalog/id_lookup/{set_id}view?key={api_key}", f"{self.base_url}catalog/id_lookup/{set_id}view?key={api_key}",
headers=self.headers, headers=self.headers,
user_agent=self.user_agent, user_agent=self.user_agent,
) as response: ) as response:
if response.status == 200: if response.status == 200:
return await response.json() return await response.json()
else: else:
raise Exception(f"ID lookup failed for set {set_id}, {response.status}") raise Exception(f"ID lookup failed for set {set_id}, {response.status}")
@classmethod @classmethod
async def catalog_bulk(self): async def catalog_bulk(self):
async with self.session.get( async with self.session.get(
f"{self.base_url}catalog/bulk/view?key={api_key}", f"{self.base_url}catalog/bulk/view?key={api_key}",
headers=self.headers, headers=self.headers,
user_agent=self.user_agent, user_agent=self.user_agent,
) as response: ) as response:
if response.status == 200: if response.status == 200:
return await response.json() return await response.json()
else: else:
raise Exception(f"Error fetching bulk, {response.status}") raise Exception(f"Error fetching bulk, {response.status}")
@classmethod @classmethod
async def catalog_search(self, query: str): async def catalog_search(self, query: str):
async with self.session.get( async with self.session.get(
f"{self.base_url}catalog/search/{query}view?key={api_key}", f"{self.base_url}catalog/search/{query}view?key={api_key}",
headers=self.headers, headers=self.headers,
user_agent=self.user_agent, user_agent=self.user_agent,
) as response: ) as response:
if response.status == 200: if response.status == 200:
return await response.json() return await response.json()
else: else:
raise Exception(f"Search term {set_id} failed, {response.status}") raise Exception(f"Search term {set_id} failed, {response.status}")
# Usage: # Usage:
# obj = await ABrickowlAPI.catalog_lookup(value) # obj = await ABrickowlAPI.catalog_lookup(value)
File renamed without changes.
+116 -116
View File
@@ -1,116 +1,116 @@
import os import os
import requests import requests
import json import json
import dotenv import dotenv
from typing import Dict, Any from typing import Dict, Any
dotenv.load_dotenv() dotenv.load_dotenv()
class BricksetAPI: class BricksetAPI:
"""Class for interacting with the Brickset API""" """Class for interacting with the Brickset API"""
def __init__(self, debug: bool = False): def __init__(self, debug: bool = False):
self.username = os.getenv("BRICKSET_USERNAME") self.username = os.getenv("BRICKSET_USERNAME")
self.password = os.getenv("BRICKSET_PASS") self.password = os.getenv("BRICKSET_PASS")
self.api_key = os.getenv("BRICKSET_API_KEY") self.api_key = os.getenv("BRICKSET_API_KEY")
self.debug = debug self.debug = debug
self.headers = {"UserAgent": "studspider/0.1"} self.headers = {"UserAgent": "studspider/0.1"}
self.user_hash = None self.user_hash = None
def login(self) -> Dict[str, Any]: def login(self) -> Dict[str, Any]:
"""Login to Brickset API and get user hash""" """Login to Brickset API and get user hash"""
url = "https://brickset.com/api/v3.asmx/login" url = "https://brickset.com/api/v3.asmx/login"
params = { params = {
"username": self.username, "username": self.username,
"password": self.password, "password": self.password,
"apiKey": self.api_key, "apiKey": self.api_key,
} }
response = requests.get(url, params=params, headers=self.headers) response = requests.get(url, params=params, headers=self.headers)
if self.debug: if self.debug:
print(f"[@DEBUG] Login response: {response.status_code}") print(f"[@DEBUG] Login response: {response.status_code}")
print(f"[@DEBUG] Response content: {response.text}") print(f"[@DEBUG] Response content: {response.text}")
try: try:
data = response.json() data = response.json()
self.user_hash = data.get("hash") self.user_hash = data.get("hash")
if not self.user_hash: if not self.user_hash:
raise ValueError("No hash received in login response") raise ValueError("No hash received in login response")
return data return data
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
print(f"[@ERROR] Failed to decode JSON response: {e}") print(f"[@ERROR] Failed to decode JSON response: {e}")
print(f"[@DEBUG] Raw response: {response.text}") print(f"[@DEBUG] Raw response: {response.text}")
raise raise
def get_set(self, set_number: str) -> Dict[str, Any]: def get_set(self, set_number: str) -> Dict[str, Any]:
"""Get LEGO set information by set number""" """Get LEGO set information by set number"""
if not self.user_hash: if not self.user_hash:
self.login() self.login()
url = "https://brickset.com/api/v3.asmx/getSets" url = "https://brickset.com/api/v3.asmx/getSets"
params = { params = {
"apiKey": self.api_key, "apiKey": self.api_key,
"userHash": self.user_hash, "userHash": self.user_hash,
"params": json.dumps({"setNumber": f"{set_number}-1"}), "params": json.dumps({"setNumber": f"{set_number}-1"}),
} }
response = requests.get(url, params=params, headers=self.headers) response = requests.get(url, params=params, headers=self.headers)
if self.debug: if self.debug:
print(f"[@DEBUG] Get set response: {response.status_code}") print(f"[@DEBUG] Get set response: {response.status_code}")
print(f"[@DEBUG] Response content: {response.text}") print(f"[@DEBUG] Response content: {response.text}")
if response.status_code == 200: if response.status_code == 200:
try: try:
data = response.json() data = response.json()
if not data.get("sets"): if not data.get("sets"):
raise ValueError(f"No set data found for {set_number}") raise ValueError(f"No set data found for {set_number}")
set_data = data["sets"][0] set_data = data["sets"][0]
our_set = { our_set = {
"set_num": set_data["number"], "set_num": set_data["number"],
"set_name": set_data["name"], "set_name": set_data["name"],
"set_image": set_data["image"]["thumbnailURL"], "set_image": set_data["image"]["thumbnailURL"],
"set_barcode": set_data["barcode"]["UPC"], "set_barcode": set_data["barcode"]["UPC"],
"set_retailprice": round(set_data["LEGOCom"]["US"]["retailPrice"]), "set_retailprice": round(set_data["LEGOCom"]["US"]["retailPrice"]),
} }
# Calculate adjusted price (10% markup) # Calculate adjusted price (10% markup)
retail_price = our_set["set_retailprice"] retail_price = our_set["set_retailprice"]
our_set["set_adjustredprice"] = retail_price + ( our_set["set_adjustredprice"] = retail_price + (
(10 / 100) * retail_price (10 / 100) * retail_price
) )
# Set Playwell inventory name # Set Playwell inventory name
if our_set["theme"] == "Duplo": if our_set["theme"] == "Duplo":
our_set["pw_name"] = ( our_set["pw_name"] = (
f"{our_set['set_num']} - DUPLO {our_set['set_name']}" f"{our_set['set_num']} - DUPLO {our_set['set_name']}"
) )
else: else:
our_set["pw_name"] = f"{our_set['set_num']} - {our_set['set_name']}" our_set["pw_name"] = f"{our_set['set_num']} - {our_set['set_name']}"
return our_set return our_set
except json.JSONDecodeError as e: except json.JSONDecodeError as e:
# print(f"[@ERROR] Failed to decode JSON response: {e}") # print(f"[@ERROR] Failed to decode JSON response: {e}")
# print(f"[@DEBUG] Raw response: {response.text}") # print(f"[@DEBUG] Raw response: {response.text}")
raise raise
except KeyError as e: except KeyError as e:
# print(f"[@ERROR] Missing expected field in response: {e}") # print(f"[@ERROR] Missing expected field in response: {e}")
# print(f"[@DEBUG] Response data: {data}") # print(f"[@DEBUG] Response data: {data}")
raise raise
elif response.status_code == 500: elif response.status_code == 500:
raise Exception( raise Exception(
"Internal Server Error: Please check API key and parameters." "Internal Server Error: Please check API key and parameters."
) )
else: else:
raise Exception(f"Unexpected response: {response.status_code}") raise Exception(f"Unexpected response: {response.status_code}")
def example(): def example():
client = BricksetAPI(debug=True) client = BricksetAPI(debug=True)
try: try:
set_info = client.get_set("10931") set_info = client.get_set("10931")
print(set_info) print(set_info)
except Exception as e: except Exception as e:
print(f"Error: {e}") print(f"Error: {e}")
File renamed without changes.
+113 -113
View File
@@ -1,113 +1,113 @@
import asyncio import asyncio
import aiohttp.web import aiohttp.web
import uuid import uuid
import os import os
from base64 import b64decode from base64 import b64decode
import dotenv import dotenv
dotenv.load_dotenv() dotenv.load_dotenv()
SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID") SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
from asyncio_throttle.throttler import Throttler from asyncio_throttle.throttler import Throttler
import asquare as sq import asquare as sq
import abricksetapi as alego import abricksetapi as alego
def valid_uuid(val): def valid_uuid(val):
try: try:
uuid.UUID(str(val)) uuid.UUID(str(val))
return True return True
except ValueError: except ValueError:
return False return False
async def fullsync(): async def fullsync():
"""' """'
full_db = await sq.acat2id(SQUARE_LEGO_CAT_ID) full_db = await sq.acat2id(SQUARE_LEGO_CAT_ID)
for l_set in full_db: for l_set in full_db:
l_setinfo = await alego.throttled_get_set(l_set) l_setinfo = await alego.throttled_get_set(l_set)
await sq.update_itm(l_set, l_setinfo)""" await sq.update_itm(l_set, l_setinfo)"""
async def stop(): async def stop():
pass pass
async def priceonly(): async def priceonly():
pass pass
async def handle_post(request): async def handle_post(request):
valid_cmd = ["RUNFULLSYNC", "STOP", "RUNPRICEONLY"] valid_cmd = ["RUNFULLSYNC", "STOP", "RUNPRICEONLY"]
async def handle_cmd(cmd: str): async def handle_cmd(cmd: str):
if cmd not in valid_cmd: if cmd not in valid_cmd:
return False return False
else: else:
return True return True
idp_key = request.headers.get("Idempotency-Key") idp_key = request.headers.get("Idempotency-Key")
user_agent = request.headers.get("User-Agent") user_agent = request.headers.get("User-Agent")
if not idp_key and not valid_uuid(idp_key) and user_agent != "Kopling/0.1": if not idp_key and not valid_uuid(idp_key) and user_agent != "Kopling/0.1":
# 412 Precondition Failed # 412 Precondition Failed
# The server does not meet one of the preconditions that the requester put on the request header fields. # The server does not meet one of the preconditions that the requester put on the request header fields.
return aiohttp.web.Response(status=412, text="412 Headers rejected!") return aiohttp.web.Response(status=412, text="412 Headers rejected!")
else: else:
data = await request.post() data = await request.post()
reqdata = data.get(b64decode("CMD")) reqdata = data.get(b64decode("CMD"))
if await handle_cmd(reqdata): if await handle_cmd(reqdata):
match reqdata: match reqdata:
case "RUNFULLSYNC": case "RUNFULLSYNC":
print("Running full sync...") print("Running full sync...")
await fullsync() await fullsync()
return True return True
case "STOP": case "STOP":
await stop() await stop()
return True return True
case "RUNPRICEONLY": case "RUNPRICEONLY":
await priceonly() await priceonly()
return True return True
else: else:
return aiohttp.web.Response(status=400, text="400 Bad Request.") return aiohttp.web.Response(status=400, text="400 Bad Request.")
result = {"message": f"studspider/0.1 received command: {reqdata}"} result = {"message": f"studspider/0.1 received command: {reqdata}"}
return aiohttp.web.json_response(status=200, data=result) return aiohttp.web.json_response(status=200, data=result)
app = aiohttp.web.Application() app = aiohttp.web.Application()
app.router.add_post("/", handle_post) app.router.add_post("/", handle_post)
import subprocess import subprocess
def generate_toilet_banner(text, font="smblock", filter_effect=None, colors=None): def generate_toilet_banner(text, font="smblock", filter_effect=None, colors=None):
command = ["toilet", "-f", font] command = ["toilet", "-f", font]
if filter_effect: if filter_effect:
command.extend(["-F", filter_effect]) command.extend(["-F", filter_effect])
if colors: if colors:
# Toilet uses a specific format for color arguments, e.g., --gay or --metal # Toilet uses a specific format for color arguments, e.g., --gay or --metal
# For general colors, you might need to use specific filters or direct color codes. # For general colors, you might need to use specific filters or direct color codes.
# This example focuses on filters that inherently use colors. # This example focuses on filters that inherently use colors.
pass # More complex color handling would be needed for arbitrary colors pass # More complex color handling would be needed for arbitrary colors
command.append(text) command.append(text)
try: try:
result = subprocess.run(command, capture_output=True, text=True, check=True) result = subprocess.run(command, capture_output=True, text=True, check=True)
return result.stdout return result.stdout
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
return f"Error executing toilet: {e}\n{e.stderr}" return f"Error executing toilet: {e}\n{e.stderr}"
async def main(): async def main():
throttler = Throttler(rate_limit=500, period=60) # 500 requests per minute throttler = Throttler(rate_limit=500, period=60) # 500 requests per minute
if __name__ == "__main__": if __name__ == "__main__":
print(generate_toilet_banner("studspider v0.1")) print(generate_toilet_banner("studspider v0.1"))
print("Built for PLAY-WELL TEKNOLOGIES by contact@traps.moe\n\n") print("Built for PLAY-WELL TEKNOLOGIES by contact@traps.moe\n\n")
aiohttp.web.run_app(app, port=7432) aiohttp.web.run_app(app, port=7432)
+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())
@@ -1,65 +1,70 @@
import os import os
import requests import requests
import json import json
import dotenv import dotenv
dotenv.load_dotenv() dotenv.load_dotenv()
USERNAME = os.getenv("BRICKSET_USERNAME") USERNAME = os.getenv("BRICKSET_USERNAME")
PASS = os.getenv("BRICKSET_PASS") PASS = os.getenv("BRICKSET_PASS")
BRICKSET_API_KEY = os.getenv("BRICKSET_API_KEY") BRICKSET_API_KEY = os.getenv("BRICKSET_API_KEY")
debug = False # Set to True to enable debug output debug = False # Set to True to enable debug output
def login(username, password): def login(username, password):
url = "https://brickset.com/api/v3.asmx/login" url = "https://brickset.com/api/v3.asmx/login"
headers = {"UserAgent": "studspider/0.1"} headers = {"UserAgent": "studspider/0.1"}
params = { params = {
"UserAgent": "studspider/0.1", "UserAgent": "studspider/0.1",
"username": username, "username": username,
"password": password, "password": password,
"apiKey": BRICKSET_API_KEY, "apiKey": BRICKSET_API_KEY,
} }
response = requests.get(url, params=params, headers=headers) response = requests.get(url, params=params, headers=headers)
if debug: if debug:
print(f"[@DEBUG] Login response: {response.status_code} - {response.text}\n") print(f"[@DEBUG] Login response: {response.status_code} - {response.text}\n")
return response.json() return response.json()
def get_set(set_number: int) -> dict: def get_set(set_number: int) -> dict:
login_response = login(USERNAME, PASS) login_response = login(USERNAME, PASS)
url = f"https://brickset.com/api/v3.asmx/getSets?apiKey={BRICKSET_API_KEY}&userHash={login_response["hash"]}&params={{'setNumber' : '{set_number}-1'}}" url = f"https://brickset.com/api/v3.asmx/getSets?apiKey={BRICKSET_API_KEY}&userHash={login_response["hash"]}&params={{'setNumber' : '{set_number}-1'}}"
response = requests.get(url) response = requests.get(url)
if response.status_code == 200: if response.status_code == 200:
data = response.json() data = response.json()
if not data["sets"] and data["message"] == "API limit exceeded": if not data["sets"] and data["message"] == "API limit exceeded":
raise Exception("API limit exceeded. Please try again later.") raise Exception("API limit exceeded. Please try again later.")
return False return False
print(f"[@DEBUG]\t{data}") print(f"[@DEBUG]\t{data}")
our_set = {
"set_num": data["sets"][0]["number"], # Try to get UPC barcode, fall back to EAN if not available
"set_name": data["sets"][0]["name"], barcode_data = data["sets"][0].get("barcode", {})
"set_theme": data["sets"][0]["theme"], upc_code = barcode_data.get("UPC") or barcode_data.get("EAN")
"set_image": data["sets"][0]["image"]["thumbnailURL"],
"set_barcode": data["sets"][0]["barcode"]["UPC"], our_set = {
"set_retailprice": round(data["sets"][0]["LEGOCom"]["US"]["retailPrice"]), "set_num": data["sets"][0]["number"],
} "set_name": data["sets"][0]["name"],
# add PW 10% b4 tax "set_theme": data["sets"][0]["theme"],
our_set.update( "set_image": data["sets"][0]["image"]["thumbnailURL"],
{"set_adjustredprice": our_set + ((10 / 100) * our_set["set_retailprice"])} "set_barcode": upc_code,
) "set_retailprice": round(data["sets"][0]["LEGOCom"]["US"]["retailPrice"]),
our_set["pw_name"] = ( }
f"{our_set['set_num']} - {our_set["set_theme"].upper()} {our_set['set_name']}" # add PW 10% b4 tax
) our_set.update(
if our_set["set_retailprice"] >= 110: {"set_adjustredprice": our_set + ((10 / 100) * our_set["set_retailprice"])}
our_set["requires_manual_price"] = True )
return our_set our_set["pw_name"] = (
f"{our_set['set_num']} - {our_set["set_theme"].upper()} - {our_set['set_name']}"
elif response.status_code == 500: )
raise Exception( if our_set["set_retailprice"] >= 110:
"Internal Server Error: Please check the API key and parameters." our_set["requires_manual_price"] = True
) return our_set
else:
raise Exception("Unexpected response from the API.") elif response.status_code == 500:
raise Exception(
"Internal Server Error: Please check the API key and parameters."
)
else:
raise Exception("Unexpected response from the API.")
@@ -1,66 +1,66 @@
import json import json
import get_sq_env import get_sq_env
SQUARE_TOKEN = get_sq_env.SQUARE_TOKEN SQUARE_TOKEN = get_sq_env.SQUARE_TOKEN
LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID
ENVIRONMENT = get_sq_env.environment ENVIRONMENT = get_sq_env.environment
from square import Square from square import Square
from square.environment import SquareEnvironment from square.environment import SquareEnvironment
def get_square_client(): def get_square_client():
"""Initialize and return a Square client.""" """Initialize and return a Square client."""
if not SQUARE_TOKEN: if not SQUARE_TOKEN:
raise ValueError("SQUARE_TOKEN environment variable is not set.") raise ValueError("SQUARE_TOKEN environment variable is not set.")
square = Square(token=SQUARE_TOKEN, environment=ENVIRONMENT) # type: ignore square = Square(token=SQUARE_TOKEN, environment=ENVIRONMENT) # type: ignore
return square return square
client = get_square_client() client = get_square_client()
def get_itm(item_id: str): def get_itm(item_id: str):
"""Fetch an item from Square by its ID.""" """Fetch an item from Square by its ID."""
result = client.catalog.search_items(text_filter=item_id, limit=1) result = client.catalog.search_items(text_filter=item_id, limit=1)
result_json = json.loads(result.model_dump_json()) result_json = json.loads(result.model_dump_json())
ids = { ids = {
"item_id": result_json["items"][0]["id"], "item_id": result_json["items"][0]["id"],
"vari_id": result_json["items"][0]["item_data"]["variations"][0][ "vari_id": result_json["items"][0]["item_data"]["variations"][0][
"item_variation_data" "item_variation_data"
]["item_id"], ]["item_id"],
} }
item = { item = {
"set_name": result_json["items"][0]["item_data"]["name"], "set_name": result_json["items"][0]["item_data"]["name"],
"set_price": result_json["items"][0]["item_data"]["variations"][0][ "set_price": result_json["items"][0]["item_data"]["variations"][0][
"item_variation_data" "item_variation_data"
]["price_money"], ]["price_money"],
"set_barcode": result_json["items"][0]["item_data"]["variations"][0][ "set_barcode": result_json["items"][0]["item_data"]["variations"][0][
"item_variation_data" "item_variation_data"
]["sku"], ]["sku"],
} }
return [ids, item] return [ids, item]
def cat2id(cat_id: str): def cat2id(cat_id: str):
"""Return a list of item IDs within the specified category ID.""" """Return a list of item IDs within the specified category ID."""
result = client.catalog.search( result = client.catalog.search(
object_types=["ITEM"], object_types=["ITEM"],
query={ query={
"exact_query": { "exact_query": {
"attribute_name": "category_id", "attribute_name": "category_id",
"attribute_value": cat_id, "attribute_value": cat_id,
} }
}, },
) )
result_json = json.loads(result.model_dump_json()) result_json = json.loads(result.model_dump_json())
item_ids = [obj["id"] for obj in result_json.get("objects", [])] item_ids = [obj["id"] for obj in result_json.get("objects", [])]
return item_ids return item_ids
def test() -> None: def test() -> None:
# all_lego_sets = cat2id(SQUARE_LEGO_CAT_ID) # type: ignore # all_lego_sets = cat2id(SQUARE_LEGO_CAT_ID) # type: ignore
# result = get_itm("71433") # result = get_itm("71433")
# print(result) # print(result)
pass pass
+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())
+16 -12
View File
@@ -1,33 +1,37 @@
import asyncio import asyncio
from asyncio_throttle import Throttler as AsyncThrottler from asyncio_throttle import Throttler as AsyncThrottler
class GlobalThrottler: class GlobalThrottler:
_instance = None _instance = None
_lock = asyncio.Lock() _lock = asyncio.Lock()
def __init__(self, rate_limit: int = 500, period: int = 60) -> None: 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.rate_limit = rate_limit
self.period = period 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 self._initialized = True
@classmethod @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: async with cls._lock:
if not cls._instance: if not cls._instance:
cls._instance = cls(rate_limit, period) cls._instance = cls(rate_limit, period)
return cls._instance return cls._instance
async def acquire(self): async def acquire(self):
await self.throttler.acquire() await self.throttler.acquire()
def release(self):
self.throttler.release()
async def __aenter__(self): async def __aenter__(self):
await self.acquire() await self.acquire()
return self return self
async def __aexit__(self, exc_type, exc_val, exc_tb): 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")
+377 -11
View File
@@ -1,14 +1,380 @@
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
import src.stats as stats
# 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: # Debug: Print configuration (redact key)
def __init__(self): if SQUARE_WEBHOOK_SIGNATURE_KEY:
self.name = "Studspider" print(f"[CONFIG] Signature key loaded: {SQUARE_WEBHOOK_SIGNATURE_KEY[:10]}...")
self.version = "0.0.1" else:
self.description = """ print("[CONFIG] WARNING: SQUARE_WEBHOOK_SIGNATURE_KEY not set!")
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.
"""
def run(self): if SQUARE_WEBHOOK_URL:
print(f"{self.name} v{self.version} is running.") 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)",
"stats": "/stats (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.get("/stats")
async def get_stats():
"""
Get daily statistics for LEGO inventory.
Returns metrics about enrichment progress, missing data, and catalog value.
"""
try:
daily_stats = await stats.get_daily_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",
"stats": daily_stats,
"completion_rate": round(completion_rate, 1),
}
except Exception as e:
return JSONResponse(
status_code=500,
content={"status": "error", "message": str(e)},
)
@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")
+2 -2
View File
@@ -1,6 +1,6 @@
import src.libsquare as sq import src.libsquare as sq
import src.bricksetapi as lego import src.bricksetapi as lego
oursq = sq.get_itm("71417") oursq = sq.get_itm("10291")
ourlego = lego.get_set(71417) ourlego = lego.get_set(10291)
print(f"{oursq}\n\n{ourlego}\n\n") print(f"{oursq}\n\n{ourlego}\n\n")
+39 -39
View File
@@ -1,39 +1,39 @@
import asyncio import asyncio
import aiohttp import aiohttp
from uuid import uuid4 from uuid import uuid4
from base64 import b64encode from base64 import b64encode
import os import os
async def idp_request(cmd: str): async def idp_request(cmd: str):
headers = { headers = {
"Idempotency-Key": f"{uuid4()}", "Idempotency-Key": f"{uuid4()}",
"Content-Type": "application/json", "Content-Type": "application/json",
"User-Agent": "Kopling/0.12", "User-Agent": "Kopling/0.12",
} }
print(f"[@DEBUG] Idempotency-Key: {headers['Idempotency-Key']}") print(f"[@DEBUG] Idempotency-Key: {headers['Idempotency-Key']}")
enc_cmd = b64encode(f"{cmd}".encode("utf-8")).decode("utf-8") enc_cmd = b64encode(f"{cmd}".encode("utf-8")).decode("utf-8")
print("[@DEBUG] Encoded command:", enc_cmd) print("[@DEBUG] Encoded command:", enc_cmd)
print("[@DEBUG] Sending request..") print("[@DEBUG] Sending request..")
try: try:
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
async with session.post( async with session.post(
"http://localhost:7432", # Added http:// protocol "http://localhost:7432", # Added http:// protocol
headers=headers, headers=headers,
json={"CMD": enc_cmd}, # Changed to json instead of data json={"CMD": enc_cmd}, # Changed to json instead of data
) as resp: ) as resp:
print(f"[@DEBUG] Response status: {resp.status}") print(f"[@DEBUG] Response status: {resp.status}")
print(f"[@DEBUG] Response body: {await resp.text()}") print(f"[@DEBUG] Response body: {await resp.text()}")
except Exception as e: except Exception as e:
print(f"\nRequest failed: {e}") print(f"\nRequest failed: {e}")
os._exit(-1) os._exit(-1)
async def main(): async def main():
# Await the task instead of just creating it # Await the task instead of just creating it
await idp_request("RUNUPDATE") await idp_request("RUNUPDATE")
if __name__ == "__main__": if __name__ == "__main__":
asyncio.run(main()) asyncio.run(main())
+438 -452
View File
@@ -1,452 +1,438 @@
import argparse import argparse
import asyncio import asyncio
import json import json
import ssl import ssl
from aiohttp import web from aiohttp import web
from base64 import b64decode from base64 import b64decode
from typing import Dict, Any from typing import Dict, Any
import uuid import uuid
import dotenv import dotenv
import os import os
import re import re
import logging import logging
from logging.handlers import RotatingFileHandler from logging.handlers import RotatingFileHandler
from pathlib import Path from pathlib import Path
import banner import banner
import get_sq_env import get_sq_env
from time import time from time import time
time = int(time()) time = int(time())
import asquare as sq import asquare as sq
# Square SDK webhook validation # Square SDK webhook validation
from square.utils.webhooks_helper import verify_signature from square.utils.webhooks_helper import verify_signature
# from asquareclass import AsyncSquareClient # type: ignore # from asquareclass import AsyncSquareClient # type: ignore
from abricksetapi import AsyncBricksetAPI from abricksetapi import AsyncBricksetAPI
print("Loading environment variables...") print("Loading environment variables...")
dotenv.load_dotenv() dotenv.load_dotenv()
SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID") SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
SQUARE_WEBHOOK_SIGNATURE_KEY = os.environ.get("SQUARE_WEBHOOK_SIGNATURE_KEY") SQUARE_WEBHOOK_SIGNATURE_KEY = os.environ.get("SQUARE_WEBHOOK_SIGNATURE_KEY")
SQUARE_WEBHOOK_URL = os.environ.get("SQUARE_WEBHOOK_URL") SQUARE_WEBHOOK_URL = os.environ.get("SQUARE_WEBHOOK_URL")
from asyncio_throttle.throttler import Throttler from asyncio_throttle.throttler import Throttler
# set up logger # set up logger
print("Setting up logger...") print("Setting up logger...")
logger = logging.getLogger("studspider") logger = logging.getLogger("studspider")
logger.setLevel(logging.DEBUG) logger.setLevel(logging.DEBUG)
handler = RotatingFileHandler("./logs/studspider.log", maxBytes=125000, backupCount=5) handler = RotatingFileHandler("./logs/studspider.log", maxBytes=125000, backupCount=5)
logger.addHandler(handler) logger.addHandler(handler)
lego = AsyncBricksetAPI(debug=True, logger=logger) lego = AsyncBricksetAPI(debug=True, logger=logger)
# sq = AsyncSquareClient(logger=logger) # sq = AsyncSquareClient(logger=logger)
# logging.basicConfig( # logging.basicConfig(
# filename="./logs/studspider.log", encoding="utf-8", level=logging.DEBUG # filename="./logs/studspider.log", encoding="utf-8", level=logging.DEBUG
# ) # )
print("Setting up rate limiter...") print("Setting up rate limiter...")
print("Doing RE things...") print("Doing RE things...")
UA = re.compile(r"^kopling\/?", re.IGNORECASE) UA = re.compile(r"^kopling\/?", re.IGNORECASE)
async def run_update(): async def run_update():
set_list = await sq.acat2id(SQUARE_LEGO_CAT_ID) # type: ignore """
# print(f"[@DEBUG]\t {set_list}") # type: ignore DEPRECATED: Use glue.enrich_bulk() instead
for l_set in set_list:
# print(f"[@DEBUG]{l_set}") Legacy RUNUPDATE command handler - kept for backwards compatibility
try: but now delegates to the new glue.enrich_bulk() function
l_setinfo = await lego.throttled_get_set(l_set["number"]) """
print(f"[@DEBUG] Found {l_setinfo["pw_name"]}") from src import glue
except Exception as e:
print(f"Error fetching set\t: {l_set}: {str(e)}") await glue.enrich_bulk()
continue
# print(f"[@DEBUG]{l_setinfo}") def valid_uuid(val):
# await sq.update_itm(l_set, l_setinfo) try:
"""for l_set in set_list: uuid.UUID(str(val))
print(f"[@DEBUG]{l_set}") return True
try: except ValueError:
l_setinfo = await lego.throttled_get_set(l_set["number"]) return False
except Exception as e:
print(f"Error fetching set\t: {l_set}: {str(e)}")
continue # In-memory store for processed event IDs (idempotency)
processed_event_ids = set()
print(f"[@DEBUG]{l_setinfo}")""" MAX_STORED_EVENT_IDS = 10000 # Prevent memory growth
# await sq.update_itm(l_set, l_setinfo)
async def hook_handler(request: web.Request) -> web.Response:
def valid_uuid(val): """Handle incoming JSON webhook POSTs on /hook endpoint"""
try: try:
uuid.UUID(str(val)) current_time = int(asyncio.get_event_loop().time())
return True
except ValueError: # Verify request method
return False if request.method != "POST":
logger.error(
f"{current_time}\tInvalid request method to /hook from %s",
# In-memory store for processed event IDs (idempotency) request.remote,
processed_event_ids = set() )
MAX_STORED_EVENT_IDS = 10000 # Prevent memory growth return web.Response(status=405, text="Method not allowed. Use POST.")
# Read raw body for signature verification
async def hook_handler(request: web.Request) -> web.Response: body = await request.read()
"""Handle incoming JSON webhook POSTs on /hook endpoint""" body_str = body.decode("utf-8")
try:
current_time = int(asyncio.get_event_loop().time()) # Verify Square webhook signature using Square SDK
signature_header = request.headers.get("x-square-hmacsha256-signature", "")
# Verify request method
if request.method != "POST": if SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL:
logger.error( if not signature_header:
f"{current_time}\tInvalid request method to /hook from %s", logger.error(
request.remote, f"{current_time}\tMissing signature header from {request.remote}"
) )
return web.Response(status=405, text="Method not allowed. Use POST.") print(f"[@DEBUG] Missing x-square-hmacsha256-signature header")
return web.Response(status=401, text="Unauthorized: Missing signature")
# Read raw body for signature verification
body = await request.read() # Use Square SDK's official validation function
body_str = body.decode("utf-8") is_valid = verify_signature(
request_body=body_str,
# Verify Square webhook signature using Square SDK signature_header=signature_header,
signature_header = request.headers.get("x-square-hmacsha256-signature", "") signature_key=SQUARE_WEBHOOK_SIGNATURE_KEY,
notification_url=SQUARE_WEBHOOK_URL,
if SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL: )
if not signature_header:
logger.error( if not is_valid:
f"{current_time}\tMissing signature header from {request.remote}" logger.error(f"{current_time}\tInvalid signature from {request.remote}")
) print(f"[@DEBUG] Invalid webhook signature!")
print(f"[@DEBUG] Missing x-square-hmacsha256-signature header") return web.Response(status=401, text="Unauthorized: Invalid signature")
return web.Response(status=401, text="Unauthorized: Missing signature")
print(f"[@DEBUG] ✓ Webhook signature verified")
# Use Square SDK's official validation function else:
is_valid = verify_signature( print(
request_body=body_str, f"[@DEBUG] ⚠️ Webhook signature verification DISABLED (set SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL in .env)"
signature_header=signature_header, )
signature_key=SQUARE_WEBHOOK_SIGNATURE_KEY,
notification_url=SQUARE_WEBHOOK_URL, # Parse JSON data
) try:
data = json.loads(body_str)
if not is_valid: except json.JSONDecodeError as e:
logger.error(f"{current_time}\tInvalid signature from {request.remote}") print(f"[@DEBUG] Invalid JSON to /hook from {request.remote}: {str(e)}")
print(f"[@DEBUG] Invalid webhook signature!") logger.error(f"Invalid JSON to /hook from {request.remote}: {str(e)}")
return web.Response(status=401, text="Unauthorized: Invalid signature") return web.Response(status=400, text="Invalid JSON in request body")
print(f"[@DEBUG] ✓ Webhook signature verified") # Check for idempotency (event_id)
else: event_id = data.get("event_id")
print( if event_id:
f"[@DEBUG] ⚠️ Webhook signature verification DISABLED (set SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL in .env)" if event_id in processed_event_ids:
) print(
f"[@DEBUG] Duplicate event {event_id} - already processed (idempotency)"
# Parse JSON data )
try: logger.info(
data = json.loads(body_str) f"{current_time}\tDuplicate event {event_id} from {request.remote}"
except json.JSONDecodeError as e: )
print(f"[@DEBUG] Invalid JSON to /hook from {request.remote}: {str(e)}") return web.json_response(
logger.error(f"Invalid JSON to /hook from {request.remote}: {str(e)}") {
return web.Response(status=400, text="Invalid JSON in request body") "status": "success",
"message": "Event already processed (idempotency)",
# Check for idempotency (event_id) "event_id": event_id,
event_id = data.get("event_id") }
if event_id: )
if event_id in processed_event_ids:
print( # Add to processed set
f"[@DEBUG] Duplicate event {event_id} - already processed (idempotency)" processed_event_ids.add(event_id)
)
logger.info( # Prevent memory growth - remove oldest if too many
f"{current_time}\tDuplicate event {event_id} from {request.remote}" if len(processed_event_ids) > MAX_STORED_EVENT_IDS:
) processed_event_ids.pop()
return web.json_response(
{ # Get event info for quick response
"status": "success", webhook_type = data.get("type", "unknown")
"message": "Event already processed (idempotency)", event_id_val = data.get("event_id", "none")
"event_id": event_id,
} # RESPOND IMMEDIATELY - Square requires fast response (< 2 seconds)
) # Don't do any processing before responding!
response_data = {
# Add to processed set "status": "success",
processed_event_ids.add(event_id) "message": "Webhook received",
"event_id": event_id_val,
# Prevent memory growth - remove oldest if too many }
if len(processed_event_ids) > MAX_STORED_EVENT_IDS:
processed_event_ids.pop() # Log and process AFTER sending response (async background tasks)
async def log_and_process():
# Get event info for quick response # Log the received webhook with detailed output
webhook_type = data.get("type", "unknown") print(f"\n{'='*60}")
event_id_val = data.get("event_id", "none") print(f"[@DEBUG] WEBHOOK RECEIVED at {current_time}")
print(f"[@DEBUG] From: {request.remote}")
# RESPOND IMMEDIATELY - Square requires fast response (< 2 seconds) print(f"[@DEBUG] Headers: {dict(request.headers)}")
# Don't do any processing before responding! print(f"[@DEBUG] Data type: {webhook_type}")
response_data = { print(f"[@DEBUG] Event ID: {event_id_val}")
"status": "success", print(f"[@DEBUG] Full payload:")
"message": "Webhook received", print(json.dumps(data, indent=2))
"event_id": event_id_val, print(f"{'='*60}\n")
}
logger.info(
# Log and process AFTER sending response (async background tasks) f"{current_time}\tWebhook received from {request.remote}: {json.dumps(data)}"
async def log_and_process(): )
# Log the received webhook with detailed output
print(f"\n{'='*60}") # Process the webhook data
print(f"[@DEBUG] WEBHOOK RECEIVED at {current_time}") if webhook_type == "inventory.count.updated":
print(f"[@DEBUG] From: {request.remote}") print(f"[@DEBUG] Processing inventory count update...")
print(f"[@DEBUG] Headers: {dict(request.headers)}") # Extract inventory data
print(f"[@DEBUG] Data type: {webhook_type}") inventory_counts = (
print(f"[@DEBUG] Event ID: {event_id_val}") data.get("data", {}).get("object", {}).get("inventory_counts", [])
print(f"[@DEBUG] Full payload:") )
print(json.dumps(data, indent=2)) for count in inventory_counts:
print(f"{'='*60}\n") print(
f"[@DEBUG] Catalog Object: {count.get('catalog_object_id')}"
logger.info( )
f"{current_time}\tWebhook received from {request.remote}: {json.dumps(data)}" print(f"[@DEBUG] Quantity: {count.get('quantity')}")
) print(f"[@DEBUG] State: {count.get('state')}")
# Process the webhook data # Start background processing (don't wait for it)
if webhook_type == "inventory.count.updated": asyncio.create_task(log_and_process())
print(f"[@DEBUG] Processing inventory count update...")
# Extract inventory data # Return response immediately
inventory_counts = ( return web.json_response(response_data)
data.get("data", {}).get("object", {}).get("inventory_counts", [])
) except json.JSONDecodeError as e:
for count in inventory_counts: error_msg = f"JSON decode error in webhook: {str(e)}"
print( print(f"[@DEBUG] {error_msg}")
f"[@DEBUG] Catalog Object: {count.get('catalog_object_id')}" logger.error(error_msg)
) return web.Response(status=400, text=error_msg)
print(f"[@DEBUG] Quantity: {count.get('quantity')}") except Exception as e:
print(f"[@DEBUG] State: {count.get('state')}") error_msg = f"Webhook handler error: {str(e)}"
print(f"[@DEBUG] {error_msg}")
# Start background processing (don't wait for it) print(f"[@DEBUG] Exception type: {type(e).__name__}")
asyncio.create_task(log_and_process()) import traceback
# Return response immediately print(f"[@DEBUG] Traceback:\n{traceback.format_exc()}")
return web.json_response(response_data) logger.error(f"{error_msg}\n{traceback.format_exc()}")
return web.Response(status=500, text="Internal server error")
except json.JSONDecodeError as e:
error_msg = f"JSON decode error in webhook: {str(e)}"
print(f"[@DEBUG] {error_msg}") async def request_handler(request: web.Request) -> web.Response:
logger.error(error_msg) """Handle incoming POST requests with command processing"""
return web.Response(status=400, text=error_msg) try:
except Exception as e: # Verify request method
error_msg = f"Webhook handler error: {str(e)}" if request.method != "POST":
print(f"[@DEBUG] {error_msg}") logger.error(f"{time}\tInvalid request method from %s at ", request.remote)
print(f"[@DEBUG] Exception type: {type(e).__name__}") return web.Response(status=405, text="Method not allowed. Use POST.")
import traceback
# Get JSON data from request
print(f"[@DEBUG] Traceback:\n{traceback.format_exc()}") try:
logger.error(f"{error_msg}\n{traceback.format_exc()}") data = await request.json()
return web.Response(status=500, text="Internal server error") except json.JSONDecodeError:
logger.error("Invalid JSON from %s", request.remote)
return web.Response(
async def request_handler(request: web.Request) -> web.Response: status=400, text="{time()}\tInvalid JSON in request body"
"""Handle incoming POST requests with command processing""" )
try: # Validate Idempotency-Key header
# Verify request method if valid_uuid(request.headers.get("Idempotency-Key")) is False:
if request.method != "POST": logger.error("{time()}\tInvalid Idempotency-Key from %s", request.remote)
logger.error(f"{time}\tInvalid request method from %s at ", request.remote) return web.Response(status=412, text="412 Headers rejected!")
return web.Response(status=405, text="Method not allowed. Use POST.") # Validate User-Agent header
if not re.match(UA, str(request.headers.get("User-Agent"))):
# Get JSON data from request logger.error("{time()}\tInvalid User-Agent from %s", request.remote)
try: return web.Response(status=412, text="412 Headers rejected!")
data = await request.json() # Validate command exists
except json.JSONDecodeError: if "CMD" not in data:
logger.error("Invalid JSON from %s", request.remote) logger.error("{time()}\tMissing CMD field from %s", request.remote)
return web.Response( return web.Response(status=400, text="Missing CMD field in request")
status=400, text="{time()}\tInvalid JSON in request body"
) # Decode and process command
# Validate Idempotency-Key header try:
if valid_uuid(request.headers.get("Idempotency-Key")) is False: decoded_cmd = b64decode(data["CMD"]).decode("utf-8")
logger.error("{time()}\tInvalid Idempotency-Key from %s", request.remote) print(
return web.Response(status=412, text="412 Headers rejected!") f"""At: {time}\nReceived command: {decoded_cmd}\nUser-Agent: {request.headers.get('User-Agent')}\nIdempotency-Key: {request.headers.get('Idempotency-Key')}"""
# Validate User-Agent header )
if not re.match(UA, str(request.headers.get("User-Agent"))): logging.info(
logger.error("{time()}\tInvalid User-Agent from %s", request.remote) f"{time}\tCommand: {decoded_cmd}\tFrom: {request.remote}\tUA: {request.headers.get('User-Agent')}\tIdempotency-Key: {request.headers.get('Idempotency-Key')}"
return web.Response(status=412, text="412 Headers rejected!") )
# Validate command exists
if "CMD" not in data: # Process the command
logger.error("{time()}\tMissing CMD field from %s", request.remote) if decoded_cmd == "RUNUPDATE":
return web.Response(status=400, text="Missing CMD field in request") response_data = {"status": "success", "message": "Update started"}
await run_update()
# Decode and process command logging.info(f"{time}\tUpdate command processed successfully")
try: return web.json_response(response_data)
decoded_cmd = b64decode(data["CMD"]).decode("utf-8") else:
print( return web.Response(status=400, text=f"Unknown command: {decoded_cmd}")
f"""At: {time}\nReceived command: {decoded_cmd}\nUser-Agent: {request.headers.get('User-Agent')}\nIdempotency-Key: {request.headers.get('Idempotency-Key')}"""
) except Exception as e:
logging.info( print(f"Error processing command: {str(e)}")
f"{time}\tCommand: {decoded_cmd}\tFrom: {request.remote}\tUA: {request.headers.get('User-Agent')}\tIdempotency-Key: {request.headers.get('Idempotency-Key')}" return web.Response(status=500, text=f"Server error: {str(e)}")
)
except Exception as e:
# Process the command print(f"Request handler error: {str(e)}")
if decoded_cmd == "RUNUPDATE": return web.Response(status=500, text="Internal server error")
response_data = {"status": "success", "message": "Update started"}
await run_update()
logging.info(f"{time}\tUpdate command processed successfully") print("Starting web server...")
return web.json_response(response_data) os.system("cls" if os.name == "nt" else "clear")
else: app = web.Application()
return web.Response(status=400, text=f"Unknown command: {decoded_cmd}") app.router.add_post("/api", request_handler)
app.router.add_post("/hook", hook_handler)
except Exception as e:
print(f"Error processing command: {str(e)}") parser = argparse.ArgumentParser(
return web.Response(status=500, text=f"Server error: {str(e)}") description="studspider - LEGO Inventory Sync Service for Square POS and Brickset",
epilog="""
except Exception as e: Examples:
print(f"Request handler error: {str(e)}") # Basic HTTP server (recommended with nginx reverse proxy)
return web.Response(status=500, text="Internal server error") python src/test_server.py --host 0.0.0.0 --port 7432
# With separate webhook port
print("Starting web server...") python src/test_server.py --port 7432 --hook-port 8080
os.system("cls" if os.name == "nt" else "clear")
app = web.Application() # HTTPS using certificates from .env (set SSL_CERT_PATH and SSL_KEY_PATH)
app.router.add_post("/api", request_handler) python src/test_server.py --hook-port 8443 --hook-ssl-only
app.router.add_post("/hook", hook_handler)
# HTTPS with command line override (self-signed cert for testing)
parser = argparse.ArgumentParser( python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only
description="studspider - LEGO Inventory Sync Service for Square POS and Brickset",
epilog=""" # Full HTTPS (both endpoints)
Examples: python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443
# Basic HTTP server (recommended with nginx reverse proxy)
python src/test_server.py --host 0.0.0.0 --port 7432 # Production behind nginx (no SSL needed in app)
python src/test_server.py --host 127.0.0.1 --port 7432
# With separate webhook port
python src/test_server.py --port 7432 --hook-port 8080 Environment Variables (.env):
SSL_CERT_PATH - Default SSL certificate path
# HTTPS using certificates from .env (set SSL_CERT_PATH and SSL_KEY_PATH) SSL_KEY_PATH - Default SSL key path
python src/test_server.py --hook-port 8443 --hook-ssl-only
For more information, see doc/
# HTTPS with command line override (self-signed cert for testing) """,
python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only formatter_class=argparse.RawDescriptionHelpFormatter,
)
# Full HTTPS (both endpoints) parser.add_argument(
python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 "--host",
type=str,
# Production behind nginx (no SSL needed in app) default="localhost",
python src/test_server.py --host 127.0.0.1 --port 7432 help="IP address to listen on (default: localhost, use 0.0.0.0 for all interfaces)",
)
Environment Variables (.env): parser.add_argument(
SSL_CERT_PATH - Default SSL certificate path "--port", type=int, default=7432, help="Port for main API endpoint (default: 7432)"
SSL_KEY_PATH - Default SSL key path )
parser.add_argument(
For more information, see doc/ "--hook-port",
""", type=int,
formatter_class=argparse.RawDescriptionHelpFormatter, default=None,
) help="Separate port for webhook endpoint (default: None - uses same port as --port)",
parser.add_argument( )
"--host", parser.add_argument(
type=str, "--ssl-cert",
default="localhost", type=str,
help="IP address to listen on (default: localhost, use 0.0.0.0 for all interfaces)", default=get_sq_env.SSL_CERT_PATH,
) help="Path to SSL certificate file in PEM format (default: from .env SSL_CERT_PATH)",
parser.add_argument( )
"--port", type=int, default=7432, help="Port for main API endpoint (default: 7432)" parser.add_argument(
) "--ssl-key",
parser.add_argument( type=str,
"--hook-port", default=get_sq_env.SSL_KEY_PATH,
type=int, help="Path to SSL private key file in PEM format (default: from .env SSL_KEY_PATH)",
default=None, )
help="Separate port for webhook endpoint (default: None - uses same port as --port)", parser.add_argument(
) "--hook-ssl-only",
parser.add_argument( action="store_true",
"--ssl-cert", help="Enable HTTPS only for webhook endpoint, main API stays HTTP (requires --ssl-cert and --ssl-key)",
type=str, )
default=get_sq_env.SSL_CERT_PATH, args = parser.parse_args()
help="Path to SSL certificate file in PEM format (default: from .env SSL_CERT_PATH)",
) if __name__ == "__main__":
parser.add_argument( print("Creating log directory if it doesn't exist...")
"--ssl-key", os.makedirs("./logs/", exist_ok=True)
type=str, print(banner.generate("studspider v0.1"))
default=get_sq_env.SSL_KEY_PATH, print("Built for PLAY-WELL TEKNOLOGIES by contact@studspider.sh\n\n")
help="Path to SSL private key file in PEM format (default: from .env SSL_KEY_PATH)",
) # Setup SSL context if certificates provided
parser.add_argument( ssl_context = None
"--hook-ssl-only", if args.ssl_cert and args.ssl_key:
action="store_true", # Check if trying to use P7B file
help="Enable HTTPS only for webhook endpoint, main API stays HTTP (requires --ssl-cert and --ssl-key)", if args.ssl_cert.endswith(".p7b") or args.ssl_key.endswith(".p7b"):
) print("ERROR: Cannot use P7B files directly for HTTPS server!")
args = parser.parse_args() print("\nP7B files are for Square's webhook signature verification only.")
print("Your server needs PEM format files (.pem)")
if __name__ == "__main__": print("\nWorkflow:")
print("Creating log directory if it doesn't exist...") print("1. Generate PEM files: python generate_cert.py")
os.makedirs("./logs/", exist_ok=True) print("2. Convert to P7B for Square: python convert_to_p7b.py")
print(banner.generate("studspider v0.1")) print("3. Upload P7B to Square Dashboard")
print("Built for PLAY-WELL TEKNOLOGIES by contact@studspider.sh\n\n") print("4. Run server with PEM files:")
print(
# Setup SSL context if certificates provided " python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443"
ssl_context = None )
if args.ssl_cert and args.ssl_key: print("\nIf you have a P7B and need PEM:")
# Check if trying to use P7B file print(" python convert_from_p7b.py your_cert.p7b")
if args.ssl_cert.endswith(".p7b") or args.ssl_key.endswith(".p7b"): exit(1)
print("ERROR: Cannot use P7B files directly for HTTPS server!")
print("\nP7B files are for Square's webhook signature verification only.") try:
print("Your server needs PEM format files (.pem)") ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
print("\nWorkflow:") ssl_context.load_cert_chain(args.ssl_cert, args.ssl_key)
print("1. Generate PEM files: python generate_cert.py") print(f"✓ SSL enabled with cert: {args.ssl_cert}")
print("2. Convert to P7B for Square: python convert_to_p7b.py") except FileNotFoundError as e:
print("3. Upload P7B to Square Dashboard") print(f"ERROR: Certificate file not found: {e}")
print("4. Run server with PEM files:") exit(1)
print( except ssl.SSLError as e:
" python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443" print(f"ERROR: Invalid certificate or key file: {e}")
) print("\nMake sure you're using PEM format files.")
print("\nIf you have a P7B and need PEM:") exit(1)
print(" python convert_from_p7b.py your_cert.p7b") elif args.ssl_cert or args.ssl_key:
exit(1) print("ERROR: Both --ssl-cert and --ssl-key must be provided together")
exit(1)
try: else:
ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH) print("ℹ️ Running without SSL (HTTP only)")
ssl_context.load_cert_chain(args.ssl_cert, args.ssl_key) print(" For production with nginx reverse proxy, this is recommended.")
print(f"✓ SSL enabled with cert: {args.ssl_cert}") print(" Nginx will handle HTTPS, forward HTTP to this server.")
except FileNotFoundError as e:
print(f"ERROR: Certificate file not found: {e}") # If hook-port is specified, run a separate server for webhooks
exit(1) if args.hook_port:
except ssl.SSLError as e: protocol_main = "https" if (ssl_context and not args.hook_ssl_only) else "http"
print(f"ERROR: Invalid certificate or key file: {e}") protocol_hook = "https" if ssl_context else "http"
print("\nMake sure you're using PEM format files.")
exit(1) # Create separate app for webhook
elif args.ssl_cert or args.ssl_key: hook_app = web.Application()
print("ERROR: Both --ssl-cert and --ssl-key must be provided together") hook_app.router.add_post("/hook", hook_handler)
exit(1)
else: # Run both servers concurrently
print("ℹ️ Running without SSL (HTTP only)") async def run_servers():
print(" For production with nginx reverse proxy, this is recommended.") runner1 = web.AppRunner(app)
print(" Nginx will handle HTTPS, forward HTTP to this server.") await runner1.setup()
# Main server uses SSL only if not hook-ssl-only mode
# If hook-port is specified, run a separate server for webhooks main_ssl = None if args.hook_ssl_only else ssl_context
if args.hook_port: site1 = web.TCPSite(runner1, args.host, args.port, ssl_context=main_ssl)
protocol_main = "https" if (ssl_context and not args.hook_ssl_only) else "http"
protocol_hook = "https" if ssl_context else "http" runner2 = web.AppRunner(hook_app)
await runner2.setup()
# Create separate app for webhook # Webhook server always uses SSL if certificates provided
hook_app = web.Application() site2 = web.TCPSite(
hook_app.router.add_post("/hook", hook_handler) runner2, args.host, args.hook_port, ssl_context=ssl_context
)
# Run both servers concurrently
async def run_servers(): await site1.start()
runner1 = web.AppRunner(app) await site2.start()
await runner1.setup()
# Main server uses SSL only if not hook-ssl-only mode print(f"Starting main server on {args.host}:{args.port} ({protocol_main})")
main_ssl = None if args.hook_ssl_only else ssl_context print(
site1 = web.TCPSite(runner1, args.host, args.port, ssl_context=main_ssl) f"Starting webhook server on {args.host}:{args.hook_port} ({protocol_hook})"
)
runner2 = web.AppRunner(hook_app)
await runner2.setup() # Keep running
# Webhook server always uses SSL if certificates provided await asyncio.Event().wait()
site2 = web.TCPSite(
runner2, args.host, args.hook_port, ssl_context=ssl_context asyncio.run(run_servers())
) else:
protocol = "https" if ssl_context else "http"
await site1.start() print(f"Server running on {args.host}:{args.port} ({protocol})")
await site2.start() print(f"Webhook endpoint: {protocol}://{args.host}:{args.port}/hook")
web.run_app(app, host=args.host, port=args.port, ssl_context=ssl_context)
print(f"Starting main server on {args.host}:{args.port} ({protocol_main})")
print(
f"Starting webhook server on {args.host}:{args.hook_port} ({protocol_hook})"
)
# Keep running
await asyncio.Event().wait()
asyncio.run(run_servers())
else:
protocol = "https" if ssl_context else "http"
print(f"Server running on {args.host}:{args.port} ({protocol})")
print(f"Webhook endpoint: {protocol}://{args.host}:{args.port}/hook")
web.run_app(app, host=args.host, port=args.port, ssl_context=ssl_context)
File renamed without changes.
File renamed without changes.