name changes

.env.example

@@ -0,0 +1,22 @@
+# 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_ENV = "sandbox"  # or "production"
+
+SQUARE_WEBHOOK_URL = "https://yourdomain.com/hook"
+
+# Sandbox credentials - test environment (fake data, safe for development)
+SANDBOX_SQUARE_APP_ID = "your_sandbox_app_id"
+SANDBOX_SQUARE_TOKEN = "your_sandbox_access_token"
+SANDBOX_SQUARE_WEBHOOK_INVENTORY_ID = "your_sandbox_webhook_signature_key"
+
+# Production credentials - live store (real inventory and sales)
+PROD_SQUARE_APP_ID = "your_production_app_id"
+PROD_SQUARE_TOKEN = "your_production_access_token"
+PROD_SQUARE_WEBHOOK_INVENTORY_ID = "your_production_webhook_signature_key"
+
+# Category ID for LEGO items in Square catalog (find in Dashboard → Items → Categories)
+SQUARE_LEGO_CAT_ID = "your_lego_category_id"

.gitignore

@@ -1,25 +1,43 @@
-logs/
-.env
-__pycache__/
-*.pyc
-.vscode/
-.idea/
-.DS_Store
-*.db
-*.sqlite3
-*.log
-poetry.lock
-bin/build/
-.zip
-snippet.py
-doc/square-src/square-python-sdk-43.0.2.20250820/
-
-doc/square-src/
-
-doc/pw-inventory/pw-inventory.zip
-
-doc/pw-inventory/SQUARE_inventory_forautomation_2025-06-18-0139.csv
-
-doc/pw-inventory/square_inventory_export_2025-06-18-0137.xlsx
-
-square-src.code-workspace
+logs/
+.env
+__pycache__/
+*.pyc
+.vscode/
+.idea/
+.DS_Store
+*.db
+*.sqlite3
+*.log
+bin/build/
+.zip
+snippet.py
+doc/square-src/square-python-sdk-43.0.2.20250820/
+
+doc/square-src/
+
+doc/pw-inventory/pw-inventory.zip
+
+doc/pw-inventory/SQUARE_inventory_forautomation_2025-06-18-0139.csv
+
+doc/pw-inventory/square_inventory_export_2025-06-18-0137.xlsx
+
+square-src.code-workspace
+
+doc/pw-inventory/
+
+studspider.db-journal
+.venv
+*.db
+*.py[cod]
+.web
+.states
+assets/external/
+__pycache__/
+win/
+.github/
+bin/
+activate/
+build/
+.pem
+.p7b
+nginx/

README.md

@@ -1,11 +1,44 @@
-# Edderkop
-
-environment: poetry
-code style: black
-
-built with
-
-- aiohttp
-- squareup
-- apscheduler
+# studspider
+
+![Static Badge](https://img.shields.io/badge/python-3.13.7-blue?logo=python) ![Static Badge](https://img.shields.io/badge/code_style-black-black) ![Static Badge](https://img.shields.io/badge/license-MPL-orange)
+
+
+
+
+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 branch 
+```
+
+Go to the project directory
+
+```bash
+  cd studspider
+```
+
+Install dependencies
+
+```bash
+  uv sync && uv lock
+```
+
+Start the server
+
+```bash
+  uv run studspider.py
+```
+
+
+## Tech 
+
+- gunicorn
+- uvicorn
+- aiohttp
+- squareup
+- apscheduler
 - aiosqlite

TODO.md

@@ -1,4 +1,11 @@
 - glue shit together
 - test sq
 - test inserting brickowl into sq
-- test kopling
+- test kopling
+
+
+square inv needs lego set category changed
+
+key lights
+pastel wrist things
+some sets

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"
+
+# Server mechanics
+daemon = False  # Run in foreground (systemd will handle daemonization)
+pidfile = "./logs/studspider.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")

doc/CERTIFICATE_FORMATS.md

@@ -0,0 +1,136 @@
+# Understanding PEM vs P7B Certificates
+
+## Quick Summary
+
+**You CANNOT use P7B files to run your HTTPS server!**
+
+- **PEM files** = Run your server with HTTPS ✓
+- **P7B files** = Upload to Square Dashboard ✓
+
+## The Complete Picture
+
+```
+┌─────────────────────────────────────────────────────────────┐
+│                    Your HTTPS Server                        │
+│                                                             │
+│  Uses: cert.pem + key.pem (PEM format)                    │
+│  Command: python src/test_server.py                        │
+│           --ssl-cert cert.pem                              │
+│           --ssl-key key.pem                                │
+│           --hook-port 8443                                 │
+└─────────────────────────────────────────────────────────────┘
+                         │
+                         │ HTTPS traffic
+                         ▼
+┌─────────────────────────────────────────────────────────────┐
+│                  Square's Servers                           │
+│                                                             │
+│  They verify your server using: cert.p7b                   │
+│  (which YOU upload to Square Dashboard)                    │
+└─────────────────────────────────────────────────────────────┘
+```
+
+## Why Two Formats?
+
+### PEM Format (.pem files)
+- **Contains**: Certificate AND private key (separate files)
+- **Used by**: Your HTTPS server to encrypt traffic
+- **Files needed**: 
+  - `cert.pem` - Your public certificate
+  - `key.pem` - Your private key (NEVER share this!)
+
+### P7B Format (.p7b files)  
+- **Contains**: Certificate ONLY (no private key)
+- **Used by**: Square to verify webhook signatures
+- **File**: `cert.p7b` - Uploaded to Square Dashboard
+
+## Step-by-Step Workflow
+
+### 1. Generate Certificates (PEM format)
+```powershell
+python generate_cert.py
+```
+**Creates:**
+- `cert.pem` - Your certificate (public)
+- `key.pem` - Your private key (secret!)
+
+### 2. Start Your HTTPS Server (uses PEM)
+```powershell
+python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only
+```
+**Server runs at:**
+- `https://localhost:8443/hook`
+
+### 3. Convert Certificate to P7B for Square
+```powershell
+python convert_to_p7b.py
+```
+**Creates:**
+- `cert.p7b` - Certificate in P7B format (for Square)
+
+### 4. Upload P7B to Square
+1. Go to: https://developer.squareup.com/apps
+2. Your App → **Webhooks** → **Manage Notification Settings**
+3. Upload `cert.p7b`
+
+### 5. Configure Square Webhook URL
+- Enter: `https://yourdomain.com:8443/hook`
+- Square will verify using the P7B you uploaded
+
+## Common Mistakes
+
+### ❌ Trying to use P7B with your server
+```powershell
+# THIS WON'T WORK!
+python src/test_server.py --ssl-cert cert.p7b --ssl-key key.p7b
+```
+**Error:** P7B doesn't contain the private key needed for HTTPS!
+
+### ✓ Correct usage
+```powershell
+# Server uses PEM files
+python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443
+
+# P7B file is uploaded to Square Dashboard (not used by your server)
+```
+
+## If You Only Have a P7B File
+
+If someone gave you a P7B file, you can extract the certificate:
+
+```powershell
+python convert_from_p7b.py your_cert.p7b --output cert.pem
+```
+
+**BUT:** You still need the private key (`key.pem`) to run HTTPS!
+
+If you don't have the private key, you must:
+1. Generate a new certificate pair: `python generate_cert.py`
+2. Or get the private key from whoever created the P7B
+
+## File Checklist
+
+To run your HTTPS server with Square webhooks, you need:
+
+- [ ] `cert.pem` - Certificate (PEM format) - For your server
+- [ ] `key.pem` - Private key (PEM format) - For your server  
+- [ ] `cert.p7b` - Certificate (P7B format) - Upload to Square
+
+Generate all three with:
+```powershell
+python generate_cert.py      # Creates cert.pem and key.pem
+python convert_to_p7b.py      # Creates cert.p7b from cert.pem
+```
+
+## Security Notes
+
+- **Never share `key.pem`** - This is your private key!
+- **`cert.pem` is public** - Safe to share
+- **`cert.p7b` is public** - You upload this to Square
+- Self-signed certificates are fine for testing but use Let's Encrypt for production
+
+## Need Help?
+
+See the detailed guides:
+- `HTTPS_SETUP.md` - Complete HTTPS setup instructions
+- `SQUARE_WEBHOOK_SETUP.md` - Square webhook configuration

doc/HTTPS_SETUP.md

@@ -0,0 +1,166 @@
+# HTTPS Webhook Setup Guide
+
+## Quick Start (Self-Signed Certificate for Testing)
+
+### 1. Generate Certificate
+```powershell
+python generate_cert.py
+```
+
+This creates `cert.pem` and `key.pem` files.
+
+### 1.5. Convert to P7B for Square (Required for Square Webhooks)
+```powershell
+python convert_to_p7b.py
+```
+
+This creates `cert.p7b` which Square needs for webhook signature verification.
+Upload this P7B file to Square Developer Dashboard → Webhooks → Manage Notification Settings.
+
+### 2. Run Server with HTTPS
+
+#### Option A: HTTPS for webhook only (recommended for development)
+```powershell
+python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only
+```
+- Main server: `http://localhost:7432/` (HTTP)
+- Webhook: `https://localhost:8443/hook` (HTTPS)
+
+#### Option B: HTTPS for everything
+```powershell
+python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443
+```
+- Main server: `https://localhost:7432/` (HTTPS)
+- Webhook: `https://localhost:8443/hook` (HTTPS)
+
+#### Option C: Single port with HTTPS
+```powershell
+python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem
+```
+- Everything on: `https://localhost:7432/`
+
+### 3. Test the Webhook
+
+Update `src/test_webhook.py` to use HTTPS:
+```python
+url = "https://localhost:8443/hook"
+```
+
+Then run:
+```powershell
+python src/test_webhook.py
+```
+
+## Production Setup (with Real Certificate)
+
+### Option 1: Let's Encrypt (Free, Recommended)
+
+On Linux/WSL:
+```bash
+# Install certbot
+sudo apt-get install certbot
+
+# Generate certificate (replace yourdomain.com)
+sudo certbot certonly --standalone -d yourdomain.com
+
+# Certificates will be at:
+# /etc/letsencrypt/live/yourdomain.com/fullchain.pem
+# /etc/letsencrypt/live/yourdomain.com/privkey.pem
+
+# Run server
+python src/test_server.py \
+  --ssl-cert /etc/letsencrypt/live/yourdomain.com/fullchain.pem \
+  --ssl-key /etc/letsencrypt/live/yourdomain.com/privkey.pem \
+  --host 0.0.0.0 \
+  --hook-port 443 \
+  --hook-ssl-only
+```
+
+### Option 2: Reverse Proxy (nginx/Caddy)
+
+Run studspider on HTTP internally, use nginx/Caddy for HTTPS:
+
+```nginx
+# nginx config
+server {
+    listen 443 ssl;
+    server_name yourdomain.com;
+    
+    ssl_certificate /path/to/cert.pem;
+    ssl_certificate_key /path/to/key.pem;
+    
+    location /hook {
+        proxy_pass http://localhost:7432/hook;
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+    }
+}
+```
+
+Then run studspider normally:
+```powershell
+python src/test_server.py
+```
+
+## Troubleshooting
+
+### "OpenSSL not found" Error
+- **Windows**: Install from https://slproweb.com/products/Win32OpenSSL.html
+- **Alternative**: Use Git Bash or WSL which include OpenSSL
+
+### Self-Signed Certificate Warnings
+- Normal for development
+- Use `--insecure` flag with curl/tools for testing
+- For production, get a proper certificate from Let's Encrypt
+
+### Port 443 Already in Use
+- Use a different port like 8443
+- Or stop the service using port 443
+
+### Certificate File Not Found
+- Use absolute paths: `--ssl-cert C:\path\to\cert.pem`
+- Or run from the directory containing the certificates
+
+## Square Webhook Configuration
+
+### Certificate Requirements
+Square requires a **P7B (PKCS#7)** certificate file for webhook signature verification.
+
+### Setup Steps:
+
+1. **Generate your certificate** (if you haven't already):
+   ```powershell
+   python generate_cert.py
+   ```
+
+2. **Convert PEM to P7B format**:
+   ```powershell
+   # Basic conversion
+   python convert_to_p7b.py
+   
+   # With intermediate certificates (production)
+   python convert_to_p7b.py --cert fullchain.pem --chain intermediate.pem --output square.p7b
+   ```
+
+3. **Upload to Square Dashboard**:
+   - Go to https://developer.squareup.com/
+   - Navigate to your application
+   - Click on "Webhooks"
+   - Click "Manage Notification Settings"
+   - Upload the `cert.p7b` file
+   - Square will use this to verify webhook signatures
+
+4. **Configure webhook URL in Square**:
+   - Use your public HTTPS URL: `https://yourdomain.com:8443/hook`
+   - Square requires valid SSL (not self-signed for production)
+
+### Testing with Square Webhooks:
+
+When setting up webhooks in Square Dashboard:
+1. Use your HTTPS URL: `https://yourdomain.com:8443/hook`
+2. Square requires valid SSL (not self-signed)
+3. Consider using ngrok for local testing:
+   ```powershell
+   ngrok http 8443
+   ```
+   Then use the ngrok HTTPS URL in Square Dashboard

doc/NGINX_SETUP.md

@@ -0,0 +1,356 @@
+# Nginx Reverse Proxy Configuration for studspider
+
+## Overview
+
+When using nginx as a reverse proxy:
+- **Nginx handles HTTPS** (SSL/TLS termination)
+- **studspider runs on HTTP** internally (no SSL needed)
+- **nginx forwards requests** to studspider
+
+## Architecture
+
+```
+Internet (HTTPS) → nginx (443) → studspider (7432/HTTP)
+                      ↓
+            Handles SSL certificates
+            Handles HTTPS traffic
+            Forwards to HTTP backend
+```
+
+## Benefits
+
+✓ **No SSL certificates needed in studspider**
+✓ **nginx handles SSL efficiently**
+✓ **Easy certificate renewal** (Let's Encrypt)
+✓ **Better performance** (nginx is optimized for this)
+✓ **Additional security** (rate limiting, IP filtering, etc.)
+
+---
+
+## Configuration Examples
+
+### Basic Configuration (HTTP → HTTPS redirect + webhook proxy)
+
+Create `/etc/nginx/sites-available/studspider`:
+
+```nginx
+# Redirect HTTP to HTTPS
+server {
+    listen 80;
+    server_name studspider.sh www.studspider.sh;
+    
+    location / {
+        return 301 https://$host$request_uri;
+    }
+}
+
+# HTTPS server
+server {
+    listen 443 ssl http2;
+    server_name studspider.sh www.studspider.sh;
+
+    # SSL certificates (Let's Encrypt)
+    ssl_certificate /etc/letsencrypt/live/studspider.sh/fullchain.pem;
+    ssl_certificate_key /etc/letsencrypt/live/studspider.sh/privkey.pem;
+    
+    # SSL configuration (modern, secure)
+    ssl_protocols TLSv1.2 TLSv1.3;
+    ssl_ciphers HIGH:!aNULL:!MD5;
+    ssl_prefer_server_ciphers on;
+
+    # Logs
+    access_log /var/log/nginx/studspider_access.log;
+    error_log /var/log/nginx/studspider_error.log;
+
+    # Webhook endpoint (Square webhooks)
+    location /hook {
+        proxy_pass http://localhost:7432/hook;
+        
+        # Forward headers
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto $scheme;
+        
+        # Timeouts
+        proxy_connect_timeout 60s;
+        proxy_send_timeout 60s;
+        proxy_read_timeout 60s;
+        
+        # Disable buffering for real-time responses
+        proxy_buffering off;
+    }
+
+    # Main API endpoint (optional)
+    location / {
+        proxy_pass http://localhost:7432/;
+        
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto $scheme;
+    }
+}
+```
+
+### Separate Port Configuration (webhook on different port)
+
+```nginx
+server {
+    listen 443 ssl http2;
+    server_name studspider.sh;
+
+    ssl_certificate /etc/letsencrypt/live/studspider.sh/fullchain.pem;
+    ssl_certificate_key /etc/letsencrypt/live/studspider.sh/privkey.pem;
+
+    # Main API
+    location / {
+        proxy_pass http://localhost:7432/;
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto https;
+    }
+}
+
+# Separate server for webhooks on different port
+server {
+    listen 8443 ssl http2;
+    server_name studspider.sh;
+
+    ssl_certificate /etc/letsencrypt/live/studspider.sh/fullchain.pem;
+    ssl_certificate_key /etc/letsencrypt/live/studspider.sh/privkey.pem;
+
+    # Webhook only
+    location /hook {
+        proxy_pass http://localhost:8080/hook;  # studspider webhook port
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto https;
+    }
+}
+```
+
+### With Rate Limiting (recommended for webhooks)
+
+```nginx
+# Define rate limit zone (outside server block)
+limit_req_zone $binary_remote_addr zone=webhook_limit:10m rate=10r/s;
+
+server {
+    listen 443 ssl http2;
+    server_name studspider.sh;
+
+    ssl_certificate /etc/letsencrypt/live/studspider.sh/fullchain.pem;
+    ssl_certificate_key /etc/letsencrypt/live/studspider.sh/privkey.pem;
+
+    # Webhook with rate limiting
+    location /hook {
+        # Rate limit: 10 requests per second per IP
+        limit_req zone=webhook_limit burst=20 nodelay;
+        
+        # Only allow POST requests
+        limit_except POST {
+            deny all;
+        }
+        
+        proxy_pass http://localhost:7432/hook;
+        proxy_set_header Host $host;
+        proxy_set_header X-Real-IP $remote_addr;
+        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
+        proxy_set_header X-Forwarded-Proto https;
+    }
+}
+```
+
+---
+
+## Setup Steps
+
+### 1. Install nginx (if not already installed)
+
+```bash
+# Ubuntu/Debian
+sudo apt update
+sudo apt install nginx
+
+# CentOS/RHEL
+sudo yum install nginx
+```
+
+### 2. Get SSL Certificate (Let's Encrypt - Free!)
+
+```bash
+# Install certbot
+sudo apt install certbot python3-certbot-nginx
+
+# Get certificate for your domain
+sudo certbot --nginx -d studspider.sh -d www.studspider.sh
+
+# Certbot will automatically:
+# 1. Generate certificate
+# 2. Configure nginx
+# 3. Set up auto-renewal
+```
+
+### 3. Create nginx Configuration
+
+```bash
+# Create config file
+sudo nano /etc/nginx/sites-available/studspider
+
+# Paste one of the configurations above
+
+# Create symbolic link to enable site
+sudo ln -s /etc/nginx/sites-available/studspider /etc/nginx/sites-enabled/
+
+# Test configuration
+sudo nginx -t
+
+# Reload nginx
+sudo systemctl reload nginx
+```
+
+### 4. Start studspider (HTTP only, no SSL needed!)
+
+```bash
+# Simple - single port
+python src/test_server.py --port 7432
+
+# Or with separate webhook port
+python src/test_server.py --port 7432 --hook-port 8080
+```
+
+### 5. Configure Square Webhook URL
+
+In Square Developer Dashboard:
+- **Webhook URL**: `https://studspider.sh/hook`
+- **Certificate**: Upload the P7B file (see below)
+
+### 6. Generate P7B for Square
+
+```bash
+# Convert your Let's Encrypt certificate to P7B for Square
+python convert_to_p7b.py --cert /etc/letsencrypt/live/studspider.sh/fullchain.pem --output square.p7b
+
+# Upload square.p7b to Square Dashboard
+```
+
+---
+
+## Systemd Service (Run studspider at boot)
+
+Create `/etc/systemd/system/studspider.service`:
+
+```ini
+[Unit]
+Description=studspider - LEGO Inventory Sync Service
+After=network.target
+
+[Service]
+Type=simple
+User=www-data
+WorkingDirectory=/opt/studspider
+Environment="PATH=/opt/studspider/.venv/bin"
+ExecStart=/opt/studspider/.venv/bin/python src/test_server.py --port 7432
+Restart=always
+RestartSec=10
+
+# Logging
+StandardOutput=append:/var/log/studspider/stdout.log
+StandardError=append:/var/log/studspider/stderr.log
+
+[Install]
+WantedBy=multi-user.target
+```
+
+Enable and start:
+
+```bash
+# Create log directory
+sudo mkdir -p /var/log/studspider
+sudo chown www-data:www-data /var/log/studspider
+
+# Enable service
+sudo systemctl enable studspider
+
+# Start service
+sudo systemctl start studspider
+
+# Check status
+sudo systemctl status studspider
+
+# View logs
+sudo journalctl -u studspider -f
+```
+
+---
+
+## Testing
+
+### Test nginx configuration
+```bash
+sudo nginx -t
+```
+
+### Test studspider locally
+```bash
+curl http://localhost:7432/hook -X POST -H "Content-Type: application/json" -d '{"test":"data"}'
+```
+
+### Test through nginx
+```bash
+curl https://studspider.sh/hook -X POST -H "Content-Type: application/json" -d '{"test":"data"}'
+```
+
+### Monitor logs
+```bash
+# nginx logs
+sudo tail -f /var/log/nginx/studspider_access.log
+sudo tail -f /var/log/nginx/studspider_error.log
+
+# studspider logs
+tail -f ./logs/studspider.log
+```
+
+---
+
+## Troubleshooting
+
+### 502 Bad Gateway
+- Check if studspider is running: `ps aux | grep test_server.py`
+- Check the port matches nginx config
+- Check firewall isn't blocking internal connections
+
+### Certificate Errors
+- Ensure Let's Encrypt renewal is working: `sudo certbot renew --dry-run`
+- Check certificate files exist and have correct permissions
+
+### Webhook Not Receiving
+- Verify nginx is forwarding: Check nginx access logs
+- Verify studspider is receiving: Check studspider logs
+- Test locally first, then through nginx
+
+### Connection Refused
+- Check studspider is listening: `sudo netstat -tlnp | grep 7432`
+- Check nginx proxy_pass points to correct host:port
+
+---
+
+## Summary
+
+### With nginx reverse proxy:
+
+✓ **studspider runs simple HTTP** (no SSL certificates needed)
+```bash
+python src/test_server.py --port 7432
+```
+
+✓ **nginx handles HTTPS** (manages certificates automatically)
+
+✓ **Square gets HTTPS webhook** (`https://studspider.sh/hook`)
+
+✓ **You upload P7B to Square** (converted from nginx's certificate)
+
+### This is the recommended production setup!

doc/SQUARE_WEBHOOK_SETUP.md

@@ -0,0 +1,191 @@
+# Square Webhook Setup - Quick Reference
+
+## Complete Setup Process
+
+### Step 1: Generate SSL Certificate
+```powershell
+# Generate PEM certificate
+python generate_cert.py
+# Creates: cert.pem, key.pem
+```
+
+### Step 2: Convert to P7B for Square
+```powershell
+# Convert PEM to P7B (PKCS#7) format
+python convert_to_p7b.py
+# Creates: cert.p7b
+```
+
+### Step 3: Start HTTPS Server
+```powershell
+# Start webhook server with HTTPS
+python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only
+
+# Server URLs:
+# Main:    http://localhost:7432/
+# Webhook: https://localhost:8443/hook (or https://yourdomain.com:8443/hook)
+```
+
+### Step 4: Configure Square Webhook
+
+1. **Upload P7B Certificate to Square:**
+   - Go to: https://developer.squareup.com/apps
+   - Select your application
+   - Navigate to: **Webhooks** → **Manage Notification Settings**
+   - Upload `cert.p7b` file
+   - This allows Square to verify webhook signatures
+
+2. **Add Webhook Subscription:**
+   - Still in Webhooks section
+   - Click **"Add Subscription"**
+   - Enter URL: `https://yourdomain.com:8443/hook`
+   - Select events to subscribe to (e.g., `inventory.count.updated`)
+   - Click **Save**
+
+### Step 5: Test the Webhook
+
+```powershell
+# Test from command line
+python src/test_webhook.py
+
+# Or manually with curl
+curl -X POST https://localhost:8443/hook \
+  -H "Content-Type: application/json" \
+  -d '{"type":"inventory.count.updated","merchant_id":"test"}'
+```
+
+## Production Deployment
+
+### Using Let's Encrypt (Recommended)
+
+```bash
+# 1. Get Let's Encrypt certificate (Linux/WSL)
+sudo certbot certonly --standalone -d yourdomain.com
+
+# 2. Convert to P7B for Square
+python convert_to_p7b.py \
+  --cert /etc/letsencrypt/live/yourdomain.com/fullchain.pem \
+  --output square.p7b
+
+# 3. Upload square.p7b to Square Dashboard
+
+# 4. Run server
+python src/test_server.py \
+  --ssl-cert /etc/letsencrypt/live/yourdomain.com/fullchain.pem \
+  --ssl-key /etc/letsencrypt/live/yourdomain.com/privkey.pem \
+  --host 0.0.0.0 \
+  --hook-port 443 \
+  --hook-ssl-only
+```
+
+### Using ngrok (Easy Testing)
+
+```powershell
+# Terminal 1: Start your server (HTTP is fine)
+python src/test_server.py --hook-port 8080
+
+# Terminal 2: Expose with ngrok (provides HTTPS automatically)
+ngrok http 8080
+
+# Use the ngrok HTTPS URL in Square:
+# Example: https://abc123.ngrok.io/hook
+
+# For P7B certificate with ngrok:
+# ngrok provides its own certificate, so you don't need to upload one to Square
+```
+
+## Webhook Event Types
+
+Common Square webhook events you can subscribe to:
+
+- `inventory.count.updated` - Inventory quantity changes
+- `catalog.version.updated` - Catalog item updates
+- `payment.created` - New payment received
+- `payment.updated` - Payment status changes
+- `order.created` - New order created
+- `order.updated` - Order status changes
+
+## Troubleshooting
+
+### Certificate Upload Fails
+- Ensure file is in P7B format (not PEM)
+- File must be valid PKCS#7 structure
+- Check file isn't corrupted: `openssl pkcs7 -print_certs -in cert.p7b`
+
+### Webhook Not Receiving Data
+1. Check server is running and accessible from internet
+2. Verify URL in Square Dashboard is correct
+3. Check firewall allows incoming traffic on webhook port
+4. Review server logs: `./logs/studspider.log`
+
+### SSL Certificate Errors
+- Self-signed certificates won't work with Square in production
+- Use Let's Encrypt or commercial CA for production
+- Use ngrok for quick testing (provides valid HTTPS)
+
+### Testing Locally
+```powershell
+# Square can't reach localhost - use one of these:
+# 1. ngrok (easiest)
+ngrok http 8443
+
+# 2. Cloudflare Tunnel
+cloudflared tunnel --url http://localhost:8443
+
+# 3. Deploy to cloud (AWS, Azure, DigitalOcean, etc.)
+```
+
+## Verifying Webhook Signatures
+
+Square signs webhook requests. Your server should verify signatures:
+
+```python
+# TODO: Add to hook_handler in test_server.py
+import hmac
+import hashlib
+
+def verify_square_signature(payload: bytes, signature: str, webhook_signature_key: str) -> bool:
+    """Verify Square webhook signature"""
+    computed_signature = hmac.new(
+        webhook_signature_key.encode('utf-8'),
+        payload,
+        hashlib.sha256
+    ).digest()
+    
+    expected_signature = base64.b64decode(signature)
+    return hmac.compare_digest(computed_signature, expected_signature)
+```
+
+Get your webhook signature key from Square Dashboard → Webhooks → Signature Key.
+
+## Environment Variables
+
+Add to your `.env` file:
+
+```env
+# Square Webhook Configuration
+SQUARE_WEBHOOK_SIGNATURE_KEY=your_signature_key_from_square_dashboard
+
+# SSL Certificate Paths (production)
+SSL_CERT_PATH=/path/to/fullchain.pem
+SSL_KEY_PATH=/path/to/privkey.pem
+```
+
+## Commands Quick Reference
+
+```powershell
+# Generate certificate
+python generate_cert.py
+
+# Convert to P7B
+python convert_to_p7b.py
+
+# Start HTTPS server
+python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only
+
+# Test webhook
+python src/test_webhook.py
+
+# Start with ngrok
+ngrok http 8443
+```

edderkop.code-workspace

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

edderkop.py

@@ -1,14 +0,0 @@
-import src.test_server
-
-
-class edderkop:
-    def __init__(self):
-        self.name = "Edderkop"
-        self.version = "0.0.1"
-        self.description = """
-        Edderkop 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):
-        print(f"{self.name} v{self.version} is running.")

pyproject.toml

@@ -1,35 +1,37 @@
-[project]
-name = "edderkop"
-version = "0.1.0"
-description = ""
-authors = [
-    {name = "CF", email = "chris-forst@pm.me"}
-]
-license = {text = "MIT"}
-readme = "README.md"
-requires-python = ">=3.13,<3.14"
-dependencies = [
-    "aiohttp (>=3.12.13,<4.0.0)",
-    "dotenv (>=0.9.9,<0.10.0)",
-    "pyinstaller (>=6.14.1,<7.0.0)",
-    "squareup(>=43.0.0.20250618,<44.0.0.0)",
-    "asyncio-throttle (>=1.0.2,<2.0.0)",
-    "requests (>=2.32.4,<3.0.0)",
-    "pytest (>=8.4.2,<9.0.0)",
-    "apscheduler (>=3.11.0,<4.0.0)",
-    "aiosqlite (>=0.21.0,<0.22.0)",
-]
-
-
-[build-system]
-requires = ["poetry-core>=2.0.0,<3.0.0"]
-build-backend = "poetry.core.masonry.api"
-
-[tool.poetry]
-package-mode = false
-
-[tool.bricksetapi]
-version = "0.1.0"
-
-[tool.abricksetapi]
-version = "0.1.0"
+[project]
+name = "studpider"
+version = "0.1.0"
+description = ""
+authors = [{ name = "CF", email = "ph0sgene@pm.me" }]
+requires-python = ">=3.13,<3.14"
+readme = "README.md"
+license = { text = "MIT" }
+dependencies = [
+    "aiohttp (>=3.12.13,<4.0.0)",
+    "dotenv (>=0.9.9,<0.10.0)",
+    "pyinstaller (>=6.14.1,<7.0.0)",
+    "squareup(>=43.0.0.20250618,<44.0.0.0)",
+    "asyncio-throttle (>=1.0.2,<2.0.0)",
+    "requests (>=2.32.4,<3.0.0)",
+    "pytest (>=8.4.2,<9.0.0)",
+    "apscheduler (>=3.11.0,<4.0.0)",
+    "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]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.uv]
+package = false
+
+[tool.bricksetapi]
+version = "0.1.0"
+
+[tool.abricksetapi]
+version = "0.1.0"

setup.py

@@ -1,19 +1,19 @@
 from setuptools import setup
 
 setup(
-    name='Edderkop',
-    version='0.1',
-    description='Bricket + Square integration application',
-    author='CF',
-    author_email='chris-forst@pm.me',
-    packages=['src'],
-    package_dir={'': 'src'},
+    name="studspider",
+    version="0.1",
+    description="Bricket + Square integration application",
+    author="CF",
+    author_email="chris-forst@pm.me",
+    packages=["src"],
+    package_dir={"": "src"},
     install_requires=[
         # List your project dependencies here
     ],
     entry_points={
-        'console_scripts': [
-            'main=main:main',  # Adjust this line based on your main function
+        "console_scripts": [
+            "main=main:main",  # Adjust this line based on your main function
         ],
     },
-)
+)

src/abricksetapi.py

@@ -1,137 +1,162 @@
-"""
-abricksetapi.py - Asynchronous Brickset API Client
-This module provides asynchronous interactions with the Brickset API.
-"""
-
-import asyncio
-import os
-import aiohttp
-import json
-import dotenv
-from typing import Dict, Any
-from asyncio_throttle.throttler import Throttler
-
-dotenv.load_dotenv()
-
-
-class AsyncBricksetAPI:
-    """Asynchronous client for the Brickset API"""
-
-    def __init__(self, debug: bool = False):
-        self.username = os.getenv("BRICKSET_USERNAME")
-        self.password = os.getenv("BRICKSET_PASS")
-        self.api_key = os.getenv("BRICKSET_API_KEY")
-        self.debug = debug
-        self.headers = {"UserAgent": "Edderkop/0.1"}
-        self.throttler = Throttler(rate_limit=500, period=60)
-        self.user_hash = None
-
-    async def login(self) -> Dict[str, Any]:
-        """Asynchronously authenticate with Brickset API"""
-        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  # type: ignore
-            ) as response:
-                if self.debug:
-                    print(f"[@DEBUG] Login response status: {response.status}")
-
-                data = await response.read()
-                result = json.loads(data.decode("utf-8"))
-                self.user_hash = result.get("hash")
-                return result
-
-    async def get_set(self, set_number: str) -> Dict[str, Any]:
-        """
-        Fetch LEGO set information asynchronously
-
-        Args:
-            set_number (str): LEGO set number
-
-        Returns:
-            Dict containing set information
-        """
-        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.debug:
-                    print(f"[@DEBUG] 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"):
-                        raise ValueError(f"No data found for set {set_number}")
-
-                    set_data = result["sets"][0]
-                    our_set = {
-                        "set_num": set_data["number"],
-                        "set_name": set_data["name"],
-                        "set_theme": set_data["theme"],
-                        "set_image": set_data["image"]["thumbnailURL"],
-                        "set_barcode": set_data["barcode"]["UPC"],
-                        "set_retailprice": round(
-                            set_data["LEGOCom"]["US"]["retailPrice"]
-                        ),
-                    }
-
-                    # Calculate adjusted price (10% markup)
-                    retail_price = our_set["set_retailprice"]
-                    our_set["set_adjustredprice"] = retail_price + (
-                        (10 / 100) * retail_price
-                    )
-
-                    # Set PW inventory name with theme
-                    our_set["pw_name"] = (
-                        f"{our_set['set_num']} - {our_set['set_theme'].upper()} {our_set['set_name']}"
-                    )
-
-                    # 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:
-                    raise Exception(
-                        "Internal Server Error: Please check API key and parameters."
-                    )
-                else:
-                    raise Exception(f"Unexpected response: {response.status}")
-
-    async def throttled_get_set(self, set_number: str) -> Dict[str, Any]:
-        """Get set information with rate limiting"""
-        async with self.throttler:
-            return await self.get_set(set_number)
-
-
-async def main():
-    """Example usage"""
-    client = AsyncBricksetAPI(debug=True)
-    try:
-        set_info = await client.throttled_get_set("10349")
-        print(json.dumps(set_info, indent=2))
-    except Exception as e:
-        print(f"Error: {e}")
-
-
-if __name__ == "__main__":
-    asyncio.run(main())
+"""
+abricksetapi.py - Asynchronous Brickset API Client
+This module provides asynchronous interactions with the Brickset API.
+"""
+
+import asyncio
+import os
+import aiohttp
+import json
+import dotenv
+from typing import Dict, Any
+import throttler
+
+Throttler = throttler.GlobalThrottler
+
+dotenv.load_dotenv()
+
+
+class AsyncBricksetAPI:
+    """Asynchronous client for the Brickset API"""
+
+    def __init__(self, debug: bool = False, logger=None):
+        self.username = os.getenv("BRICKSET_USERNAME")
+        self.password = os.getenv("BRICKSET_PASS")
+        self.api_key = os.getenv("BRICKSET_API_KEY")
+        self.debug = debug
+        self.headers = {"UserAgent": "studspider/0.1"}
+        self.throttler = Throttler(rate_limit=500, period=60)
+        self.user_hash = None
+        self.logger = logger
+
+    async def login(self) -> Dict[str, Any]:
+        """Asynchronously authenticate with Brickset API"""
+        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  # type: ignore
+            ) as response:
+                if self.debug:
+                    print(f"[@DEBUG] Login response status: {response.status}")
+                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_set(self, set_number: int) -> Dict[str, Any]:
+        """
+        Fetch LEGO set information asynchronously
+
+        Args:
+            set_number (str): LEGO set number
+
+        Returns:
+            Dict containing set information
+        """
+        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"}),
+        }
+        print(f"[@DEBUG] {params}...")
+
+        async with aiohttp.ClientSession() as session:
+            async with session.get(
+                url, params=params, headers=self.headers
+            ) as response:
+                if self.debug:
+                    print(f"[@DEBUG] Get set response status: {response.status}")
+                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]
+                    our_set = {
+                        "set_num": set_data["number"],
+                        "set_name": set_data["name"],
+                        "set_theme": set_data["theme"],
+                        "set_image": set_data["image"]["thumbnailURL"],
+                        "set_barcode": set_data["barcode"]["UPC"],
+                        "set_retailprice": round(
+                            set_data["LEGOCom"]["US"]["retailPrice"]
+                        ),
+                    }
+
+                    # Calculate adjusted price (10% markup)
+                    retail_price = our_set["set_retailprice"]
+                    our_set["set_adjustredprice"] = retail_price + (
+                        (10 / 100) * retail_price
+                    )
+
+                    # Set PW inventory name with theme
+                    our_set["pw_name"] = (
+                        f"{our_set['set_num']} - {our_set['set_theme'].upper()} {our_set['set_name']}"
+                    )
+
+                    # 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: str) -> Dict[str, Any]:
+        """Get set information with rate limiting"""
+        async with self.throttler:
+            return await self.get_set(set_number)
+
+
+async def main():
+    """Example usage"""
+    client = AsyncBricksetAPI(debug=True)
+    try:
+        set_info = await client.throttled_get_set("10349")
+        print(json.dumps(set_info, indent=2))
+    except Exception as e:
+        print(f"Error: {e}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

src/asquare.py

@@ -4,6 +4,9 @@ import json
 import asyncio
 import uuid
 import re
+from time import time
+
+import db
 
 import get_sq_env
 
@@ -52,10 +55,14 @@ async def update_itm(lego_set: dict):
     """
     Asynchronously updates an item in the Square catalog based on the provided LEGO set information.
     Args:
-        lego_set (dict): A dictionary containing LEGO set details. Expected keys include:
-            - "sets": List of set dictionaries, each with "name", "number", "barcode" (with "UPC"), and "image" (with "thumbnailURL").
-            - "pw_name": The name associated with the LEGO set in another system.
-            - "set_adjustredprice": The adjusted price for the set.
+        lego_set (dict): A dictionary containing LEGO set details from abricksetapi. Expected keys include:
+            - "set_num": LEGO set number (e.g., "75192")
+            - "set_name": Official LEGO set name
+            - "set_theme": LEGO theme (e.g., "Star Wars")
+            - "set_barcode": UPC barcode
+            - "set_image": Thumbnail URL
+            - "pw_name": Formatted name for Square inventory (e.g., "75192 - STAR WARS UCS Millennium Falcon")
+            - "set_adjustredprice": Retail price with 10% markup applied
     Returns:
         bool: True if the update operation was successful.
     Raises:
@@ -63,10 +70,10 @@ async def update_itm(lego_set: dict):
         Exception: For errors during Square API interaction.
     """
 
-    set_name = lego_set["sets"][0]["name"]
-    set_number = lego_set["sets"][0]["number"]
-    set_upc = lego_set["sets"][0]["barcode"]["UPC"]
-    set_image = lego_set["sets"][0]["image"]["thumbnailURL"]
+    set_number = lego_set["set_num"]
+    set_name = lego_set["set_name"]
+    set_upc = lego_set["set_barcode"]
+    set_image = lego_set["set_image"]
     pw_name = lego_set["pw_name"]
     set_adjustedprice = lego_set["set_adjustredprice"]
 
@@ -103,7 +110,17 @@ async def acat2id(cat_id: str):
         {
             "id": obj["id"],
             "name": obj.get("item_data", {}).get("name", "Unknown"),
-            "number": get_set_number(obj.get("item_data", {}).get("name", ""))
+            "number": (
+                None
+                if not obj.get("item_data", {}).get("name", "")
+                else (
+                    get_set_number(obj.get("item_data", {}).get("name", ""))
+                    if re.findall(r"\d{5}", obj.get("item_data", {}).get("name", ""))
+                    else None
+                )
+            ),
+            "updated_at": obj.get("updated_at"),
+            "version": obj.get("version"),
         }
         for obj in result_json.get("objects", [])
     ]
@@ -112,14 +129,10 @@ async def acat2id(cat_id: str):
 
 
 async def test() -> None:
-    task = asyncio.create_task(acat2id(LEGO_CAT_ID))  # type: ignore
-    await asyncio.sleep(1)  # Give the task some time to complete
-    result = await task
+    result = await acat2id(LEGO_CAT_ID)  # type: ignore
     print("OBJECTS IN LEGO CATEGORY:")
     for item in result:
-        print(f"ID: {item['id']}, Name: {item['name']}")
-    # eturn
-    # pass
+        print(f"ID: {item['id']}, Name: {item['name']}, Updated: {item['updated_at']}")
 
 
 """

src/asquareclass.py

@@ -0,0 +1,218 @@
+import os
+import json
+import asyncio
+import re
+from typing import List, Dict, Any, Optional
+from square import AsyncSquare
+import get_sq_env
+
+
+class AsyncSquareClient:
+    """Asynchronous client for the Square API with focus on catalog operations"""
+
+    def __init__(
+        self,
+        token: Optional[str] = None,
+        environment: Optional[str] = None,
+        logger=None,
+    ):
+        """
+        Initialize the AsyncSquareClient.
+
+        Args:
+            token (str, optional): Square API token. If not provided, fetches from environment.
+            environment (str, optional): Square environment (sandbox/production). If not provided, fetches from environment.
+            logger: Optional logger object for logging operations.
+        """
+        self.token = token or get_sq_env.SQUARE_TOKEN
+        self.environment = environment or get_sq_env.environment
+        self.lego_cat_id = get_sq_env.LEGO_CAT_ID
+        self.client = AsyncSquare(token=self.token, environment=self.environment)  # type: ignore
+        self.logger = logger
+
+    @staticmethod
+    def get_set_number(text: str) -> int:
+        """
+        Extract a five-digit LEGO set number from a string.
+
+        Args:
+            text (str): The text to search for a set number.
+
+        Returns:
+            int: The extracted set number.
+
+        Raises:
+            ValueError: If no five-digit number is found.
+        """
+        numbers = re.findall(r"\d{5}", text)
+        if not numbers:
+            raise ValueError(f"No five-digit number found in string {text}")
+        return int(numbers[0])
+
+    async def get_item(self, sku: str) -> Dict[str, str]:
+        """
+        Asynchronously retrieves item details from Square Catalog using a SKU.
+
+        Args:
+            sku (str): The SKU or text filter to search for the item.
+
+        Returns:
+            dict: A dictionary containing the item's name and ID.
+
+        Raises:
+            IndexError: If no items are found matching the SKU.
+            Exception: If there is an error communicating with the Square API.
+        """
+        if self.logger:
+            self.logger.debug(f"Searching for item with SKU: {sku}")
+
+        result = await self.client.catalog.search_items(text_filter=sku, limit=1)
+        result_json = json.loads(result.model_dump_json())
+
+        try:
+            name = result_json["items"][0]["item_data"]["name"]
+            itm_id = result_json["items"][0]["id"]
+
+            if self.logger:
+                self.logger.info(f"Found item: {name} with ID: {itm_id}")
+
+            return {"name": name, "id": itm_id}
+        except (KeyError, IndexError) as e:
+            if self.logger:
+                self.logger.error(f"Error retrieving item with SKU {sku}: {str(e)}")
+            raise
+
+    async def update_item(self, lego_set: Dict[str, Any]) -> bool:
+        """
+        Asynchronously updates an item in the Square catalog based on provided LEGO set information.
+
+        Args:
+            lego_set (dict): A dictionary containing LEGO set details including:
+                - sets: List containing set information
+                - pw_name: PriceWise system name
+                - set_adjustredprice: Adjusted retail price
+
+        Returns:
+            bool: True if update was successful.
+
+        Raises:
+            KeyError: If required keys are missing from input dictionary.
+            Exception: For Square API communication errors.
+        """
+        if self.logger:
+            self.logger.debug(
+                f"Updating item with set info: {lego_set['sets'][0]['number']}"
+            )
+
+        try:
+            set_name = lego_set["sets"][0]["name"]
+            set_number = lego_set["sets"][0]["number"]
+            set_upc = lego_set["sets"][0]["barcode"]["UPC"]
+            set_image = lego_set["sets"][0]["image"]["thumbnailURL"]
+            pw_name = lego_set["pw_name"]
+            set_adjustedprice = lego_set["set_adjustredprice"]
+
+            result = await self.client.catalog.search_items(
+                text_filter=set_number, limit=1
+            )
+            itm_id = json.loads(result.model_dump_json())["items"][0]["id"]
+
+            if self.logger:
+                self.logger.info(f"Successfully updated item {set_number}")
+
+            # TODO: Implement actual update logic here
+            # await self.client.catalog.object.upsert()
+            return True
+
+        except Exception as e:
+            if self.logger:
+                self.logger.error(f"Error updating item: {str(e)}")
+            raise
+
+    async def get_category_items(
+        self, cat_id: Optional[str] = None
+    ) -> List[Dict[str, Any]]:
+        """
+        Return a list of items within the specified category ID.
+
+        Args:
+            cat_id (str, optional): The category ID to search for items.
+                                  If not provided, uses the default LEGO category ID.
+
+        Returns:
+            list[dict]: A list of dictionaries, each containing:
+                - 'id': The item's ID
+                - 'name': The item's name
+                - 'number': The LEGO set number
+        """
+        category_id = cat_id or self.lego_cat_id
+
+        if self.logger:
+            self.logger.debug(f"Fetching items for category: {category_id}")
+
+        try:
+            # Create a properly structured query for the Square API
+            search_request = {
+                "object_types": ["ITEM"],
+                "query": {
+                    "exact_query": {
+                        "attribute_name": "category_id",
+                        "attribute_value": category_id,
+                    }
+                },
+            }
+
+            result = await self.client.catalog.search(**search_request)
+            result_json = json.loads(result.model_dump_json())
+
+            items = [
+                {
+                    "id": obj["id"],
+                    "name": obj.get("item_data", {}).get("name", "Unknown"),
+                    "number": self.get_set_number(
+                        obj.get("item_data", {}).get("name", "")
+                    ),
+                }
+                for obj in result_json.get("objects", [])
+            ]
+
+            if self.logger:
+                self.logger.info(f"Found {len(items)} items in category {category_id}")
+
+            return items
+
+        except Exception as e:
+            if self.logger:
+                self.logger.error(f"Error fetching category items: {str(e)}")
+            raise
+
+
+# Provide acat2id as a module-level function for backward compatibility
+async def acat2id(cat_id: str) -> List[Dict[str, Any]]:
+    """
+    Backward compatibility function for getting category items.
+
+    Args:
+        cat_id (str): The category ID to search for items.
+
+    Returns:
+        list[dict]: A list of dictionaries containing item information.
+    """
+    client = AsyncSquareClient()
+    return await client.get_category_items(cat_id)
+
+
+async def test() -> None:
+    """Test function to demonstrate class usage"""
+    client = AsyncSquareClient()
+    print(f"@[DEBUG] ENV:\t{client.environment}")
+    print(f"@[DEBUG] CATID:\t{client.lego_cat_id}")
+
+    result = await client.get_category_items()
+    print("\nOBJECTS IN LEGO CATEGORY:")
+    for item in result:
+        print(f"ID: {item['id']}, Name: {item['name']}")
+
+
+if __name__ == "__main__":
+    asyncio.run(test())

src/bricksetapi.py

@@ -14,9 +14,9 @@ debug = False  # Set to True to enable debug output
 
 def login(username, password):
     url = "https://brickset.com/api/v3.asmx/login"
-    headers = {"UserAgent": "Edderkop/0.1"}
+    headers = {"UserAgent": "studspider/0.1"}
     params = {
-        "UserAgent": "Edderkop/0.1",
+        "UserAgent": "studspider/0.1",
         "username": username,
         "password": password,
         "apiKey": BRICKSET_API_KEY,
@@ -33,6 +33,11 @@ def get_set(set_number: int) -> dict:
     response = requests.get(url)
     if response.status_code == 200:
         data = response.json()
+        if not data["sets"] and data["message"] == "API limit exceeded":
+            raise Exception("API limit exceeded. Please try again later.")
+            return False
+
+        print(f"[@DEBUG]\t{data}")
         our_set = {
             "set_num": data["sets"][0]["number"],
             "set_name": data["sets"][0]["name"],

src/db.py

@@ -0,0 +1,61 @@
+import aiosqlite
+from time import time
+
+time = int(time())  # we dont need no stinkin floats
+
+# db = await aiosqlite.connect("studspider.db")  # type: ignore
+
+
+async def close(db):
+    await db.close()
+
+
+async def fetch(table, sqobj):
+    """
+    Asynchronously fetch a record from the REVIEW table by sq_object.
+
+    Args:
+        sqobj (str): The sq_object identifier to search for.
+    Returns:
+        dict or None: The fetched record as a dictionary, or None if not found.
+    """
+    async with aiosqlite.connect("studspider.db") as db:  # type: ignore
+        async with db.execute(
+            f"SELECT * FROM {table} WHERE sq_object = {sqobj}"
+        ) as cursor:
+            async for row in cursor:
+                return row
+
+    return None
+
+
+async def upsert(table: str, data: dict, unique_key: str) -> bool:
+    """
+    Asynchronously insert or update a record in the specified table.
+
+    Args:
+        table (str): The name of the table to upsert into.
+        data (dict): A dictionary of column-value pairs to insert or update.
+        unique_key (str): The column name that serves as the unique key for upsert.
+
+    Returns:
+        None
+    """
+    print(f"Upserting into {table}: {data}")
+
+    columns = ", ".join(data.keys())
+    placeholders = ", ".join("?" * len(data))
+    updates = ", ".join(
+        f"{col}=excluded.{col}" for col in data.keys() if col != unique_key
+    )
+
+    sql = f"""
+    INSERT INTO {table} ({columns})
+    VALUES ({placeholders})
+    ON CONFLICT({unique_key}) DO UPDATE SET
+    {updates};
+    """
+    async with aiosqlite.connect("studspider.db") as db:  # type: ignore
+        async with db.execute(sql, tuple(data.values())) as cursor:
+            await db.commit()
+    return True

src/get_sq_env.py

@@ -6,11 +6,17 @@ dotenv.load_dotenv()
 
 LEGO_CAT_ID = os.getenv("SQUARE_LEGO_CAT_ID")
 
+SQUARE_WEBHOOK_URL = os.getenv("SQUARE_WEBHOOK_URL")
+
 if os.getenv("SQUARE_ENV") == "production":
     SQUARE_APP_ID = os.getenv("PROD_SQUARE_APP_ID")
     SQUARE_TOKEN = os.getenv("PROD_SQUARE_TOKEN")
+    SQUARE_WEBHOOK_SIGNATURE_KEY = os.getenv("PROD_SQUARE_WEBHOOK_SIGNATURE_KEY")
+    SQUARE_WEBHOOK_SUB_ID = os.getenv("PROD_SQUARE_WEBHOOK_SUB_ID")
     environment = SquareEnvironment.PRODUCTION
 else:
     SQUARE_APP_ID = os.getenv("SANDBOX_SQUARE_APP_ID")
     SQUARE_TOKEN = os.getenv("SANDBOX_SQUARE_TOKEN")
-    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

src/glue.py

@@ -0,0 +1,41 @@
+import asyncio
+import json
+import re
+from . import get_sq_env
+import aiosqlite
+from time import time
+
+from . import asquare
+from . import db
+from . import abricksetapi
+from . import throttler 
+Throttler = throttler.GlobalThrottler
+
+
+LEGO_CAT_ID = get_sq_env.LEGO_CAT_ID
+
+# big glue boy
+async def do_review() -> bool:
+    result = await asquare.acat2id(LEGO_CAT_ID) # type: ignore
+    for item in result:
+        if not re.match(r".*\d{5}.*", item["name"]):
+            missing = []
+            print(f"{item} is missing its set number!")
+            #print(f"[@DEBUG] Upserting {item['id']} into REVIEW table")
+            item_sanitized = {
+                "sq_object": item["id"],
+                "set_name": item["name"],
+                "set_number": None,
+                "time_added": time
+            }
+            missing.append(item_sanitized)
+    
+    for item in missing:
+        print(item)
+        # sq_item = await db.fetch(item["id"])  
+        if item["id"] != sq_item["sq_object"]:
+            print(f"[@DEBUG] {item['id']} is missing)")
+            #await db.upsert("MISSING", item_sanitized, "sq_object")
+        #    #print(f"[@DEBUG] Upserted {item['id']} into REVIEW table")
+        #print(f"[@DEBUG] {item['id']} already in REVIEW table")
+    return True

src/old/ebricksetapi.py

@@ -1,116 +1,116 @@
-import os
-import requests
-import json
-import dotenv
-from typing import Dict, Any
-
-dotenv.load_dotenv()
-
-
-class BricksetAPI:
-    """Class for interacting with the Brickset API"""
-
-    def __init__(self, debug: bool = False):
-        self.username = os.getenv("BRICKSET_USERNAME")
-        self.password = os.getenv("BRICKSET_PASS")
-        self.api_key = os.getenv("BRICKSET_API_KEY")
-        self.debug = debug
-        self.headers = {"UserAgent": "Edderkop/0.1"}
-        self.user_hash = None
-
-    def login(self) -> Dict[str, Any]:
-        """Login to Brickset API and get user hash"""
-        url = "https://brickset.com/api/v3.asmx/login"
-        params = {
-            "username": self.username,
-            "password": self.password,
-            "apiKey": self.api_key,
-        }
-
-        response = requests.get(url, params=params, headers=self.headers)
-        if self.debug:
-            print(f"[@DEBUG] Login response: {response.status_code}")
-            print(f"[@DEBUG] Response content: {response.text}")
-
-        try:
-            data = response.json()
-            self.user_hash = data.get("hash")
-            if not self.user_hash:
-                raise ValueError("No hash received in login response")
-            return data
-        except json.JSONDecodeError as e:
-            print(f"[@ERROR] Failed to decode JSON response: {e}")
-            print(f"[@DEBUG] Raw response: {response.text}")
-            raise
-
-    def get_set(self, set_number: str) -> Dict[str, Any]:
-        """Get LEGO set information by set number"""
-        if not self.user_hash:
-            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"}),
-        }
-
-        response = requests.get(url, params=params, headers=self.headers)
-
-        if self.debug:
-            print(f"[@DEBUG] Get set response: {response.status_code}")
-            print(f"[@DEBUG] Response content: {response.text}")
-
-        if response.status_code == 200:
-            try:
-                data = response.json()
-                if not data.get("sets"):
-                    raise ValueError(f"No set data found for {set_number}")
-
-                set_data = data["sets"][0]
-                our_set = {
-                    "set_num": set_data["number"],
-                    "set_name": set_data["name"],
-                    "set_image": set_data["image"]["thumbnailURL"],
-                    "set_barcode": set_data["barcode"]["UPC"],
-                    "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
-                )
-                # Set Playwell inventory name
-                if our_set["theme"] == "Duplo":
-                    our_set["pw_name"] = (
-                        f"{our_set['set_num']} - DUPLO {our_set['set_name']}"
-                    )
-                else:
-                    our_set["pw_name"] = f"{our_set['set_num']} - {our_set['set_name']}"
-
-                return our_set
-
-            except json.JSONDecodeError as e:
-                # print(f"[@ERROR] Failed to decode JSON response: {e}")
-                # print(f"[@DEBUG] Raw response: {response.text}")
-                raise
-            except KeyError as e:
-                # print(f"[@ERROR] Missing expected field in response: {e}")
-                # print(f"[@DEBUG] Response data: {data}")
-                raise
-        elif response.status_code == 500:
-            raise Exception(
-                "Internal Server Error: Please check API key and parameters."
-            )
-        else:
-            raise Exception(f"Unexpected response: {response.status_code}")
-
-
-def example():
-    client = BricksetAPI(debug=True)
-    try:
-        set_info = client.get_set("10931")
-        print(set_info)
-    except Exception as e:
-        print(f"Error: {e}")
+import os
+import requests
+import json
+import dotenv
+from typing import Dict, Any
+
+dotenv.load_dotenv()
+
+
+class BricksetAPI:
+    """Class for interacting with the Brickset API"""
+
+    def __init__(self, debug: bool = False):
+        self.username = os.getenv("BRICKSET_USERNAME")
+        self.password = os.getenv("BRICKSET_PASS")
+        self.api_key = os.getenv("BRICKSET_API_KEY")
+        self.debug = debug
+        self.headers = {"UserAgent": "studspider/0.1"}
+        self.user_hash = None
+
+    def login(self) -> Dict[str, Any]:
+        """Login to Brickset API and get user hash"""
+        url = "https://brickset.com/api/v3.asmx/login"
+        params = {
+            "username": self.username,
+            "password": self.password,
+            "apiKey": self.api_key,
+        }
+
+        response = requests.get(url, params=params, headers=self.headers)
+        if self.debug:
+            print(f"[@DEBUG] Login response: {response.status_code}")
+            print(f"[@DEBUG] Response content: {response.text}")
+
+        try:
+            data = response.json()
+            self.user_hash = data.get("hash")
+            if not self.user_hash:
+                raise ValueError("No hash received in login response")
+            return data
+        except json.JSONDecodeError as e:
+            print(f"[@ERROR] Failed to decode JSON response: {e}")
+            print(f"[@DEBUG] Raw response: {response.text}")
+            raise
+
+    def get_set(self, set_number: str) -> Dict[str, Any]:
+        """Get LEGO set information by set number"""
+        if not self.user_hash:
+            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"}),
+        }
+
+        response = requests.get(url, params=params, headers=self.headers)
+
+        if self.debug:
+            print(f"[@DEBUG] Get set response: {response.status_code}")
+            print(f"[@DEBUG] Response content: {response.text}")
+
+        if response.status_code == 200:
+            try:
+                data = response.json()
+                if not data.get("sets"):
+                    raise ValueError(f"No set data found for {set_number}")
+
+                set_data = data["sets"][0]
+                our_set = {
+                    "set_num": set_data["number"],
+                    "set_name": set_data["name"],
+                    "set_image": set_data["image"]["thumbnailURL"],
+                    "set_barcode": set_data["barcode"]["UPC"],
+                    "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
+                )
+                # Set Playwell inventory name
+                if our_set["theme"] == "Duplo":
+                    our_set["pw_name"] = (
+                        f"{our_set['set_num']} - DUPLO {our_set['set_name']}"
+                    )
+                else:
+                    our_set["pw_name"] = f"{our_set['set_num']} - {our_set['set_name']}"
+
+                return our_set
+
+            except json.JSONDecodeError as e:
+                # print(f"[@ERROR] Failed to decode JSON response: {e}")
+                # print(f"[@DEBUG] Raw response: {response.text}")
+                raise
+            except KeyError as e:
+                # print(f"[@ERROR] Missing expected field in response: {e}")
+                # print(f"[@DEBUG] Response data: {data}")
+                raise
+        elif response.status_code == 500:
+            raise Exception(
+                "Internal Server Error: Please check API key and parameters."
+            )
+        else:
+            raise Exception(f"Unexpected response: {response.status_code}")
+
+
+def example():
+    client = BricksetAPI(debug=True)
+    try:
+        set_info = client.get_set("10931")
+        print(set_info)
+    except Exception as e:
+        print(f"Error: {e}")

src/old/server.py

@@ -1,113 +1,113 @@
-import asyncio
-import aiohttp.web
-import uuid
-import os
-from base64 import b64decode
-import dotenv
-
-dotenv.load_dotenv()
-SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
-
-from asyncio_throttle.throttler import Throttler
-
-import asquare as sq
-import abricksetapi as alego
-
-
-def valid_uuid(val):
-    try:
-        uuid.UUID(str(val))
-        return True
-    except ValueError:
-        return False
-
-
-async def fullsync():
-    """'
-    full_db = await sq.acat2id(SQUARE_LEGO_CAT_ID)
-    for l_set in full_db:
-        l_setinfo = await alego.throttled_get_set(l_set)
-        await sq.update_itm(l_set, l_setinfo)"""
-
-
-async def stop():
-    pass
-
-
-async def priceonly():
-    pass
-
-
-async def handle_post(request):
-
-    valid_cmd = ["RUNFULLSYNC", "STOP", "RUNPRICEONLY"]
-
-    async def handle_cmd(cmd: str):
-        if cmd not in valid_cmd:
-            return False
-        else:
-            return True
-
-    idp_key = request.headers.get("Idempotency-Key")
-    user_agent = request.headers.get("User-Agent")
-
-    if not idp_key and not valid_uuid(idp_key) and user_agent != "Kopling/0.1":
-        # 412 Precondition Failed
-        # 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!")
-
-    else:
-        data = await request.post()
-        reqdata = data.get(b64decode("CMD"))
-
-        if await handle_cmd(reqdata):
-            match reqdata:
-                case "RUNFULLSYNC":
-                    print("Running full sync...")
-                    await fullsync()
-                    return True
-                case "STOP":
-                    await stop()
-                    return True
-                case "RUNPRICEONLY":
-                    await priceonly()
-                    return True
-        else:
-            return aiohttp.web.Response(status=400, text="400 Bad Request.")
-        result = {"message": f"Edderkop/0.1 received command: {reqdata}"}
-        return aiohttp.web.json_response(status=200, data=result)
-
-
-app = aiohttp.web.Application()
-app.router.add_post("/", handle_post)
-
-import subprocess
-
-
-def generate_toilet_banner(text, font="smblock", filter_effect=None, colors=None):
-    command = ["toilet", "-f", font]
-    if filter_effect:
-        command.extend(["-F", filter_effect])
-    if colors:
-        # 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.
-        # This example focuses on filters that inherently use colors.
-        pass  # More complex color handling would be needed for arbitrary colors
-
-    command.append(text)
-
-    try:
-        result = subprocess.run(command, capture_output=True, text=True, check=True)
-        return result.stdout
-    except subprocess.CalledProcessError as e:
-        return f"Error executing toilet: {e}\n{e.stderr}"
-
-
-async def main():
-    throttler = Throttler(rate_limit=500, period=60)  # 500 requests per minute
-
-
-if __name__ == "__main__":
-    print(generate_toilet_banner("Edderkop v0.1"))
-    print("Built for PLAY-WELL TEKNOLOGIES by contact@traps.moe\n\n")
-    aiohttp.web.run_app(app, port=7432)
+import asyncio
+import aiohttp.web
+import uuid
+import os
+from base64 import b64decode
+import dotenv
+
+dotenv.load_dotenv()
+SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
+
+from asyncio_throttle.throttler import Throttler
+
+import asquare as sq
+import abricksetapi as alego
+
+
+def valid_uuid(val):
+    try:
+        uuid.UUID(str(val))
+        return True
+    except ValueError:
+        return False
+
+
+async def fullsync():
+    """'
+    full_db = await sq.acat2id(SQUARE_LEGO_CAT_ID)
+    for l_set in full_db:
+        l_setinfo = await alego.throttled_get_set(l_set)
+        await sq.update_itm(l_set, l_setinfo)"""
+
+
+async def stop():
+    pass
+
+
+async def priceonly():
+    pass
+
+
+async def handle_post(request):
+
+    valid_cmd = ["RUNFULLSYNC", "STOP", "RUNPRICEONLY"]
+
+    async def handle_cmd(cmd: str):
+        if cmd not in valid_cmd:
+            return False
+        else:
+            return True
+
+    idp_key = request.headers.get("Idempotency-Key")
+    user_agent = request.headers.get("User-Agent")
+
+    if not idp_key and not valid_uuid(idp_key) and user_agent != "Kopling/0.1":
+        # 412 Precondition Failed
+        # 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!")
+
+    else:
+        data = await request.post()
+        reqdata = data.get(b64decode("CMD"))
+
+        if await handle_cmd(reqdata):
+            match reqdata:
+                case "RUNFULLSYNC":
+                    print("Running full sync...")
+                    await fullsync()
+                    return True
+                case "STOP":
+                    await stop()
+                    return True
+                case "RUNPRICEONLY":
+                    await priceonly()
+                    return True
+        else:
+            return aiohttp.web.Response(status=400, text="400 Bad Request.")
+        result = {"message": f"studspider/0.1 received command: {reqdata}"}
+        return aiohttp.web.json_response(status=200, data=result)
+
+
+app = aiohttp.web.Application()
+app.router.add_post("/", handle_post)
+
+import subprocess
+
+
+def generate_toilet_banner(text, font="smblock", filter_effect=None, colors=None):
+    command = ["toilet", "-f", font]
+    if filter_effect:
+        command.extend(["-F", filter_effect])
+    if colors:
+        # 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.
+        # This example focuses on filters that inherently use colors.
+        pass  # More complex color handling would be needed for arbitrary colors
+
+    command.append(text)
+
+    try:
+        result = subprocess.run(command, capture_output=True, text=True, check=True)
+        return result.stdout
+    except subprocess.CalledProcessError as e:
+        return f"Error executing toilet: {e}\n{e.stderr}"
+
+
+async def main():
+    throttler = Throttler(rate_limit=500, period=60)  # 500 requests per minute
+
+
+if __name__ == "__main__":
+    print(generate_toilet_banner("studspider v0.1"))
+    print("Built for PLAY-WELL TEKNOLOGIES by contact@traps.moe\n\n")
+    aiohttp.web.run_app(app, port=7432)

src/scheduler.py

@@ -1,221 +1,221 @@
-import aiosqlite
-import logging
-from typing import Optional, List, Dict, Any
-import json
-import os
-
-logger = logging.getLogger(__name__)
-
-# Get the absolute path to edderkop.db in the project root
-DEFAULT_DB_PATH = os.path.join(
-    os.path.dirname(os.path.dirname(__file__)), "edderkop.db"
-)
-
-if not os.path.exists(DEFAULT_DB_PATH):
-    raise FileNotFoundError(f"Database file not found at {DEFAULT_DB_PATH}")
-    print('Please run "src/scripts/EDDERKOP.SQL" to setup the database.')
-    os.exit(0)
-
-
-class AsyncSchedulerDB:
-    """
-    Asynchronous SQLite database manager for APScheduler.
-    Handles job storage, retrieval, and database operations using the main edderkop.db database.
-    """
-
-    def __init__(self, db_path: str = DEFAULT_DB_PATH):
-        """
-        Initialize the database manager.
-
-        Args:
-            db_path (str, optional): Path to the SQLite database file.
-                                   Defaults to edderkop.db in project root.
-        """
-        self.db_path = db_path
-        self._initialized = False
-        logger.info(f"Using database at: {self.db_path}")
-
-    async def initialize(self):
-        """Initialize the database schema if it doesn't exist."""
-        if self._initialized:
-            return
-
-        async with aiosqlite.connect(self.db_path) as db:
-            # Create apscheduler jobs table if it doesn't exist
-            await db.execute(
-                """
-                CREATE TABLE IF NOT EXISTS apscheduler_jobs (
-                    id VARCHAR(191) NOT NULL,
-                    next_run_time FLOAT,
-                    job_state BLOB NOT NULL,
-                    PRIMARY KEY (id)
-                )
-            """
-            )
-            await db.commit()
-
-        self._initialized = True
-        logger.info(f"Initialized scheduler database at {self.db_path}")
-
-    async def get_all_jobs(self) -> List[Dict[str, Any]]:
-        """
-        Retrieve all jobs from the database.
-
-        Returns:
-            List[Dict[str, Any]]: List of job dictionaries
-        """
-        await self.initialize()
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
-            async with db.execute(
-                "SELECT * FROM apscheduler_jobs ORDER BY next_run_time"
-            ) as cursor:
-                rows = await cursor.fetchall()
-                return [
-                    {
-                        "id": row["id"],
-                        "next_run_time": row["next_run_time"],
-                        "job_state": json.loads(row["job_state"]),
-                    }
-                    for row in rows
-                ]
-
-    async def get_job(self, job_id: str) -> Optional[Dict[str, Any]]:
-        """
-        Retrieve a specific job by ID.
-
-        Args:
-            job_id (str): The ID of the job to retrieve
-
-        Returns:
-            Optional[Dict[str, Any]]: Job dictionary if found, None otherwise
-        """
-        await self.initialize()
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
-            async with db.execute(
-                "SELECT * FROM apscheduler_jobs WHERE id = ?", (job_id,)
-            ) as cursor:
-                row = await cursor.fetchone()
-                if row:
-                    return {
-                        "id": row["id"],
-                        "next_run_time": row["next_run_time"],
-                        "job_state": json.loads(row["job_state"]),
-                    }
-                return None
-
-    async def add_job(
-        self, job_id: str, next_run_time: float, job_state: Dict[str, Any]
-    ) -> bool:
-        """
-        Add a new job to the database.
-
-        Args:
-            job_id (str): Unique identifier for the job
-            next_run_time (float): Unix timestamp for next scheduled run
-            job_state (Dict[str, Any]): Job state dictionary
-
-        Returns:
-            bool: True if successful, False otherwise
-        """
-        await self.initialize()
-        try:
-            async with aiosqlite.connect(self.db_path) as db:
-                await db.execute(
-                    "INSERT INTO apscheduler_jobs (id, next_run_time, job_state) VALUES (?, ?, ?)",
-                    (job_id, next_run_time, json.dumps(job_state)),
-                )
-                await db.commit()
-                return True
-        except aiosqlite.IntegrityError:
-            logger.error(f"Job with ID {job_id} already exists")
-            return False
-        except Exception as e:
-            logger.error(f"Error adding job: {e}")
-            return False
-
-    async def update_job(
-        self, job_id: str, next_run_time: float, job_state: Dict[str, Any]
-    ) -> bool:
-        """
-        Update an existing job in the database.
-
-        Args:
-            job_id (str): ID of the job to update
-            next_run_time (float): New next run time
-            job_state (Dict[str, Any]): Updated job state
-
-        Returns:
-            bool: True if successful, False if job not found
-        """
-        await self.initialize()
-        async with aiosqlite.connect(self.db_path) as db:
-            cursor = await db.execute(
-                """UPDATE apscheduler_jobs 
-                   SET next_run_time = ?, job_state = ?
-                   WHERE id = ?""",
-                (next_run_time, json.dumps(job_state), job_id),
-            )
-            await db.commit()
-            return cursor.rowcount > 0
-
-    async def remove_job(self, job_id: str) -> bool:
-        """
-        Remove a job from the database.
-
-        Args:
-            job_id (str): ID of the job to remove
-
-        Returns:
-            bool: True if job was removed, False if not found
-        """
-        await self.initialize()
-        async with aiosqlite.connect(self.db_path) as db:
-            cursor = await db.execute(
-                "DELETE FROM apscheduler_jobs WHERE id = ?", (job_id,)
-            )
-            await db.commit()
-            return cursor.rowcount > 0
-
-    async def remove_all_jobs(self) -> int:
-        """
-        Remove all jobs from the database.
-
-        Returns:
-            int: Number of jobs removed
-        """
-        await self.initialize()
-        async with aiosqlite.connect(self.db_path) as db:
-            cursor = await db.execute("DELETE FROM apscheduler_jobs")
-            await db.commit()
-            return cursor.rowcount
-
-    async def get_due_jobs(self, now: float) -> List[Dict[str, Any]]:
-        """
-        Get all jobs that are due to run.
-
-        Args:
-            now (float): Current timestamp
-
-        Returns:
-            List[Dict[str, Any]]: List of due jobs
-        """
-        await self.initialize()
-        async with aiosqlite.connect(self.db_path) as db:
-            db.row_factory = aiosqlite.Row
-            async with db.execute(
-                """SELECT * FROM apscheduler_jobs 
-                   WHERE next_run_time <= ? 
-                   ORDER BY next_run_time""",
-                (now,),
-            ) as cursor:
-                rows = await cursor.fetchall()
-                return [
-                    {
-                        "id": row["id"],
-                        "next_run_time": row["next_run_time"],
-                        "job_state": json.loads(row["job_state"]),
-                    }
-                    for row in rows
-                ]
+import aiosqlite
+import logging
+from typing import Optional, List, Dict, Any
+import json
+import os
+
+logger = logging.getLogger(__name__)
+
+# Get the absolute path to studspider.db in the project root
+DEFAULT_DB_PATH = os.path.join(
+    os.path.dirname(os.path.dirname(__file__)), "studspider.db"
+)
+
+if not os.path.exists(DEFAULT_DB_PATH):
+    raise FileNotFoundError(f"Database file not found at {DEFAULT_DB_PATH}")
+    print('Please run "src/scripts/studspider.SQL" to setup the database.')
+    os.exit(0)
+
+
+class AsyncSchedulerDB:
+    """
+    Asynchronous SQLite database manager for APScheduler.
+    Handles job storage, retrieval, and database operations using the main studspider.db database.
+    """
+
+    def __init__(self, db_path: str = DEFAULT_DB_PATH):
+        """
+        Initialize the database manager.
+
+        Args:
+            db_path (str, optional): Path to the SQLite database file.
+                                   Defaults to studspider.db in project root.
+        """
+        self.db_path = db_path
+        self._initialized = False
+        logger.info(f"Using database at: {self.db_path}")
+
+    async def initialize(self):
+        """Initialize the database schema if it doesn't exist."""
+        if self._initialized:
+            return
+
+        async with aiosqlite.connect(self.db_path) as db:
+            # Create apscheduler jobs table if it doesn't exist
+            await db.execute(
+                """
+                CREATE TABLE IF NOT EXISTS apscheduler_jobs (
+                    id VARCHAR(191) NOT NULL,
+                    next_run_time FLOAT,
+                    job_state BLOB NOT NULL,
+                    PRIMARY KEY (id)
+                )
+            """
+            )
+            await db.commit()
+
+        self._initialized = True
+        logger.info(f"Initialized scheduler database at {self.db_path}")
+
+    async def get_all_jobs(self) -> List[Dict[str, Any]]:
+        """
+        Retrieve all jobs from the database.
+
+        Returns:
+            List[Dict[str, Any]]: List of job dictionaries
+        """
+        await self.initialize()
+        async with aiosqlite.connect(self.db_path) as db:
+            db.row_factory = aiosqlite.Row
+            async with db.execute(
+                "SELECT * FROM apscheduler_jobs ORDER BY next_run_time"
+            ) as cursor:
+                rows = await cursor.fetchall()
+                return [
+                    {
+                        "id": row["id"],
+                        "next_run_time": row["next_run_time"],
+                        "job_state": json.loads(row["job_state"]),
+                    }
+                    for row in rows
+                ]
+
+    async def get_job(self, job_id: str) -> Optional[Dict[str, Any]]:
+        """
+        Retrieve a specific job by ID.
+
+        Args:
+            job_id (str): The ID of the job to retrieve
+
+        Returns:
+            Optional[Dict[str, Any]]: Job dictionary if found, None otherwise
+        """
+        await self.initialize()
+        async with aiosqlite.connect(self.db_path) as db:
+            db.row_factory = aiosqlite.Row
+            async with db.execute(
+                "SELECT * FROM apscheduler_jobs WHERE id = ?", (job_id,)
+            ) as cursor:
+                row = await cursor.fetchone()
+                if row:
+                    return {
+                        "id": row["id"],
+                        "next_run_time": row["next_run_time"],
+                        "job_state": json.loads(row["job_state"]),
+                    }
+                return None
+
+    async def add_job(
+        self, job_id: str, next_run_time: float, job_state: Dict[str, Any]
+    ) -> bool:
+        """
+        Add a new job to the database.
+
+        Args:
+            job_id (str): Unique identifier for the job
+            next_run_time (float): Unix timestamp for next scheduled run
+            job_state (Dict[str, Any]): Job state dictionary
+
+        Returns:
+            bool: True if successful, False otherwise
+        """
+        await self.initialize()
+        try:
+            async with aiosqlite.connect(self.db_path) as db:
+                await db.execute(
+                    "INSERT INTO apscheduler_jobs (id, next_run_time, job_state) VALUES (?, ?, ?)",
+                    (job_id, next_run_time, json.dumps(job_state)),
+                )
+                await db.commit()
+                return True
+        except aiosqlite.IntegrityError:
+            logger.error(f"Job with ID {job_id} already exists")
+            return False
+        except Exception as e:
+            logger.error(f"Error adding job: {e}")
+            return False
+
+    async def update_job(
+        self, job_id: str, next_run_time: float, job_state: Dict[str, Any]
+    ) -> bool:
+        """
+        Update an existing job in the database.
+
+        Args:
+            job_id (str): ID of the job to update
+            next_run_time (float): New next run time
+            job_state (Dict[str, Any]): Updated job state
+
+        Returns:
+            bool: True if successful, False if job not found
+        """
+        await self.initialize()
+        async with aiosqlite.connect(self.db_path) as db:
+            cursor = await db.execute(
+                """UPDATE apscheduler_jobs 
+                   SET next_run_time = ?, job_state = ?
+                   WHERE id = ?""",
+                (next_run_time, json.dumps(job_state), job_id),
+            )
+            await db.commit()
+            return cursor.rowcount > 0
+
+    async def remove_job(self, job_id: str) -> bool:
+        """
+        Remove a job from the database.
+
+        Args:
+            job_id (str): ID of the job to remove
+
+        Returns:
+            bool: True if job was removed, False if not found
+        """
+        await self.initialize()
+        async with aiosqlite.connect(self.db_path) as db:
+            cursor = await db.execute(
+                "DELETE FROM apscheduler_jobs WHERE id = ?", (job_id,)
+            )
+            await db.commit()
+            return cursor.rowcount > 0
+
+    async def remove_all_jobs(self) -> int:
+        """
+        Remove all jobs from the database.
+
+        Returns:
+            int: Number of jobs removed
+        """
+        await self.initialize()
+        async with aiosqlite.connect(self.db_path) as db:
+            cursor = await db.execute("DELETE FROM apscheduler_jobs")
+            await db.commit()
+            return cursor.rowcount
+
+    async def get_due_jobs(self, now: float) -> List[Dict[str, Any]]:
+        """
+        Get all jobs that are due to run.
+
+        Args:
+            now (float): Current timestamp
+
+        Returns:
+            List[Dict[str, Any]]: List of due jobs
+        """
+        await self.initialize()
+        async with aiosqlite.connect(self.db_path) as db:
+            db.row_factory = aiosqlite.Row
+            async with db.execute(
+                """SELECT * FROM apscheduler_jobs 
+                   WHERE next_run_time <= ? 
+                   ORDER BY next_run_time""",
+                (now,),
+            ) as cursor:
+                rows = await cursor.fetchall()
+                return [
+                    {
+                        "id": row["id"],
+                        "next_run_time": row["next_run_time"],
+                        "job_state": json.loads(row["job_state"]),
+                    }
+                    for row in rows
+                ]

src/scripts/convert_from_p7b.py

@@ -0,0 +1,110 @@
+"""
+Convert P7B (PKCS#7) certificate to PEM format for use with the server
+Use this if you have a P7B file and need to extract PEM certificates
+"""
+
+import subprocess
+import sys
+import os
+
+
+def convert_p7b_to_pem(p7b_file, cert_output="cert.pem", extract_key=False):
+    """
+    Convert P7B certificate to PEM format
+
+    Args:
+        p7b_file: Input P7B (PKCS#7) file
+        cert_output: Output PEM certificate file
+        extract_key: If True, try to extract private key (rarely works with P7B)
+    """
+
+    # Check if OpenSSL is available
+    try:
+        subprocess.run(["openssl", "version"], capture_output=True, check=True)
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        print("ERROR: OpenSSL not found!")
+        print("\nFor Windows, you can:")
+        print(
+            "1. Install OpenSSL from: https://slproweb.com/products/Win32OpenSSL.html"
+        )
+        print("2. Or use Git Bash which includes OpenSSL")
+        print("3. Or use WSL (Windows Subsystem for Linux)")
+        return False
+
+    # Check if input file exists
+    if not os.path.exists(p7b_file):
+        print(f"ERROR: P7B file not found: {p7b_file}")
+        print(f"Current directory: {os.getcwd()}")
+        return False
+
+    print(f"Converting {p7b_file} to PEM format...")
+
+    # Convert P7B to PEM
+    cmd = ["openssl", "pkcs7", "-print_certs", "-in", p7b_file, "-out", cert_output]
+
+    try:
+        result = subprocess.run(cmd, check=True, capture_output=True, text=True)
+
+        if os.path.exists(cert_output):
+            print(f"\n✓ PEM certificate created: {os.path.abspath(cert_output)}")
+            print(f"  File size: {os.path.getsize(cert_output)} bytes")
+
+            # Check what was extracted
+            with open(cert_output, "r") as f:
+                content = f.read()
+                cert_count = content.count("BEGIN CERTIFICATE")
+                print(f"  Certificates found: {cert_count}")
+
+            print("\n" + "=" * 60)
+            print("IMPORTANT:")
+            print("=" * 60)
+            print("⚠️  P7B files usually contain ONLY certificates, not private keys!")
+            print("\nTo run an HTTPS server, you need:")
+            print("1. Certificate (cert.pem) - ✓ Extracted")
+            print("2. Private Key (key.pem) - ✗ NOT in P7B file")
+            print("\nIf you don't have the private key:")
+            print("- Generate a new certificate pair: python generate_cert.py")
+            print("- Or obtain the private key from your certificate provider")
+            print("=" * 60)
+
+            return True
+        else:
+            print("ERROR: Output file was not created")
+            return False
+
+    except subprocess.CalledProcessError as e:
+        print(f"ERROR: Conversion failed: {e}")
+        if e.stderr:
+            print(f"OpenSSL error: {e.stderr}")
+        return False
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser(
+        description="Convert P7B (PKCS#7) certificate to PEM format"
+    )
+    parser.add_argument("p7b_file", help="Input P7B certificate file")
+    parser.add_argument(
+        "--output",
+        default="cert.pem",
+        help="Output PEM certificate file (default: cert.pem)",
+    )
+
+    args = parser.parse_args()
+
+    print("=" * 60)
+    print("P7B to PEM Certificate Converter")
+    print("=" * 60)
+    print()
+
+    success = convert_p7b_to_pem(args.p7b_file, args.output)
+
+    if success:
+        print("\n✓ Conversion complete!")
+        print("\nNext steps:")
+        print("1. If you have the private key separately, use it with --ssl-key")
+        print("2. If not, generate a new certificate: python generate_cert.py")
+    else:
+        sys.exit(1)

src/scripts/convert_to_p7b.py

@@ -0,0 +1,166 @@
+"""
+Convert PEM certificate to P7B (PKCS#7) format for Square webhooks
+Square requires P7B format for webhook signature verification
+"""
+
+import subprocess
+import sys
+import os
+
+
+def convert_pem_to_p7b(cert_file="cert.pem", output_file="cert.p7b"):
+    """
+    Convert PEM certificate to P7B (PKCS#7) format
+
+    Args:
+        cert_file: Input PEM certificate file
+        output_file: Output P7B file
+    """
+
+    # Check if OpenSSL is available
+    try:
+        subprocess.run(["openssl", "version"], capture_output=True, check=True)
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        print("ERROR: OpenSSL not found!")
+        print("\nFor Windows, you can:")
+        print(
+            "1. Install OpenSSL from: https://slproweb.com/products/Win32OpenSSL.html"
+        )
+        print("2. Or use Git Bash which includes OpenSSL")
+        print("3. Or use WSL (Windows Subsystem for Linux)")
+        return False
+
+    # Check if input file exists
+    if not os.path.exists(cert_file):
+        print(f"ERROR: Certificate file not found: {cert_file}")
+        print(f"Current directory: {os.getcwd()}")
+        print("\nMake sure you've generated the certificate first:")
+        print("  python generate_cert.py")
+        return False
+
+    print(f"Converting {cert_file} to P7B format...")
+
+    # Convert PEM to P7B (PKCS#7)
+    # -certfile can include intermediate certificates if needed
+    cmd = [
+        "openssl",
+        "crl2pkcs7",
+        "-nocrl",
+        "-certfile",
+        cert_file,
+        "-out",
+        output_file,
+    ]
+
+    try:
+        result = subprocess.run(cmd, check=True, capture_output=True, text=True)
+
+        if os.path.exists(output_file):
+            print(f"\n✓ P7B certificate created: {os.path.abspath(output_file)}")
+            print(f"  File size: {os.path.getsize(output_file)} bytes")
+
+            print("\n" + "=" * 60)
+            print("Square Webhook Configuration:")
+            print("=" * 60)
+            print(f"1. Upload this file to Square Developer Dashboard:")
+            print(f"   {os.path.abspath(output_file)}")
+            print("\n2. In Square Dashboard:")
+            print("   - Go to Webhooks settings")
+            print("   - Click 'Manage Notification Settings'")
+            print("   - Upload the P7B file for signature verification")
+            print("\n3. Square will use this to verify webhook signatures")
+            print("=" * 60)
+
+            # Show file contents preview
+            print("\nP7B file preview (first 500 chars):")
+            with open(output_file, "r") as f:
+                content = f.read(500)
+                print(content)
+                if len(content) == 500:
+                    print("...")
+
+            return True
+        else:
+            print("ERROR: Output file was not created")
+            return False
+
+    except subprocess.CalledProcessError as e:
+        print(f"ERROR: Conversion failed: {e}")
+        if e.stderr:
+            print(f"OpenSSL error: {e.stderr}")
+        return False
+
+
+def convert_with_chain(cert_file="cert.pem", chain_file=None, output_file="cert.p7b"):
+    """
+    Convert PEM certificate with certificate chain to P7B format
+    Use this if you have intermediate certificates
+
+    Args:
+        cert_file: Your certificate file
+        chain_file: Intermediate/chain certificate file (optional)
+        output_file: Output P7B file
+    """
+
+    if not chain_file:
+        return convert_pem_to_p7b(cert_file, output_file)
+
+    # Check files exist
+    for f in [cert_file, chain_file]:
+        if not os.path.exists(f):
+            print(f"ERROR: File not found: {f}")
+            return False
+
+    print(f"Converting {cert_file} with chain {chain_file} to P7B format...")
+
+    # Create combined cert file
+    combined = "combined_cert.pem"
+    with open(combined, "w") as out:
+        with open(cert_file, "r") as f:
+            out.write(f.read())
+        with open(chain_file, "r") as f:
+            out.write(f.read())
+
+    result = convert_pem_to_p7b(combined, output_file)
+
+    # Clean up combined file
+    if os.path.exists(combined):
+        os.remove(combined)
+
+    return result
+
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser(
+        description="Convert PEM certificate to P7B (PKCS#7) for Square webhooks"
+    )
+    parser.add_argument(
+        "--cert",
+        default="cert.pem",
+        help="Input PEM certificate file (default: cert.pem)",
+    )
+    parser.add_argument(
+        "--chain", default=None, help="Optional: Intermediate certificate chain file"
+    )
+    parser.add_argument(
+        "--output", default="cert.p7b", help="Output P7B file (default: cert.p7b)"
+    )
+
+    args = parser.parse_args()
+
+    print("=" * 60)
+    print("PEM to P7B Certificate Converter for Square Webhooks")
+    print("=" * 60)
+    print()
+
+    if args.chain:
+        success = convert_with_chain(args.cert, args.chain, args.output)
+    else:
+        success = convert_pem_to_p7b(args.cert, args.output)
+
+    if not success:
+        sys.exit(1)
+
+    print("\n✓ Conversion complete!")

src/scripts/edderkop.sql

@@ -1,7 +0,0 @@
-CREATE TABLE "REVIEW" (
-	"sq_object"	TEXT NOT NULL UNIQUE,
-	"time_added"	INTEGER,
-	"set_number"	INTEGER,
-	"time_modified"	INTEGER,
-	PRIMARY KEY("sq_object")
-)

src/scripts/generate_cert.py

@@ -0,0 +1,76 @@
+"""
+Generate self-signed SSL certificate for testing HTTPS webhooks
+Run this script to create cert.pem and key.pem files
+"""
+
+import subprocess
+import sys
+import os
+
+
+def generate_certificate():
+    """Generate self-signed certificate using OpenSSL"""
+
+    cert_file = "cert.pem"
+    key_file = "key.pem"
+
+    # Check if OpenSSL is available
+    try:
+        subprocess.run(["openssl", "version"], capture_output=True, check=True)
+    except (subprocess.CalledProcessError, FileNotFoundError):
+        print("ERROR: OpenSSL not found!")
+        print("\nFor Windows, you can:")
+        print(
+            "1. Install OpenSSL from: https://slproweb.com/products/Win32OpenSSL.html"
+        )
+        print("2. Or use Git Bash which includes OpenSSL")
+        print("3. Or use WSL (Windows Subsystem for Linux)")
+        return False
+
+    print("Generating self-signed certificate...")
+    print("This will create cert.pem and key.pem files")
+    print()
+
+    # Generate certificate valid for 365 days
+    cmd = [
+        "openssl",
+        "req",
+        "-x509",
+        "-newkey",
+        "rsa:4096",
+        "-keyout",
+        key_file,
+        "-out",
+        cert_file,
+        "-days",
+        "365",
+        "-nodes",  # No passphrase
+        "-subj",
+        "/C=US/ST=State/L=City/O=Organization/CN=localhost",
+    ]
+
+    try:
+        subprocess.run(cmd, check=True)
+        print(f"\n✓ Certificate created: {os.path.abspath(cert_file)}")
+        print(f"✓ Private key created: {os.path.abspath(key_file)}")
+        print("\nYou can now run the server with HTTPS:")
+        print(
+            f"  python src/test_server.py --ssl-cert {cert_file} --ssl-key {key_file} --hook-port 8443"
+        )
+        print("\nOr webhook-only HTTPS:")
+        print(
+            f"  python src/test_server.py --ssl-cert {cert_file} --ssl-key {key_file} --hook-port 8443 --hook-ssl-only"
+        )
+        print(
+            "\nNote: Self-signed certificates will show security warnings in browsers."
+        )
+        print("For production, use certificates from a trusted CA like Let's Encrypt.")
+        return True
+    except subprocess.CalledProcessError as e:
+        print(f"ERROR: Failed to generate certificate: {e}")
+        return False
+
+
+if __name__ == "__main__":
+    if not generate_certificate():
+        sys.exit(1)

src/scripts/setup_db.sqlite

@@ -0,0 +1,15 @@
+CREATE TABLE "REVIEW" (
+	"sq_object"	TEXT NOT NULL UNIQUE,
+	"price"	INTEGER NOT NULL,
+	"date_modified"	INTEGER,
+	PRIMARY KEY("sq_object")
+)
+
+CREATE TABLE "MISSING" (
+	"sq_object"	TEXT NOT NULL UNIQUE,
+	"time_added"	INTEGER,
+	"set_number"	INTEGER,
+	"time_modified"	INTEGER,
+	"set_name"	TEXT,
+	PRIMARY KEY("sq_object")
+)

src/scripts/throttler_example.py

@@ -1,15 +1,15 @@
-import aiohttp
-import asyncio
-
-from asyncio_throttle.throttler import Throttler
-
-throttler = Throttler(rate_limit=500, period=60)  # 500 requests per minute
-
-
-async def worker(throttler, session):
-    while True:
-        async with throttler:
-            async with session.get("http://example.com") as resp:
-                do_some_job_with(await resp.text())
-
-        await asyncio.sleep(0.05)
+import aiohttp
+import asyncio
+
+from asyncio_throttle.throttler import Throttler
+
+throttler = Throttler(rate_limit=500, period=60)  # 500 requests per minute
+
+
+async def worker(throttler, session):
+    while True:
+        async with throttler:
+            async with session.get("http://example.com") as resp:
+                do_some_job_with(await resp.text())
+
+        await asyncio.sleep(0.05)

src/stats.py

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

src/test_server.py

@@ -1,121 +0,0 @@
-import asyncio
-import json
-from aiohttp import web
-from base64 import b64decode
-from typing import Dict, Any
-import uuid
-from time import time
-import dotenv
-import os
-import re
-import logging
-from pathlib import Path
-
-Path("./logs/").mkdir(parents=True, exist_ok=True)
-
-import banner
-import get_sq_env
-import asquare as sq
-from abricksetapi import AsyncBricksetAPI
-
-lego = AsyncBricksetAPI(debug=True)
-
-dotenv.load_dotenv()
-SQUARE_LEGO_CAT_ID = os.environ.get("SQUARE_LEGO_CAT_ID")
-
-
-from asyncio_throttle.throttler import Throttler
-
-logger = logging.getLogger("aiohttp.access")
-logging.basicConfig(
-    filename="./logs/edderkop.log", encoding="utf-8", level=logging.DEBUG
-)
-
-UA = re.compile(r"^kopling\/?", re.IGNORECASE)
-
-
-async def run_update():
-    set_list = await sq.acat2id(SQUARE_LEGO_CAT_ID)
-    for l_set in set_list:
-        print(f"[@DEBUG]{l_set}")
-        l_setinfo = await lego.throttled_get_set(l_set["number"])
-        print(f"[@DEBUG]{l_setinfo}")
-    #    await sq.update_itm(l_set, l_setinfo)
-
-
-def valid_uuid(val):
-    try:
-        uuid.UUID(str(val))
-        return True
-    except ValueError:
-        return False
-
-
-async def request_handler(request: web.Request) -> web.Response:
-    """Handle incoming POST requests with command processing"""
-    try:
-        # 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.")
-
-        # Get JSON data from request
-        try:
-            data = await request.json()
-        except json.JSONDecodeError:
-            logger.error("Invalid JSON from %s", request.remote)
-            return web.Response(
-                status=400, text="{time()}\tInvalid JSON in request body"
-            )
-        # Validate Idempotency-Key header
-        if valid_uuid(request.headers.get("Idempotency-Key")) is False:
-            logger.error("{time()}\tInvalid Idempotency-Key from %s", request.remote)
-            return web.Response(status=412, text="412 Headers rejected!")
-        # Validate User-Agent header
-        if not re.match(UA, str(request.headers.get("User-Agent"))):
-            logger.error("{time()}\tInvalid User-Agent from %s", request.remote)
-            return web.Response(status=412, text="412 Headers rejected!")
-        # Validate command exists
-        if "CMD" not in data:
-            logger.error("{time()}\tMissing CMD field from %s", request.remote)
-            return web.Response(status=400, text="Missing CMD field in request")
-
-        # Decode and process command
-        try:
-            decoded_cmd = b64decode(data["CMD"]).decode("utf-8")
-            print(
-                f"""At: {time()}\nReceived command: {decoded_cmd}\nUser-Agent: {request.headers.get('User-Agent')}\nIdempotency-Key: {request.headers.get('Idempotency-Key')}"""
-            )
-            logging.info(
-                f"{time()}\tCommand: {decoded_cmd}\tFrom: {request.remote}\tUA: {request.headers.get('User-Agent')}\tIdempotency-Key: {request.headers.get('Idempotency-Key')}"
-            )
-
-            # Process the command
-            if decoded_cmd == "RUNUPDATE":
-                # Add your command handling logic here
-                await run_update()
-                response_data = {"status": "success", "message": "Update started"}
-                logging.info(f"{time()}\tUpdate command processed successfully")
-
-                return web.json_response(response_data)
-            else:
-                return web.Response(status=400, text=f"Unknown command: {decoded_cmd}")
-
-        except Exception as e:
-            print(f"Error processing command: {str(e)}")
-            return web.Response(status=500, text=f"Server error: {str(e)}")
-
-    except Exception as e:
-        print(f"Request handler error: {str(e)}")
-        return web.Response(status=500, text="Internal server error")
-
-
-app = web.Application()
-app.router.add_post("/", request_handler)
-
-if __name__ == "__main__":
-    print(banner.generate("Edderkop v0.1"))
-    print("Built for PLAY-WELL TEKNOLOGIES by contact@traps.moe\n\n")
-    web.run_app(app, host="localhost", port=7432)

src/throttler.py

@@ -0,0 +1,33 @@
+import asyncio
+from asyncio_throttle import Throttler as AsyncThrottler
+
+class GlobalThrottler:
+    _instance = None
+    _lock = asyncio.Lock()
+    
+    def __init__(self, rate_limit: int = 500, period: int = 60) -> None:
+        if not hasattr(self, '_initialized'):
+            self.rate_limit = rate_limit
+            self.period = period
+            self.throttler = AsyncThrottler(rate_limit=self.rate_limit, period=self.period)
+            self._initialized = True
+    
+    @classmethod
+    async def get_instance(cls, rate_limit: int = 500, period: int = 60) -> 'GlobalThrottler':
+        async with cls._lock:
+            if not cls._instance:
+                cls._instance = cls(rate_limit, period)
+            return cls._instance
+    
+    async def acquire(self):
+        await self.throttler.acquire()
+        
+    def release(self):
+        self.throttler.release()
+    
+    async def __aenter__(self):
+        await self.acquire()
+        return self
+        
+    async def __aexit__(self, exc_type, exc_val, exc_tb):
+        self.release()

studspider.py

@@ -0,0 +1,274 @@
+import os
+import json
+import asyncio
+from typing import Dict, Any
+from fastapi import FastAPI, Request, Response, status
+from fastapi.responses import JSONResponse
+import src.get_sq_env as sq_env
+
+# Square webhook validation
+from square.utils.webhooks_helper import verify_signature
+
+# Configuration
+SQUARE_WEBHOOK_SIGNATURE_KEY = sq_env.SQUARE_WEBHOOK_SIGNATURE_KEY
+SQUARE_WEBHOOK_URL = sq_env.SQUARE_WEBHOOK_URL
+
+# 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 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.py

@@ -1,6 +1,6 @@
-import src.libsquare as sq
-import src.bricksetapi as lego
-
-oursq = sq.get_itm("71417")
-ourlego = lego.get_set(71417)
-print(f"{oursq}\n\n{ourlego}\n\n")
+import src.libsquare as sq
+import src.bricksetapi as lego
+
+oursq = sq.get_itm("71417")
+ourlego = lego.get_set(71417)
+print(f"{oursq}\n\n{ourlego}\n\n")

tests/test_client.py

@@ -1,39 +1,39 @@
-import asyncio
-import aiohttp
-from uuid import uuid4
-from base64 import b64encode
-import os
-
-
-async def idp_request(cmd: str):
-    headers = {
-        "Idempotency-Key": f"{uuid4()}",
-        "Content-Type": "application/json",
-        "User-Agent": "Kopling/0.12",
-    }
-    print(f"[@DEBUG] Idempotency-Key: {headers['Idempotency-Key']}")
-
-    enc_cmd = b64encode(f"{cmd}".encode("utf-8")).decode("utf-8")
-    print("[@DEBUG] Encoded command:", enc_cmd)
-    print("[@DEBUG] Sending request..")
-    try:
-        async with aiohttp.ClientSession() as session:
-            async with session.post(
-                "http://localhost:7432",  # Added http:// protocol
-                headers=headers,
-                json={"CMD": enc_cmd},  # Changed to json instead of data
-            ) as resp:
-                print(f"[@DEBUG] Response status: {resp.status}")
-                print(f"[@DEBUG] Response body: {await resp.text()}")
-    except Exception as e:
-        print(f"\nRequest failed: {e}")
-        os._exit(-1)
-
-
-async def main():
-    # Await the task instead of just creating it
-    await idp_request("RUNUPDATE")
-
-
-if __name__ == "__main__":
-    asyncio.run(main())
+import asyncio
+import aiohttp
+from uuid import uuid4
+from base64 import b64encode
+import os
+
+
+async def idp_request(cmd: str):
+    headers = {
+        "Idempotency-Key": f"{uuid4()}",
+        "Content-Type": "application/json",
+        "User-Agent": "Kopling/0.12",
+    }
+    print(f"[@DEBUG] Idempotency-Key: {headers['Idempotency-Key']}")
+
+    enc_cmd = b64encode(f"{cmd}".encode("utf-8")).decode("utf-8")
+    print("[@DEBUG] Encoded command:", enc_cmd)
+    print("[@DEBUG] Sending request..")
+    try:
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                "http://localhost:7432",  # Added http:// protocol
+                headers=headers,
+                json={"CMD": enc_cmd},  # Changed to json instead of data
+            ) as resp:
+                print(f"[@DEBUG] Response status: {resp.status}")
+                print(f"[@DEBUG] Response body: {await resp.text()}")
+    except Exception as e:
+        print(f"\nRequest failed: {e}")
+        os._exit(-1)
+
+
+async def main():
+    # Await the task instead of just creating it
+    await idp_request("RUNUPDATE")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

tests/test_server.py

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

tests/test_webhook.py

@@ -0,0 +1,102 @@
+"""
+Test script for webhook endpoint
+Sends the Square inventory.count.updated webhook payload
+"""
+
+import asyncio
+import aiohttp
+import json
+import ssl
+import sys
+
+
+async def test_webhook(verify_ssl=True):
+    url = "https://studspider.sh:8443/hook"
+
+    # Your Square webhook payload
+    payload = {
+        "merchant_id": "6SSW7HV8K2ST5",
+        "type": "inventory.count.updated",
+        "event_id": "df5f3813-a913-45a1-94e9-fdc3f7d5e3b6",
+        "created_at": "2019-10-29T18:38:45.455006797Z",
+        "data": {
+            "type": "inventory_counts",
+            "id": "84e4ac73-d605-4dbd-a9e5-ffff794ddb9d",
+            "object": {
+                "inventory_counts": [
+                    {
+                        "calculated_at": "2019-10-29T18:38:45.10296Z",
+                        "catalog_object_id": "FGQ5JJWT2PYTHF35CKZ2DSKP",
+                        "catalog_object_type": "ITEM_VARIATION",
+                        "location_id": "YYQR03DGCTXA4",
+                        "quantity": "10",
+                        "state": "IN_STOCK",
+                    }
+                ]
+            },
+        },
+    }
+
+    print("Sending webhook test payload...")
+    print(f"URL: {url}")
+    print(
+        f"SSL Verification: {'Enabled' if verify_ssl else 'Disabled (accepting self-signed certs)'}"
+    )
+    print(json.dumps(payload, indent=2))
+    print("\n" + "=" * 60 + "\n")
+
+    # Create SSL context that allows self-signed certificates if verify_ssl is False
+    ssl_context = None
+    if not verify_ssl:
+        ssl_context = ssl.create_default_context()
+        ssl_context.check_hostname = False
+        ssl_context.verify_mode = ssl.CERT_NONE
+
+    try:
+        connector = aiohttp.TCPConnector(ssl=ssl_context)
+        async with aiohttp.ClientSession(connector=connector) as session:
+            async with session.post(url, json=payload) as resp:
+                print(f"Status: {resp.status}")
+                print(f"Response headers: {dict(resp.headers)}")
+
+                response_text = await resp.text()
+                print(f"\nResponse body:")
+
+                try:
+                    response_json = json.loads(response_text)
+                    print(json.dumps(response_json, indent=2))
+                except json.JSONDecodeError:
+                    print(response_text)
+
+    except aiohttp.ClientConnectorError as e:
+        print(f"Connection error: {e}")
+        print("\nTroubleshooting:")
+        print("1. Make sure the server is running")
+        print("2. Check if the hostname and port are correct")
+        print("3. For self-signed certificates, the script now accepts them by default")
+        print("\nServer start command:")
+        print(
+            "  python src/test_server.py --ssl-cert cert.pem --ssl-key key.pem --hook-port 8443 --hook-ssl-only"
+        )
+    except aiohttp.ServerTimeoutError as e:
+        print(f"Timeout error: {e}")
+        print("The server took too long to respond")
+    except Exception as e:
+        print(f"Error: {type(e).__name__}: {e}")
+
+
+if __name__ == "__main__":
+    # Check command line arguments
+    verify_ssl = True
+    if len(sys.argv) > 1 and sys.argv[1] in ["--no-verify", "--insecure", "-k"]:
+        verify_ssl = False
+        print("⚠️  SSL verification disabled - accepting self-signed certificates\n")
+    elif len(sys.argv) > 1:
+        print("Usage: python src/test_webhook.py [--no-verify]")
+        print(
+            "  --no-verify: Disable SSL certificate verification (for self-signed certs)"
+        )
+        sys.exit(1)
+
+    # Default to no verification since we're likely using self-signed certs
+    asyncio.run(test_webhook(verify_ssl=False))