p7b.....

.gitignore

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

HTTPS_SETUP.md

@@ -0,0 +1,125 @@
+# 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.
+
+### 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
+
+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

README.md

@@ -1,11 +1,43 @@
-# Edderkop
+# studspider
-
+
-environment: poetry
+![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)
-code style: black
+
-
+
-built with
+
-
+
-- aiohttp
+is a tool used to talk to brickset.com and a square inventory to pull lego set information for entering in the square db
-- squareup
+## Run me
-- apscheduler
+
-- aiosqlite
+Clone the project
+
+```bash
+  git clone http://git.traps.moe/ph0sgene/studspider.git
+  git checkout uv 
+```
+
+Go to the project directory
+
+```bash
+  cd studspider
+```
+
+Install dependencies
+
+```bash
+  uv sync && uv lock
+```
+
+Start the server
+
+```bash
+  uvx run studspider.py
+```
+
+
+## Tech 
+
+- aiohttp
+- squareup
+- apscheduler
+- aiosqlite
+

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
+```

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

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!")

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)

pyproject.toml

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

setup.py

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

src/abricksetapi.py

@@ -1,137 +1,162 @@
-"""
+"""
-abricksetapi.py - Asynchronous Brickset API Client
+abricksetapi.py - Asynchronous Brickset API Client
-This module provides asynchronous interactions with the Brickset API.
+This module provides asynchronous interactions with the Brickset API.
-"""
+"""
-
+
-import asyncio
+import asyncio
-import os
+import os
-import aiohttp
+import aiohttp
-import json
+import json
-import dotenv
+import dotenv
-from typing import Dict, Any
+from typing import Dict, Any
-from asyncio_throttle.throttler import Throttler
+import throttler
-
+
-dotenv.load_dotenv()
+Throttler = throttler.GlobalThrottler
-
+
-
+dotenv.load_dotenv()
-class AsyncBricksetAPI:
+
-    """Asynchronous client for the Brickset API"""
+
-
+class AsyncBricksetAPI:
-    def __init__(self, debug: bool = False):
+    """Asynchronous client for the Brickset API"""
-        self.username = os.getenv("BRICKSET_USERNAME")
+
-        self.password = os.getenv("BRICKSET_PASS")
+    def __init__(self, debug: bool = False, logger=None):
-        self.api_key = os.getenv("BRICKSET_API_KEY")
+        self.username = os.getenv("BRICKSET_USERNAME")
-        self.debug = debug
+        self.password = os.getenv("BRICKSET_PASS")
-        self.headers = {"UserAgent": "Edderkop/0.1"}
+        self.api_key = os.getenv("BRICKSET_API_KEY")
-        self.throttler = Throttler(rate_limit=500, period=60)
+        self.debug = debug
-        self.user_hash = None
+        self.headers = {"UserAgent": "studspider/0.1"}
-
+        self.throttler = Throttler(rate_limit=500, period=60)
-    async def login(self) -> Dict[str, Any]:
+        self.user_hash = None
-        """Asynchronously authenticate with Brickset API"""
+        self.logger = logger
-        url = "https://brickset.com/api/v3.asmx/login"
+
-        params = {
+    async def login(self) -> Dict[str, Any]:
-            "username": self.username,
+        """Asynchronously authenticate with Brickset API"""
-            "password": self.password,
+        url = "https://brickset.com/api/v3.asmx/login"
-            "apiKey": self.api_key,
+        params = {
-        }
+            "username": self.username,
-
+            "password": self.password,
-        async with aiohttp.ClientSession() as session:
+            "apiKey": self.api_key,
-            async with session.get(
+        }
-                url, params=params, headers=self.headers  # type: ignore
+
-            ) as response:
+        async with aiohttp.ClientSession() as session:
-                if self.debug:
+            async with session.get(
-                    print(f"[@DEBUG] Login response status: {response.status}")
+                url, params=params, headers=self.headers  # type: ignore
-
+            ) as response:
-                data = await response.read()
+                if self.debug:
-                result = json.loads(data.decode("utf-8"))
+                    print(f"[@DEBUG] Login response status: {response.status}")
-                self.user_hash = result.get("hash")
+                if self.logger:
-                return result
+                    self.logger.debug(f"Login response status: {response.status}")
-
+
-    async def get_set(self, set_number: str) -> Dict[str, Any]:
+                data = await response.read()
-        """
+                result = json.loads(data.decode("utf-8"))
-        Fetch LEGO set information asynchronously
+                self.user_hash = result.get("hash")
-
+
-        Args:
+                if self.logger:
-            set_number (str): LEGO set number
+                    self.logger.info("Successfully authenticated with Brickset API")
-
+                return result
-        Returns:
+
-            Dict containing set information
+    async def get_set(self, set_number: int) -> Dict[str, Any]:
-        """
+        """
-        if not self.user_hash:
+        Fetch LEGO set information asynchronously
-            await self.login()
+
-
+        Args:
-        url = "https://brickset.com/api/v3.asmx/getSets"
+            set_number (str): LEGO set number
-        params = {
+
-            "apiKey": self.api_key,
+        Returns:
-            "userHash": self.user_hash,
+            Dict containing set information
-            "params": json.dumps({"setNumber": f"{set_number}-1"}),
+        """
-        }
+        if not self.user_hash:
-
+            await self.login()
-        async with aiohttp.ClientSession() as session:
+
-            async with session.get(
+        url = "https://brickset.com/api/v3.asmx/getSets"
-                url, params=params, headers=self.headers
+        params = {
-            ) as response:
+            "apiKey": self.api_key,
-                if self.debug:
+            "userHash": self.user_hash,
-                    print(f"[@DEBUG] Get set response status: {response.status}")
+            "params": json.dumps({"setNumber": f"{set_number}-1"}),
-
+        }
-                if response.status == 200:
+        print(f"[@DEBUG] {params}...")
-                    data = await response.read()
+
-                    result = json.loads(data.decode("utf-8"))
+        async with aiohttp.ClientSession() as session:
-
+            async with session.get(
-                    if not result.get("sets"):
+                url, params=params, headers=self.headers
-                        raise ValueError(f"No data found for set {set_number}")
+            ) as response:
-
+                if self.debug:
-                    set_data = result["sets"][0]
+                    print(f"[@DEBUG] Get set response status: {response.status}")
-                    our_set = {
+                if self.logger:
-                        "set_num": set_data["number"],
+                    self.logger.debug(f"Get set response status: {response.status}")
-                        "set_name": set_data["name"],
+
-                        "set_theme": set_data["theme"],
+                if response.status == 200:
-                        "set_image": set_data["image"]["thumbnailURL"],
+                    data = await response.read()
-                        "set_barcode": set_data["barcode"]["UPC"],
+                    result = json.loads(data.decode("utf-8"))
-                        "set_retailprice": round(
+
-                            set_data["LEGOCom"]["US"]["retailPrice"]
+                    if not result.get("sets"):
-                        ),
+                        error_msg = f"No data found for set {set_number}"
-                    }
+                        if self.logger:
-
+                            self.logger.error(error_msg)
-                    # Calculate adjusted price (10% markup)
+                        raise ValueError(error_msg)
-                    retail_price = our_set["set_retailprice"]
+
-                    our_set["set_adjustredprice"] = retail_price + (
+                    if self.logger:
-                        (10 / 100) * retail_price
+                        self.logger.info(
-                    )
+                            f"Successfully retrieved data for set {set_number}"
-
+                        )
-                    # Set PW inventory name with theme
+
-                    our_set["pw_name"] = (
+                    set_data = result["sets"][0]
-                        f"{our_set['set_num']} - {our_set['set_theme'].upper()} {our_set['set_name']}"
+                    our_set = {
-                    )
+                        "set_num": set_data["number"],
-
+                        "set_name": set_data["name"],
-                    # Flag for manual price review if needed
+                        "set_theme": set_data["theme"],
-                    if our_set["set_retailprice"] >= 110:
+                        "set_image": set_data["image"]["thumbnailURL"],
-                        our_set["requires_manual_price"] = True
+                        "set_barcode": set_data["barcode"]["UPC"],
-
+                        "set_retailprice": round(
-                    return our_set
+                            set_data["LEGOCom"]["US"]["retailPrice"]
-
+                        ),
-                elif response.status == 500:
+                    }
-                    raise Exception(
+
-                        "Internal Server Error: Please check API key and parameters."
+                    # Calculate adjusted price (10% markup)
-                    )
+                    retail_price = our_set["set_retailprice"]
-                else:
+                    our_set["set_adjustredprice"] = retail_price + (
-                    raise Exception(f"Unexpected response: {response.status}")
+                        (10 / 100) * retail_price
-
+                    )
-    async def throttled_get_set(self, set_number: str) -> Dict[str, Any]:
+
-        """Get set information with rate limiting"""
+                    # Set PW inventory name with theme
-        async with self.throttler:
+                    our_set["pw_name"] = (
-            return await self.get_set(set_number)
+                        f"{our_set['set_num']} - {our_set['set_theme'].upper()} {our_set['set_name']}"
-
+                    )
-
+
-async def main():
+                    # Flag for manual price review if needed
-    """Example usage"""
+                    if our_set["set_retailprice"] >= 110:
-    client = AsyncBricksetAPI(debug=True)
+                        our_set["requires_manual_price"] = True
-    try:
+
-        set_info = await client.throttled_get_set("10349")
+                    return our_set
-        print(json.dumps(set_info, indent=2))
+
-    except Exception as e:
+                elif response.status == 500:
-        print(f"Error: {e}")
+                    error_msg = (
-
+                        "Internal Server Error: Please check API key and parameters."
-
+                    )
-if __name__ == "__main__":
+                    if self.logger:
-    asyncio.run(main())
+                        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
 
@@ -103,7 +106,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 +125,10 @@ async def acat2id(cat_id: str):
 
 
 async def test() -> None:
-    task = asyncio.create_task(acat2id(LEGO_CAT_ID))  # type: ignore
+    result = await acat2id(LEGO_CAT_ID)  # type: ignore
-    await asyncio.sleep(1)  # Give the task some time to complete
-    result = await task
     print("OBJECTS IN LEGO CATEGORY:")
     for item in result:
-        print(f"ID: {item['id']}, Name: {item['name']}")
+        print(f"ID: {item['id']}, Name: {item['name']}, Updated: {item['updated_at']}")
-    # eturn
-    # pass
 
 
 """

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/ebricksetapi.py

@@ -15,7 +15,7 @@ class BricksetAPI:
         self.password = os.getenv("BRICKSET_PASS")
         self.api_key = os.getenv("BRICKSET_API_KEY")
         self.debug = debug
-        self.headers = {"UserAgent": "Edderkop/0.1"}
+        self.headers = {"UserAgent": "studspider/0.1"}
         self.user_hash = None
 
     def login(self) -> Dict[str, Any]:

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/scheduler.py

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

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/edderkop.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 aiohttp
-import asyncio
+import asyncio
-
+
-from asyncio_throttle.throttler import Throttler
+from asyncio_throttle.throttler import Throttler
-
+
-throttler = Throttler(rate_limit=500, period=60)  # 500 requests per minute
+throttler = Throttler(rate_limit=500, period=60)  # 500 requests per minute
-
+
-
+
-async def worker(throttler, session):
+async def worker(throttler, session):
-    while True:
+    while True:
-        async with throttler:
+        async with throttler:
-            async with session.get("http://example.com") as resp:
+            async with session.get("http://example.com") as resp:
-                do_some_job_with(await resp.text())
+                do_some_job_with(await resp.text())
-
+
-        await asyncio.sleep(0.05)
+        await asyncio.sleep(0.05)

src/server.py

@@ -74,7 +74,7 @@ async def handle_post(request):
                     return True
         else:
             return aiohttp.web.Response(status=400, text="400 Bad Request.")
-        result = {"message": f"Edderkop/0.1 received command: {reqdata}"}
+        result = {"message": f"studspider/0.1 received command: {reqdata}"}
         return aiohttp.web.json_response(status=200, data=result)
 
 
@@ -108,6 +108,6 @@ async def main():
 
 
 if __name__ == "__main__":
-    print(generate_toilet_banner("Edderkop v0.1"))
+    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/test_server.py

@@ -1,45 +1,79 @@
+import argparse
 import asyncio
 import json
+import ssl
 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 logging.handlers import RotatingFileHandler
 from pathlib import Path
 
-Path("./logs/").mkdir(parents=True, exist_ok=True)
+print("Creating log directory if it doesn't exist...")
+os.makedirs("./logs/", exist_ok=True)
 
 import banner
 import get_sq_env
+from time import time
+
+time = int(time())
 import asquare as sq
+
+# from asquareclass import AsyncSquareClient  # type: ignore
 from abricksetapi import AsyncBricksetAPI
 
-lego = AsyncBricksetAPI(debug=True)
+print("Loading environment variables...")
-
 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")
+# set up logger
-logging.basicConfig(
+print("Setting up logger...")
-    filename="./logs/edderkop.log", encoding="utf-8", level=logging.DEBUG
+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)
+    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}")
-        l_setinfo = await lego.throttled_get_set(l_set["number"])
+        try:
-        print(f"[@DEBUG]{l_setinfo}")
+            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)
 
 
@@ -51,14 +85,89 @@ def valid_uuid(val):
         return False
 
 
+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.")
+
+        # Get JSON data from request
+        try:
+            data = await request.json()
+        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")
+
+        # 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: {data.get('type', 'unknown')}")
+        print(f"[@DEBUG] Event ID: {data.get('event_id', 'none')}")
+        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 here
+        # TODO: Add your webhook processing logic
+        webhook_type = data.get("type", "unknown")
+
+        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')}")
+
+        response_data = {
+            "status": "success",
+            "message": f"Webhook received: {webhook_type}",
+            "timestamp": current_time,
+            "event_id": data.get("event_id"),
+            "data_received": data,
+        }
+
+        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(
+            logger.error(f"{time}\tInvalid request method from %s at ", request.remote)
-                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
@@ -86,19 +195,17 @@ async def request_handler(request: web.Request) -> web.Response:
         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')}"""
+                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')}"
+                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")
+                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}")
@@ -112,10 +219,129 @@ async def request_handler(request: web.Request) -> web.Response:
         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("/", request_handler)
+app.router.add_post("/api", request_handler)
+app.router.add_post("/hook", hook_handler)
+
+parser = argparse.ArgumentParser(description="studspider")
+parser.add_argument(
+    "--host",
+    type=str,
+    default="localhost",
+    help="IP address to listen on. default: localhost",
+)
+parser.add_argument(
+    "--port", type=int, default=7432, help="Port to listen on. default: 7432"
+)
+parser.add_argument(
+    "--hook-port",
+    type=int,
+    default=8080,
+    help="Optional separate port for webhook endpoint. default: 8080",
+)
+parser.add_argument(
+    "--ssl-cert",
+    type=str,
+    default=None,
+    help="Path to SSL certificate file for HTTPS (PEM format)",
+)
+parser.add_argument(
+    "--ssl-key",
+    type=str,
+    default=None,
+    help="Path to SSL private key file for HTTPS (PEM format)",
+)
+parser.add_argument(
+    "--hook-ssl-only",
+    action="store_true",
+    help="Enable HTTPS only for webhook endpoint (main server stays HTTP)",
+)
+args = parser.parse_args()
 
 if __name__ == "__main__":
-    print(banner.generate("Edderkop v0.1"))
+    print(banner.generate("studspider v0.1"))
     print("Built for PLAY-WELL TEKNOLOGIES by contact@traps.moe\n\n")
-    web.run_app(app, host="localhost", port=7432)
+
+    # 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.")
+            print("Generate certificates with: python generate_cert.py")
+            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)

src/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))

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

@@ -1,14 +1,14 @@
-import src.test_server
+import src.test_server
-
+
-
+
-class edderkop:
+class Studspider:
-    def __init__(self):
+    def __init__(self):
-        self.name = "Edderkop"
+        self.name = "Studspider"
-        self.version = "0.0.1"
+        self.version = "0.0.1"
-        self.description = """
+        self.description = """
-        Edderkop is an application designed to interact with the Brickset API and directly with Square.
+        studspider is an application designed to interact with the Brickset API and directly with Square.
-        It allows users to search for items, manage collections, and perform various operations related to bricksets.
+        It allows users to search for items, manage collections, and perform various operations related to bricksets.
-        """
+        """
-
+
-    def run(self):
+    def run(self):
-        print(f"{self.name} v{self.version} is running.")
+        print(f"{self.name} v{self.version} is running.")

tests/test.py

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