im actually gonna crash out rn dude

.env.example

@@ -1,72 +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 (not recommended with nginx)
-# For nginx reverse proxy setup, leave these empty
-
-# Path to SSL certificate in PEM format
-# SSL_CERT_PATH=/path/to/cert.pem
-
-# Path to SSL private key in PEM format
-# SSL_KEY_PATH=/path/to/key.pem
-
-# ============================================================================
-# Square Webhook Certificate (P7B Format)
-# ============================================================================
-
-# Path to P7B certificate file for Square webhook signature verification
-# This file is UPLOADED to Square Dashboard, not used by your server
-# Generate with: python convert_to_p7b.py
-SQUARE_P7B_PATH=./cert.p7b
-
-# ============================================================================
-# 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

.gitignore

@@ -37,4 +37,7 @@ win/
 .github/
 bin/
 activate/
-build/
+build/
+.pem
+.p7b
+nginx/

README.md

@@ -5,14 +5,14 @@
 
 
 
-is a tool used to talk to brickset.com and a square inventory to pull lego set information for entering in the square db
+is a tool used to talk to brickset.com, pull lego set information and automatically enter it into the square inventory!
 ## Run me
 
 Clone the project
 
 ```bash
   git clone http://git.traps.moe/ph0sgene/studspider.git
-  git checkout uv 
+  git checkout branch 
 ```
 
 Go to the project directory
@@ -30,14 +30,15 @@ Install dependencies
 Start the server
 
 ```bash
-  uvx run studspider.py
+  uv run studspider.py
 ```
 
 
 ## Tech 
 
+- gunicorn
+- uvicorn
 - aiohttp
 - squareup
 - apscheduler
-- aiosqlite
+- aiosqlite
-

config/gunicorn.conf.py

@@ -0,0 +1,75 @@
+import multiprocessing
+import os
+
+# Server socket
+bind = "127.0.0.1:7433"  # nginx will proxy to this
+backlog = 2048
+
+# Worker processes
+workers = multiprocessing.cpu_count() * 2 + 1  # Recommended formula
+worker_class = "uvicorn.workers.UvicornWorker"  # Async worker for FastAPI
+worker_connections = 1000
+max_requests = 1000  # Restart workers after this many requests
+max_requests_jitter = 50  # Add randomness to prevent all workers restarting at once
+
+# Timeouts
+timeout = 30  # Workers silent for more than this are killed
+keepalive = 2  # Wait this long for requests on Keep-Alive connections
+
+# Logging
+accesslog = "./logs/webhook_access.log"
+errorlog = "./logs/webhook_error.log"
+loglevel = "info"
+access_log_format = '%(h)s %(l)s %(u)s %(t)s "%(r)s" %(s)s %(b)s "%(f)s" "%(a)s" %(D)s'
+
+# Process naming
+proc_name = "studspider-webhook"
+
+# Server mechanics
+daemon = False  # Run in foreground (systemd will handle daemonization)
+pidfile = "./logs/webhook.pid"
+umask = 0
+user = None  # Run as current user (systemd will set user)
+group = None
+tmp_upload_dir = None
+
+# SSL (not needed with nginx, but here for reference)
+# keyfile = None
+# certfile = None
+
+# Security
+limit_request_line = 4096
+limit_request_fields = 100
+limit_request_field_size = 8190
+
+# Debugging (disable in production)
+reload = os.getenv("GUNICORN_RELOAD", "false").lower() == "true"
+reload_engine = "auto"
+
+
+# Hooks
+def on_starting(server):
+    """Called just before the master process is initialized."""
+    print("Studspider WEB starting...")
+
+
+def on_reload(server):
+    """Called to recycle workers during a reload via SIGHUP."""
+    print("🔄 Reloading workers...")
+
+
+def when_ready(server):
+    """Called just after the server is started."""
+    print(f"STUDSPIDER HOOK v0.1 RUNNING ON {bind}")
+    print(f"WORKERS: {workers}")
+    print(f"WORKING CLASS: {worker_class}\n")
+
+
+def worker_int(worker):
+    """Called just after a worker exited on SIGINT or SIGQUIT."""
+    print(f"⚠️  Worker {worker.pid} received INT or QUIT signal")
+
+
+def worker_abort(worker):
+    """Called when a worker received the SIGABRT signal."""
+    print(f"❌ Worker {worker.pid} aborted")

edderkop.code-workspace

@@ -1,11 +0,0 @@
-{
-	"folders": [
-		{
-			"path": "."
-		},
-		{
-			"path": "../kopling"
-		}
-	],
-	"settings": {}
-}

pyproject.toml

@@ -18,6 +18,9 @@ dependencies = [
     "aiosqlite (>=0.21.0,<0.22.0)",
     "bs4>=0.0.2",
     "reflex>=0.8.14.post1",
+    "fastapi>=0.128.0",
+    "uvicorn>=0.40.0",
+    "gunicorn>=23.0.0",
 ]
 
 [build-system]

src/get_sq_env.py

@@ -10,9 +10,10 @@ LEGO_CAT_ID = os.getenv("SQUARE_LEGO_CAT_ID")
 SSL_CERT_PATH = os.getenv("SSL_CERT_PATH")
 SSL_KEY_PATH = os.getenv("SSL_KEY_PATH")
 
-# P7B certificate for Square webhook signature verification
+# Square Webhook Configuration
-# This is uploaded to Square Dashboard, not used by the server
+# Get signature key from Square Developer Dashboard → Webhooks
-SQUARE_P7B_PATH = os.getenv("SQUARE_P7B_PATH")
+SQUARE_WEBHOOK_SIGNATURE_KEY = os.getenv("SQUARE_WEBHOOK_SIGNATURE_KEY")
+SQUARE_WEBHOOK_URL = os.getenv("SQUARE_WEBHOOK_URL")
 
 if os.getenv("SQUARE_ENV") == "production":
     SQUARE_APP_ID = os.getenv("PROD_SQUARE_APP_ID")

src/old/ebricksetapi.py

@@ -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}")

src/old/server.py

@@ -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)

studspider.py

@@ -1,14 +1,279 @@
-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 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_SIGNATURE_KEY") or os.getenv(
+    "PROD_SQUARE_SIGNATURE_KEY"
+)
+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")
 
 
-class Studspider:
+@app.get("/")
-    def __init__(self):
+async def root():
-        self.name = "Studspider"
+    """Landing page"""
-        self.version = "0.0.1"
+    return {
-        self.description = """
+        "service": "Studspider",
-        studspider is an application designed to interact with the Brickset API and directly with Square.
+        "status": "running",
-        It allows users to search for items, manage collections, and perform various operations related to bricksets.
+        "endpoints": {"webhook": "/hook (POST)", "health": "/health (GET)"},
-        """
+    }
 
-    def run(self):
+
-        print(f"{self.name} v{self.version} is running.")
+@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 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")

tests/test_client.py

@@ -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())

tests/test_server.py

@@ -1,400 +1,452 @@
-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
-
+
-# from asquareclass import AsyncSquareClient  # type: ignore
+# Square SDK webhook validation
-from abricksetapi import AsyncBricksetAPI
+from square.utils.webhooks_helper import verify_signature
-
+
-print("Loading environment variables...")
+# from asquareclass import AsyncSquareClient  # type: ignore
-dotenv.load_dotenv()
+from abricksetapi import AsyncBricksetAPI
-SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
+
-
+print("Loading environment variables...")
-from asyncio_throttle.throttler import Throttler
+dotenv.load_dotenv()
-
+SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
-# set up logger
+SQUARE_WEBHOOK_SIGNATURE_KEY = os.environ.get("SQUARE_WEBHOOK_SIGNATURE_KEY")
-print("Setting up logger...")
+SQUARE_WEBHOOK_URL = os.environ.get("SQUARE_WEBHOOK_URL")
-logger = logging.getLogger("studspider")
+
-logger.setLevel(logging.DEBUG)
+from asyncio_throttle.throttler import Throttler
-handler = RotatingFileHandler("./logs/studspider.log", maxBytes=125000, backupCount=5)
+
-logger.addHandler(handler)
+# set up logger
-
+print("Setting up logger...")
-lego = AsyncBricksetAPI(debug=True, logger=logger)
+logger = logging.getLogger("studspider")
-# sq = AsyncSquareClient(logger=logger)
+logger.setLevel(logging.DEBUG)
-# logging.basicConfig(
+handler = RotatingFileHandler("./logs/studspider.log", maxBytes=125000, backupCount=5)
-#    filename="./logs/studspider.log", encoding="utf-8", level=logging.DEBUG
+logger.addHandler(handler)
-# )
+
-print("Setting up rate limiter...")
+lego = AsyncBricksetAPI(debug=True, logger=logger)
-print("Doing RE things...")
+# sq = AsyncSquareClient(logger=logger)
-UA = re.compile(r"^kopling\/?", re.IGNORECASE)
+# logging.basicConfig(
-
+#    filename="./logs/studspider.log", encoding="utf-8", level=logging.DEBUG
-
+# )
-async def run_update():
+print("Setting up rate limiter...")
-    set_list = await sq.acat2id(SQUARE_LEGO_CAT_ID)  # type: ignore
+print("Doing RE things...")
-    # print(f"[@DEBUG]\t {set_list}")  # type: ignore
+UA = re.compile(r"^kopling\/?", re.IGNORECASE)
-    for l_set in set_list:
+
-        # print(f"[@DEBUG]{l_set}")
+
-        try:
+async def run_update():
-            l_setinfo = await lego.throttled_get_set(l_set["number"])
+    set_list = await sq.acat2id(SQUARE_LEGO_CAT_ID)  # type: ignore
-            print(f"[@DEBUG] Found {l_setinfo["pw_name"]}")
+    # print(f"[@DEBUG]\t {set_list}")  # type: ignore
-        except Exception as e:
+    for l_set in set_list:
-            print(f"Error fetching set\t: {l_set}: {str(e)}")
+        # print(f"[@DEBUG]{l_set}")
-            continue
+        try:
-
+            l_setinfo = await lego.throttled_get_set(l_set["number"])
-        # print(f"[@DEBUG]{l_setinfo}")
+            print(f"[@DEBUG] Found {l_setinfo["pw_name"]}")
-        # await sq.update_itm(l_set, l_setinfo)
+        except Exception as e:
-    """for l_set in set_list:
+            print(f"Error fetching set\t: {l_set}: {str(e)}")
-        print(f"[@DEBUG]{l_set}")
+            continue
-        try:
+
-            l_setinfo = await lego.throttled_get_set(l_set["number"])
+        # print(f"[@DEBUG]{l_setinfo}")
-        except Exception as e:
+        # await sq.update_itm(l_set, l_setinfo)
-            print(f"Error fetching set\t: {l_set}: {str(e)}")
+    """for l_set in set_list:
-            continue
+        print(f"[@DEBUG]{l_set}")
-
+        try:
-        print(f"[@DEBUG]{l_setinfo}")"""
+            l_setinfo = await lego.throttled_get_set(l_set["number"])
-    #    await sq.update_itm(l_set, l_setinfo)
+        except Exception as e:
-
+            print(f"Error fetching set\t: {l_set}: {str(e)}")
-
+            continue
-def valid_uuid(val):
+
-    try:
+        print(f"[@DEBUG]{l_setinfo}")"""
-        uuid.UUID(str(val))
+    #    await sq.update_itm(l_set, l_setinfo)
-        return True
+
-    except ValueError:
+
-        return False
+def valid_uuid(val):
-
+    try:
-
+        uuid.UUID(str(val))
-async def hook_handler(request: web.Request) -> web.Response:
+        return True
-    """Handle incoming JSON webhook POSTs on /hook endpoint"""
+    except ValueError:
-    try:
+        return False
-        current_time = int(asyncio.get_event_loop().time())
+
-
+
-        # Verify request method
+# In-memory store for processed event IDs (idempotency)
-        if request.method != "POST":
+processed_event_ids = set()
-            logger.error(
+MAX_STORED_EVENT_IDS = 10000  # Prevent memory growth
-                f"{current_time}\tInvalid request method to /hook from %s",
+
-                request.remote,
+
-            )
+async def hook_handler(request: web.Request) -> web.Response:
-            return web.Response(status=405, text="Method not allowed. Use POST.")
+    """Handle incoming JSON webhook POSTs on /hook endpoint"""
-
+    try:
-        # Get JSON data from request
+        current_time = int(asyncio.get_event_loop().time())
-        try:
+
-            data = await request.json()
+        # Verify request method
-        except json.JSONDecodeError as e:
+        if request.method != "POST":
-            print(f"[@DEBUG] Invalid JSON to /hook from {request.remote}: {str(e)}")
+            logger.error(
-            logger.error(f"Invalid JSON to /hook from {request.remote}: {str(e)}")
+                f"{current_time}\tInvalid request method to /hook from %s",
-            return web.Response(status=400, text="Invalid JSON in request body")
+                request.remote,
-
+            )
-        # Log the received webhook with detailed output
+            return web.Response(status=405, text="Method not allowed. Use POST.")
-        print(f"\n{'='*60}")
+
-        print(f"[@DEBUG] WEBHOOK RECEIVED at {current_time}")
+        # Read raw body for signature verification
-        print(f"[@DEBUG] From: {request.remote}")
+        body = await request.read()
-        print(f"[@DEBUG] Headers: {dict(request.headers)}")
+        body_str = body.decode("utf-8")
-        print(f"[@DEBUG] Data type: {data.get('type', 'unknown')}")
+
-        print(f"[@DEBUG] Event ID: {data.get('event_id', 'none')}")
+        # Verify Square webhook signature using Square SDK
-        print(f"[@DEBUG] Full payload:")
+        signature_header = request.headers.get("x-square-hmacsha256-signature", "")
-        print(json.dumps(data, indent=2))
+
-        print(f"{'='*60}\n")
+        if SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL:
-
+            if not signature_header:
-        logger.info(
+                logger.error(
-            f"{current_time}\tWebhook received from {request.remote}: {json.dumps(data)}"
+                    f"{current_time}\tMissing signature header from {request.remote}"
-        )
+                )
-
+                print(f"[@DEBUG] Missing x-square-hmacsha256-signature header")
-        # Process the webhook data here
+                return web.Response(status=401, text="Unauthorized: Missing signature")
-        # TODO: Add your webhook processing logic
+
-        webhook_type = data.get("type", "unknown")
+            # Use Square SDK's official validation function
-
+            is_valid = verify_signature(
-        if webhook_type == "inventory.count.updated":
+                request_body=body_str,
-            print(f"[@DEBUG] Processing inventory count update...")
+                signature_header=signature_header,
-            # Extract inventory data
+                signature_key=SQUARE_WEBHOOK_SIGNATURE_KEY,
-            inventory_counts = (
+                notification_url=SQUARE_WEBHOOK_URL,
-                data.get("data", {}).get("object", {}).get("inventory_counts", [])
+            )
-            )
+
-            for count in inventory_counts:
+            if not is_valid:
-                print(f"[@DEBUG]   Catalog Object: {count.get('catalog_object_id')}")
+                logger.error(f"{current_time}\tInvalid signature from {request.remote}")
-                print(f"[@DEBUG]   Quantity: {count.get('quantity')}")
+                print(f"[@DEBUG] Invalid webhook signature!")
-                print(f"[@DEBUG]   State: {count.get('state')}")
+                return web.Response(status=401, text="Unauthorized: Invalid signature")
-
+
-        response_data = {
+            print(f"[@DEBUG] ✓ Webhook signature verified")
-            "status": "success",
+        else:
-            "message": f"Webhook received: {webhook_type}",
+            print(
-            "timestamp": current_time,
+                f"[@DEBUG] ⚠️  Webhook signature verification DISABLED (set SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL in .env)"
-            "event_id": data.get("event_id"),
+            )
-            "data_received": data,
+
-        }
+        # Parse JSON data
-
+        try:
-        return web.json_response(response_data)
+            data = json.loads(body_str)
-
+        except json.JSONDecodeError as e:
-    except json.JSONDecodeError as e:
+            print(f"[@DEBUG] Invalid JSON to /hook from {request.remote}: {str(e)}")
-        error_msg = f"JSON decode error in webhook: {str(e)}"
+            logger.error(f"Invalid JSON to /hook from {request.remote}: {str(e)}")
-        print(f"[@DEBUG] {error_msg}")
+            return web.Response(status=400, text="Invalid JSON in request body")
-        logger.error(error_msg)
+
-        return web.Response(status=400, text=error_msg)
+        # Check for idempotency (event_id)
-    except Exception as e:
+        event_id = data.get("event_id")
-        error_msg = f"Webhook handler error: {str(e)}"
+        if event_id:
-        print(f"[@DEBUG] {error_msg}")
+            if event_id in processed_event_ids:
-        print(f"[@DEBUG] Exception type: {type(e).__name__}")
+                print(
-        import traceback
+                    f"[@DEBUG] Duplicate event {event_id} - already processed (idempotency)"
-
+                )
-        print(f"[@DEBUG] Traceback:\n{traceback.format_exc()}")
+                logger.info(
-        logger.error(f"{error_msg}\n{traceback.format_exc()}")
+                    f"{current_time}\tDuplicate event {event_id} from {request.remote}"
-        return web.Response(status=500, text="Internal server error")
+                )
-
+                return web.json_response(
-
+                    {
-async def request_handler(request: web.Request) -> web.Response:
+                        "status": "success",
-    """Handle incoming POST requests with command processing"""
+                        "message": "Event already processed (idempotency)",
-    try:
+                        "event_id": event_id,
-        # Verify request method
+                    }
-        if request.method != "POST":
+                )
-            logger.error(f"{time}\tInvalid request method from %s at ", request.remote)
+
-            return web.Response(status=405, text="Method not allowed. Use POST.")
+            # Add to processed set
-
+            processed_event_ids.add(event_id)
-        # Get JSON data from request
+
-        try:
+            # Prevent memory growth - remove oldest if too many
-            data = await request.json()
+            if len(processed_event_ids) > MAX_STORED_EVENT_IDS:
-        except json.JSONDecodeError:
+                processed_event_ids.pop()
-            logger.error("Invalid JSON from %s", request.remote)
+
-            return web.Response(
+        # Get event info for quick response
-                status=400, text="{time()}\tInvalid JSON in request body"
+        webhook_type = data.get("type", "unknown")
-            )
+        event_id_val = data.get("event_id", "none")
-        # Validate Idempotency-Key header
+
-        if valid_uuid(request.headers.get("Idempotency-Key")) is False:
+        # RESPOND IMMEDIATELY - Square requires fast response (< 2 seconds)
-            logger.error("{time()}\tInvalid Idempotency-Key from %s", request.remote)
+        # Don't do any processing before responding!
-            return web.Response(status=412, text="412 Headers rejected!")
+        response_data = {
-        # Validate User-Agent header
+            "status": "success",
-        if not re.match(UA, str(request.headers.get("User-Agent"))):
+            "message": "Webhook received",
-            logger.error("{time()}\tInvalid User-Agent from %s", request.remote)
+            "event_id": event_id_val,
-            return web.Response(status=412, text="412 Headers rejected!")
+        }
-        # Validate command exists
+
-        if "CMD" not in data:
+        # Log and process AFTER sending response (async background tasks)
-            logger.error("{time()}\tMissing CMD field from %s", request.remote)
+        async def log_and_process():
-            return web.Response(status=400, text="Missing CMD field in request")
+            # Log the received webhook with detailed output
-
+            print(f"\n{'='*60}")
-        # Decode and process command
+            print(f"[@DEBUG] WEBHOOK RECEIVED at {current_time}")
-        try:
+            print(f"[@DEBUG] From: {request.remote}")
-            decoded_cmd = b64decode(data["CMD"]).decode("utf-8")
+            print(f"[@DEBUG] Headers: {dict(request.headers)}")
-            print(
+            print(f"[@DEBUG] Data type: {webhook_type}")
-                f"""At: {time}\nReceived command: {decoded_cmd}\nUser-Agent: {request.headers.get('User-Agent')}\nIdempotency-Key: {request.headers.get('Idempotency-Key')}"""
+            print(f"[@DEBUG] Event ID: {event_id_val}")
-            )
+            print(f"[@DEBUG] Full payload:")
-            logging.info(
+            print(json.dumps(data, indent=2))
-                f"{time}\tCommand: {decoded_cmd}\tFrom: {request.remote}\tUA: {request.headers.get('User-Agent')}\tIdempotency-Key: {request.headers.get('Idempotency-Key')}"
+            print(f"{'='*60}\n")
-            )
+
-
+            logger.info(
-            # Process the command
+                f"{current_time}\tWebhook received from {request.remote}: {json.dumps(data)}"
-            if decoded_cmd == "RUNUPDATE":
+            )
-                response_data = {"status": "success", "message": "Update started"}
+
-                await run_update()
+            # Process the webhook data
-                logging.info(f"{time}\tUpdate command processed successfully")
+            if webhook_type == "inventory.count.updated":
-                return web.json_response(response_data)
+                print(f"[@DEBUG] Processing inventory count update...")
-            else:
+                # Extract inventory data
-                return web.Response(status=400, text=f"Unknown command: {decoded_cmd}")
+                inventory_counts = (
-
+                    data.get("data", {}).get("object", {}).get("inventory_counts", [])
-        except Exception as e:
+                )
-            print(f"Error processing command: {str(e)}")
+                for count in inventory_counts:
-            return web.Response(status=500, text=f"Server error: {str(e)}")
+                    print(
-
+                        f"[@DEBUG]   Catalog Object: {count.get('catalog_object_id')}"
-    except Exception as e:
+                    )
-        print(f"Request handler error: {str(e)}")
+                    print(f"[@DEBUG]   Quantity: {count.get('quantity')}")
-        return web.Response(status=500, text="Internal server error")
+                    print(f"[@DEBUG]   State: {count.get('state')}")
-
+
-
+        # Start background processing (don't wait for it)
-print("Starting web server...")
+        asyncio.create_task(log_and_process())
-os.system("cls" if os.name == "nt" else "clear")
+
-app = web.Application()
+        # Return response immediately
-app.router.add_post("/api", request_handler)
+        return web.json_response(response_data)
-app.router.add_post("/hook", hook_handler)
+
-
+    except json.JSONDecodeError as e:
-parser = argparse.ArgumentParser(
+        error_msg = f"JSON decode error in webhook: {str(e)}"
-    description="studspider - LEGO Inventory Sync Service for Square POS and Brickset",
+        print(f"[@DEBUG] {error_msg}")
-    epilog="""
+        logger.error(error_msg)
-Examples:
+        return web.Response(status=400, text=error_msg)
-  # Basic HTTP server (recommended with nginx reverse proxy)
+    except Exception as e:
-  python src/test_server.py --host 0.0.0.0 --port 7432
+        error_msg = f"Webhook handler error: {str(e)}"
-
+        print(f"[@DEBUG] {error_msg}")
-  # With separate webhook port
+        print(f"[@DEBUG] Exception type: {type(e).__name__}")
-  python src/test_server.py --port 7432 --hook-port 8080
+        import traceback
-
+
-  # HTTPS using certificates from .env (set SSL_CERT_PATH and SSL_KEY_PATH)
+        print(f"[@DEBUG] Traceback:\n{traceback.format_exc()}")
-  python src/test_server.py --hook-port 8443 --hook-ssl-only
+        logger.error(f"{error_msg}\n{traceback.format_exc()}")
-
+        return web.Response(status=500, text="Internal server error")
-  # 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
+
-
+async def request_handler(request: web.Request) -> web.Response:
-  # Full HTTPS (both endpoints)
+    """Handle incoming POST requests with command processing"""
-  python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443
+    try:
-
+        # Verify request method
-  # Production behind nginx (no SSL needed in app)
+        if request.method != "POST":
-  python src/test_server.py --host 127.0.0.1 --port 7432
+            logger.error(f"{time}\tInvalid request method from %s at ", request.remote)
-
+            return web.Response(status=405, text="Method not allowed. Use POST.")
-Environment Variables (.env):
+
-  SSL_CERT_PATH       - Default SSL certificate path
+        # Get JSON data from request
-  SSL_KEY_PATH        - Default SSL key path  
+        try:
-  SQUARE_P7B_PATH     - P7B file to upload to Square Dashboard
+            data = await request.json()
-
+        except json.JSONDecodeError:
-For more information, see doc/ 
+            logger.error("Invalid JSON from %s", request.remote)
-    """,
+            return web.Response(
-    formatter_class=argparse.RawDescriptionHelpFormatter,
+                status=400, text="{time()}\tInvalid JSON in request body"
-)
+            )
-parser.add_argument(
+        # Validate Idempotency-Key header
-    "--host",
+        if valid_uuid(request.headers.get("Idempotency-Key")) is False:
-    type=str,
+            logger.error("{time()}\tInvalid Idempotency-Key from %s", request.remote)
-    default="localhost",
+            return web.Response(status=412, text="412 Headers rejected!")
-    help="IP address to listen on (default: localhost, use 0.0.0.0 for all interfaces)",
+        # Validate User-Agent header
-)
+        if not re.match(UA, str(request.headers.get("User-Agent"))):
-parser.add_argument(
+            logger.error("{time()}\tInvalid User-Agent from %s", request.remote)
-    "--port", type=int, default=7432, help="Port for main API endpoint (default: 7432)"
+            return web.Response(status=412, text="412 Headers rejected!")
-)
+        # Validate command exists
-parser.add_argument(
+        if "CMD" not in data:
-    "--hook-port",
+            logger.error("{time()}\tMissing CMD field from %s", request.remote)
-    type=int,
+            return web.Response(status=400, text="Missing CMD field in request")
-    default=None,
+
-    help="Separate port for webhook endpoint (default: None - uses same port as --port)",
+        # Decode and process command
-)
+        try:
-parser.add_argument(
+            decoded_cmd = b64decode(data["CMD"]).decode("utf-8")
-    "--ssl-cert",
+            print(
-    type=str,
+                f"""At: {time}\nReceived command: {decoded_cmd}\nUser-Agent: {request.headers.get('User-Agent')}\nIdempotency-Key: {request.headers.get('Idempotency-Key')}"""
-    default=get_sq_env.SSL_CERT_PATH,
+            )
-    help="Path to SSL certificate file in PEM format (default: from .env SSL_CERT_PATH)",
+            logging.info(
-)
+                f"{time}\tCommand: {decoded_cmd}\tFrom: {request.remote}\tUA: {request.headers.get('User-Agent')}\tIdempotency-Key: {request.headers.get('Idempotency-Key')}"
-parser.add_argument(
+            )
-    "--ssl-key",
+
-    type=str,
+            # Process the command
-    default=get_sq_env.SSL_KEY_PATH,
+            if decoded_cmd == "RUNUPDATE":
-    help="Path to SSL private key file in PEM format (default: from .env SSL_KEY_PATH)",
+                response_data = {"status": "success", "message": "Update started"}
-)
+                await run_update()
-parser.add_argument(
+                logging.info(f"{time}\tUpdate command processed successfully")
-    "--hook-ssl-only",
+                return web.json_response(response_data)
-    action="store_true",
+            else:
-    help="Enable HTTPS only for webhook endpoint, main API stays HTTP (requires --ssl-cert and --ssl-key)",
+                return web.Response(status=400, text=f"Unknown command: {decoded_cmd}")
-)
+
-args = parser.parse_args()
+        except Exception as e:
-
+            print(f"Error processing command: {str(e)}")
-if __name__ == "__main__":
+            return web.Response(status=500, text=f"Server error: {str(e)}")
-    print("Creating log directory if it doesn't exist...")
+
-    os.makedirs("./logs/", exist_ok=True)
+    except Exception as e:
-    print(banner.generate("studspider v0.1"))
+        print(f"Request handler error: {str(e)}")
-    print("Built for PLAY-WELL TEKNOLOGIES by contact@studspider.sh\n\n")
+        return web.Response(status=500, text="Internal server error")
-
+
-    # Setup SSL context if certificates provided
+
-    ssl_context = None
+print("Starting web server...")
-    if args.ssl_cert and args.ssl_key:
+os.system("cls" if os.name == "nt" else "clear")
-        # Check if trying to use P7B file
+app = web.Application()
-        if args.ssl_cert.endswith(".p7b") or args.ssl_key.endswith(".p7b"):
+app.router.add_post("/api", request_handler)
-            print("ERROR: Cannot use P7B files directly for HTTPS server!")
+app.router.add_post("/hook", hook_handler)
-            print("\nP7B files are for Square's webhook signature verification only.")
+
-            print("Your server needs PEM format files (.pem)")
+parser = argparse.ArgumentParser(
-            print("\nWorkflow:")
+    description="studspider - LEGO Inventory Sync Service for Square POS and Brickset",
-            print("1. Generate PEM files: python generate_cert.py")
+    epilog="""
-            print("2. Convert to P7B for Square: python convert_to_p7b.py")
+Examples:
-            print("3. Upload P7B to Square Dashboard")
+  # Basic HTTP server (recommended with nginx reverse proxy)
-            print("4. Run server with PEM files:")
+  python src/test_server.py --host 0.0.0.0 --port 7432
-            print(
+
-                "   python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443"
+  # With separate webhook port
-            )
+  python src/test_server.py --port 7432 --hook-port 8080
-            print("\nIf you have a P7B and need PEM:")
+
-            print("   python convert_from_p7b.py your_cert.p7b")
+  # HTTPS using certificates from .env (set SSL_CERT_PATH and SSL_KEY_PATH)
-            exit(1)
+  python src/test_server.py --hook-port 8443 --hook-ssl-only
-
+
-        try:
+  # HTTPS with command line override (self-signed cert for testing)
-            ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
+  python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only
-            ssl_context.load_cert_chain(args.ssl_cert, args.ssl_key)
+
-            print(f"✓ SSL enabled with cert: {args.ssl_cert}")
+  # Full HTTPS (both endpoints)
-
+  python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443
-            # Show P7B info if configured
+
-            if get_sq_env.SQUARE_P7B_PATH:
+  # Production behind nginx (no SSL needed in app)
-                print(
+  python src/test_server.py --host 127.0.0.1 --port 7432
-                    f"ℹ️  Square P7B certificate for upload: {get_sq_env.SQUARE_P7B_PATH}"
+
-                )
+Environment Variables (.env):
-                if os.path.exists(get_sq_env.SQUARE_P7B_PATH):
+  SSL_CERT_PATH       - Default SSL certificate path
-                    print(f"   ✓ P7B file exists")
+  SSL_KEY_PATH        - Default SSL key path
-                else:
+
-                    print(f"   ⚠️  P7B file not found - run: python convert_to_p7b.py")
+For more information, see doc/ 
-        except FileNotFoundError as e:
+    """,
-            print(f"ERROR: Certificate file not found: {e}")
+    formatter_class=argparse.RawDescriptionHelpFormatter,
-            exit(1)
+)
-        except ssl.SSLError as e:
+parser.add_argument(
-            print(f"ERROR: Invalid certificate or key file: {e}")
+    "--host",
-            print("\nMake sure you're using PEM format files.")
+    type=str,
-            print("Generate certificates with: python generate_cert.py")
+    default="localhost",
-            exit(1)
+    help="IP address to listen on (default: localhost, use 0.0.0.0 for all interfaces)",
-    elif args.ssl_cert or args.ssl_key:
+)
-        print("ERROR: Both --ssl-cert and --ssl-key must be provided together")
+parser.add_argument(
-        exit(1)
+    "--port", type=int, default=7432, help="Port for main API endpoint (default: 7432)"
-    else:
+)
-        print("ℹ️  Running without SSL (HTTP only)")
+parser.add_argument(
-        print("   For production with nginx reverse proxy, this is recommended.")
+    "--hook-port",
-        print("   Nginx will handle HTTPS, forward HTTP to this server.")
+    type=int,
-
+    default=None,
-        # Show P7B info even without SSL (for nginx setup)
+    help="Separate port for webhook endpoint (default: None - uses same port as --port)",
-        if get_sq_env.SQUARE_P7B_PATH:
+)
-            print(
+parser.add_argument(
-                f"\nℹ️  Square P7B certificate configured: {get_sq_env.SQUARE_P7B_PATH}"
+    "--ssl-cert",
-            )
+    type=str,
-            if os.path.exists(get_sq_env.SQUARE_P7B_PATH):
+    default=get_sq_env.SSL_CERT_PATH,
-                print(f"   ✓ P7B file exists - upload this to Square Dashboard")
+    help="Path to SSL certificate file in PEM format (default: from .env SSL_CERT_PATH)",
-                print(f"   Square Dashboard → Webhooks → Manage Notification Settings")
+)
-            else:
+parser.add_argument(
-                print(f"   ⚠️  P7B file not found")
+    "--ssl-key",
-                print(
+    type=str,
-                    f"   Generate with: python convert_to_p7b.py --cert /path/to/nginx/cert.pem"
+    default=get_sq_env.SSL_KEY_PATH,
-                )
+    help="Path to SSL private key file in PEM format (default: from .env SSL_KEY_PATH)",
-
+)
-    # If hook-port is specified, run a separate server for webhooks
+parser.add_argument(
-    if args.hook_port:
+    "--hook-ssl-only",
-        protocol_main = "https" if (ssl_context and not args.hook_ssl_only) else "http"
+    action="store_true",
-        protocol_hook = "https" if ssl_context else "http"
+    help="Enable HTTPS only for webhook endpoint, main API stays HTTP (requires --ssl-cert and --ssl-key)",
-
+)
-        # Create separate app for webhook
+args = parser.parse_args()
-        hook_app = web.Application()
+
-        hook_app.router.add_post("/hook", hook_handler)
+if __name__ == "__main__":
-
+    print("Creating log directory if it doesn't exist...")
-        # Run both servers concurrently
+    os.makedirs("./logs/", exist_ok=True)
-        async def run_servers():
+    print(banner.generate("studspider v0.1"))
-            runner1 = web.AppRunner(app)
+    print("Built for PLAY-WELL TEKNOLOGIES by contact@studspider.sh\n\n")
-            await runner1.setup()
+
-            # Main server uses SSL only if not hook-ssl-only mode
+    # Setup SSL context if certificates provided
-            main_ssl = None if args.hook_ssl_only else ssl_context
+    ssl_context = None
-            site1 = web.TCPSite(runner1, args.host, args.port, ssl_context=main_ssl)
+    if args.ssl_cert and args.ssl_key:
-
+        # Check if trying to use P7B file
-            runner2 = web.AppRunner(hook_app)
+        if args.ssl_cert.endswith(".p7b") or args.ssl_key.endswith(".p7b"):
-            await runner2.setup()
+            print("ERROR: Cannot use P7B files directly for HTTPS server!")
-            # Webhook server always uses SSL if certificates provided
+            print("\nP7B files are for Square's webhook signature verification only.")
-            site2 = web.TCPSite(
+            print("Your server needs PEM format files (.pem)")
-                runner2, args.host, args.hook_port, ssl_context=ssl_context
+            print("\nWorkflow:")
-            )
+            print("1. Generate PEM files: python generate_cert.py")
-
+            print("2. Convert to P7B for Square: python convert_to_p7b.py")
-            await site1.start()
+            print("3. Upload P7B to Square Dashboard")
-            await site2.start()
+            print("4. Run server with PEM files:")
-
+            print(
-            print(f"Starting main server on {args.host}:{args.port} ({protocol_main})")
+                "   python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443"
-            print(
+            )
-                f"Starting webhook server on {args.host}:{args.hook_port} ({protocol_hook})"
+            print("\nIf you have a P7B and need PEM:")
-            )
+            print("   python convert_from_p7b.py your_cert.p7b")
-
+            exit(1)
-            # Keep running
+
-            await asyncio.Event().wait()
+        try:
-
+            ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
-        asyncio.run(run_servers())
+            ssl_context.load_cert_chain(args.ssl_cert, args.ssl_key)
-    else:
+            print(f"✓ SSL enabled with cert: {args.ssl_cert}")
-        protocol = "https" if ssl_context else "http"
+        except FileNotFoundError as e:
-        print(f"Server running on {args.host}:{args.port} ({protocol})")
+            print(f"ERROR: Certificate file not found: {e}")
-        print(f"Webhook endpoint: {protocol}://{args.host}:{args.port}/hook")
+            exit(1)
-        web.run_app(app, host=args.host, port=args.port, ssl_context=ssl_context)
+        except ssl.SSLError as e:
+            print(f"ERROR: Invalid certificate or key file: {e}")
+            print("\nMake sure you're using PEM format files.")
+            exit(1)
+    elif args.ssl_cert or args.ssl_key:
+        print("ERROR: Both --ssl-cert and --ssl-key must be provided together")
+        exit(1)
+    else:
+        print("ℹ️  Running without SSL (HTTP only)")
+        print("   For production with nginx reverse proxy, this is recommended.")
+        print("   Nginx will handle HTTPS, forward HTTP to this server.")
+
+    # If hook-port is specified, run a separate server for webhooks
+    if args.hook_port:
+        protocol_main = "https" if (ssl_context and not args.hook_ssl_only) else "http"
+        protocol_hook = "https" if ssl_context else "http"
+
+        # Create separate app for webhook
+        hook_app = web.Application()
+        hook_app.router.add_post("/hook", hook_handler)
+
+        # Run both servers concurrently
+        async def run_servers():
+            runner1 = web.AppRunner(app)
+            await runner1.setup()
+            # Main server uses SSL only if not hook-ssl-only mode
+            main_ssl = None if args.hook_ssl_only else ssl_context
+            site1 = web.TCPSite(runner1, args.host, args.port, ssl_context=main_ssl)
+
+            runner2 = web.AppRunner(hook_app)
+            await runner2.setup()
+            # Webhook server always uses SSL if certificates provided
+            site2 = web.TCPSite(
+                runner2, args.host, args.hook_port, ssl_context=ssl_context
+            )
+
+            await site1.start()
+            await site2.start()
+
+            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)

tests/test_square_webhook.py

@@ -0,0 +1,51 @@
+"""
+Simple test script to verify Square webhook signature verification works correctly.
+
+This uses the same test data from Square's documentation to ensure our implementation
+matches their expected behavior.
+"""
+
+import sys
+
+sys.path.insert(0, "src")
+
+from square.utils.webhooks_helper import verify_signature
+
+# Test data from Square's documentation
+TEST_BODY = '{"hello":"world"}'
+TEST_SIGNATURE = "2kRE5qRU2tR+tBGlDwMEw2avJ7QM4ikPYD/PJ3bd9Og="
+TEST_SIGNATURE_KEY = "asdf1234"
+TEST_NOTIFICATION_URL = "https://example.com/webhook"
+
+
+def test_signature_validation():
+    """Test that signature validation works with Square SDK"""
+
+    print("Testing Square webhook signature validation...")
+    print(f"Body: {TEST_BODY}")
+    print(f"Signature: {TEST_SIGNATURE}")
+    print(f"Key: {TEST_SIGNATURE_KEY}")
+    print(f"URL: {TEST_NOTIFICATION_URL}")
+    print()
+
+    # This should return True with Square's test data
+    is_valid = verify_signature(
+        request_body=TEST_BODY,
+        signature_header=TEST_SIGNATURE,
+        signature_key=TEST_SIGNATURE_KEY,
+        notification_url=TEST_NOTIFICATION_URL,
+    )
+
+    if is_valid:
+        print("✅ SUCCESS: Signature validation passed!")
+        print("The webhook implementation is using Square SDK correctly.")
+        return True
+    else:
+        print("❌ FAILURE: Signature validation failed!")
+        print("Something is wrong with the signature verification.")
+        return False
+
+
+if __name__ == "__main__":
+    success = test_signature_validation()
+    sys.exit(0 if success else 1)

uv.lock

@@ -92,6 +92,15 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/4d/3f/3bc3f1d83f6e4a7fcb834d3720544ca597590425be5ba9db032b2bf322a2/altgraph-0.17.4-py2.py3-none-any.whl", hash = "sha256:642743b4750de17e655e6711601b077bc6598dbfa3ba5fa2b2a35ce12b508dff", size = 21212, upload-time = "2023-09-25T09:04:50.691Z" },
 ]
 
+[[package]]
+name = "annotated-doc"
+version = "0.0.4"
+source = { registry = "https://pypi.org/simple" }
+sdist = { url = "https://files.pythonhosted.org/packages/57/ba/046ceea27344560984e26a590f90bc7f4a75b06701f653222458922b558c/annotated_doc-0.0.4.tar.gz", hash = "sha256:fbcda96e87e9c92ad167c2e53839e57503ecfda18804ea28102353485033faa4", size = 7288, upload-time = "2025-11-10T22:07:42.062Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/1e/d3/26bf1008eb3d2daa8ef4cacc7f3bfdc11818d111f7e2d0201bc6e3b49d45/annotated_doc-0.0.4-py3-none-any.whl", hash = "sha256:571ac1dc6991c450b25a9c2d84a3705e2ae7a53467b5d111c24fa8baabbed320", size = 5303, upload-time = "2025-11-10T22:07:40.673Z" },
+]
+
 [[package]]
 name = "annotated-types"
 version = "0.7.0"
@@ -239,6 +248,21 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/b2/b7/545d2c10c1fc15e48653c91efde329a790f2eecfbbf2bd16003b5db2bab0/dotenv-0.9.9-py2.py3-none-any.whl", hash = "sha256:29cf74a087b31dafdb5a446b6d7e11cbce8ed2741540e2339c69fbef92c94ce9", size = 1892, upload-time = "2025-02-19T22:15:01.647Z" },
 ]
 
+[[package]]
+name = "fastapi"
+version = "0.128.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "annotated-doc" },
+    { name = "pydantic" },
+    { name = "starlette" },
+    { name = "typing-extensions" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/52/08/8c8508db6c7b9aae8f7175046af41baad690771c9bcde676419965e338c7/fastapi-0.128.0.tar.gz", hash = "sha256:1cc179e1cef10a6be60ffe429f79b829dce99d8de32d7acb7e6c8dfdf7f2645a", size = 365682, upload-time = "2025-12-27T15:21:13.714Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/5c/05/5cbb59154b093548acd0f4c7c474a118eda06da25aa75c616b72d8fcd92a/fastapi-0.128.0-py3-none-any.whl", hash = "sha256:aebd93f9716ee3b4f4fcfe13ffb7cf308d99c9f3ab5622d8877441072561582d", size = 103094, upload-time = "2025-12-27T15:21:12.154Z" },
+]
+
 [[package]]
 name = "frozenlist"
 version = "1.7.0"
@@ -330,9 +354,23 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/ee/43/3cecdc0349359e1a527cbf2e3e28e5f8f06d3343aaf82ca13437a9aa290f/greenlet-3.2.4-cp313-cp313-manylinux_2_24_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:23768528f2911bcd7e475210822ffb5254ed10d71f4028387e5a99b4c6699671", size = 610497, upload-time = "2025-08-07T13:18:31.636Z" },
     { url = "https://files.pythonhosted.org/packages/b8/19/06b6cf5d604e2c382a6f31cafafd6f33d5dea706f4db7bdab184bad2b21d/greenlet-3.2.4-cp313-cp313-musllinux_1_1_aarch64.whl", hash = "sha256:00fadb3fedccc447f517ee0d3fd8fe49eae949e1cd0f6a611818f4f6fb7dc83b", size = 1121662, upload-time = "2025-08-07T13:42:41.117Z" },
     { url = "https://files.pythonhosted.org/packages/a2/15/0d5e4e1a66fab130d98168fe984c509249c833c1a3c16806b90f253ce7b9/greenlet-3.2.4-cp313-cp313-musllinux_1_1_x86_64.whl", hash = "sha256:d25c5091190f2dc0eaa3f950252122edbbadbb682aa7b1ef2f8af0f8c0afefae", size = 1149210, upload-time = "2025-08-07T13:18:24.072Z" },
+    { url = "https://files.pythonhosted.org/packages/1c/53/f9c440463b3057485b8594d7a638bed53ba531165ef0ca0e6c364b5cc807/greenlet-3.2.4-cp313-cp313-musllinux_1_2_aarch64.whl", hash = "sha256:6e343822feb58ac4d0a1211bd9399de2b3a04963ddeec21530fc426cc121f19b", size = 1564759, upload-time = "2025-11-04T12:42:19.395Z" },
+    { url = "https://files.pythonhosted.org/packages/47/e4/3bb4240abdd0a8d23f4f88adec746a3099f0d86bfedb623f063b2e3b4df0/greenlet-3.2.4-cp313-cp313-musllinux_1_2_x86_64.whl", hash = "sha256:ca7f6f1f2649b89ce02f6f229d7c19f680a6238af656f61e0115b24857917929", size = 1634288, upload-time = "2025-11-04T12:42:21.174Z" },
     { url = "https://files.pythonhosted.org/packages/0b/55/2321e43595e6801e105fcfdee02b34c0f996eb71e6ddffca6b10b7e1d771/greenlet-3.2.4-cp313-cp313-win_amd64.whl", hash = "sha256:554b03b6e73aaabec3745364d6239e9e012d64c68ccd0b8430c64ccc14939a8b", size = 299685, upload-time = "2025-08-07T13:24:38.824Z" },
 ]
 
+[[package]]
+name = "gunicorn"
+version = "23.0.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "packaging" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/34/72/9614c465dc206155d93eff0ca20d42e1e35afc533971379482de953521a4/gunicorn-23.0.0.tar.gz", hash = "sha256:f014447a0101dc57e294f6c18ca6b40227a4c90e9bdb586042628030cba004ec", size = 375031, upload-time = "2024-08-10T20:25:27.378Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/cb/7d/6dac2a6e1eba33ee43f318edbed4ff29151a49b5d37f080aad1e6469bca4/gunicorn-23.0.0-py3-none-any.whl", hash = "sha256:ec400d38950de4dfd418cff8328b2c8faed0edb0d517d3394e457c317908ca4d", size = 85029, upload-time = "2024-08-10T20:25:24.996Z" },
+]
+
 [[package]]
 name = "h11"
 version = "0.16.0"
@@ -949,11 +987,14 @@ dependencies = [
     { name = "asyncio-throttle" },
     { name = "bs4" },
     { name = "dotenv" },
+    { name = "fastapi" },
+    { name = "gunicorn" },
     { name = "pyinstaller" },
     { name = "pytest" },
     { name = "reflex" },
     { name = "requests" },
     { name = "squareup" },
+    { name = "uvicorn" },
 ]
 
 [package.metadata]
@@ -964,11 +1005,14 @@ requires-dist = [
     { name = "asyncio-throttle", specifier = ">=1.0.2,<2.0.0" },
     { name = "bs4", specifier = ">=0.0.2" },
     { name = "dotenv", specifier = ">=0.9.9,<0.10.0" },
+    { name = "fastapi", specifier = ">=0.128.0" },
+    { name = "gunicorn", specifier = ">=23.0.0" },
     { name = "pyinstaller", specifier = ">=6.14.1,<7.0.0" },
     { name = "pytest", specifier = ">=8.4.2,<9.0.0" },
     { name = "reflex", specifier = ">=0.8.14.post1" },
     { name = "requests", specifier = ">=2.32.4,<3.0.0" },
     { name = "squareup", specifier = ">=43.0.0.20250618,<44.0.0.0" },
+    { name = "uvicorn", specifier = ">=0.40.0" },
 ]
 
 [[package]]
@@ -1022,6 +1066,19 @@ wheels = [
     { url = "https://files.pythonhosted.org/packages/a7/c2/fe1e52489ae3122415c51f387e221dd0773709bad6c6cdaa599e8a2c5185/urllib3-2.5.0-py3-none-any.whl", hash = "sha256:e6b01673c0fa6a13e374b50871808eb3bf7046c4b125b216f6bf1cc604cff0dc", size = 129795, upload-time = "2025-06-18T14:07:40.39Z" },
 ]
 
+[[package]]
+name = "uvicorn"
+version = "0.40.0"
+source = { registry = "https://pypi.org/simple" }
+dependencies = [
+    { name = "click" },
+    { name = "h11" },
+]
+sdist = { url = "https://files.pythonhosted.org/packages/c3/d1/8f3c683c9561a4e6689dd3b1d345c815f10f86acd044ee1fb9a4dcd0b8c5/uvicorn-0.40.0.tar.gz", hash = "sha256:839676675e87e73694518b5574fd0f24c9d97b46bea16df7b8c05ea1a51071ea", size = 81761, upload-time = "2025-12-21T14:16:22.45Z" }
+wheels = [
+    { url = "https://files.pythonhosted.org/packages/3d/d8/2083a1daa7439a66f3a48589a57d576aa117726762618f6bb09fe3798796/uvicorn-0.40.0-py3-none-any.whl", hash = "sha256:c6c8f55bc8bf13eb6fa9ff87ad62308bbbc33d0b67f84293151efe87e0d5f2ee", size = 68502, upload-time = "2025-12-21T14:16:21.041Z" },
+]
+
 [[package]]
 name = "watchfiles"
 version = "1.1.1"