import os import json import asyncio from typing import Dict, Any from fastapi import FastAPI, Request, Response, status from fastapi.responses import JSONResponse import src.get_sq_env as sq_env import src.stats as stats # Square webhook validation from square.utils.webhooks_helper import verify_signature # Configuration SQUARE_WEBHOOK_SIGNATURE_KEY = sq_env.SQUARE_WEBHOOK_SIGNATURE_KEY SQUARE_WEBHOOK_URL = sq_env.SQUARE_WEBHOOK_URL SQUARE_LEGO_CAT_ID = sq_env.LEGO_CAT_ID # Debug: Print configuration only in development or once per process # When running under gunicorn, the on_starting hook handles this if not os.getenv("GUNICORN_CMD_ARGS"): 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)", "stats": "/stats (GET)", }, } @app.get("/health") async def health_check(): """Health check endpoint for monitoring""" return { "status": "healthy", "signature_verification": ( "enabled" if SQUARE_WEBHOOK_SIGNATURE_KEY else "disabled" ), } @app.get("/stats") async def get_stats(): """ Get daily statistics for LEGO inventory. Returns metrics about enrichment progress, missing data, catalog value, and cache stats. """ try: daily_stats = await stats.get_daily_stats() cache_stats = await stats.get_brickset_cache_stats() # Calculate completion percentage if daily_stats["total_items"] > 0: completion_rate = ( daily_stats["enriched_items"] / daily_stats["total_items"] * 100 ) else: completion_rate = 0.0 return { "status": "success", "version": "0.3.0", "git_commit": stats.get_git_commit(), "stats": daily_stats, "completion_rate": round(completion_rate, 1), "brickset_cache": cache_stats, } except Exception as e: return JSONResponse( status_code=500, content={"status": "error", "message": str(e)}, ) @app.post("/hook") async def webhook_handler(request: Request): """ Handle Square webhook events - MUST respond within 2 seconds! Square sends POST requests with: - x-square-hmacsha256-signature header - JSON body with event data - event_id for idempotency """ try: # Log incoming request print(f"\n[WEBHOOK] Incoming request from {request.client.host}") print(f"[WEBHOOK] Method: {request.method}") print(f"[WEBHOOK] URL: {request.url}") print(f"[WEBHOOK] Headers: {dict(request.headers)}") # Read raw body for signature verification body = await request.body() body_str = body.decode("utf-8") print(f"[WEBHOOK] Body length: {len(body_str)} bytes") # Get signature header signature_header = request.headers.get("x-square-hmacsha256-signature", "") # Verify Square webhook signature (if configured) if SQUARE_WEBHOOK_SIGNATURE_KEY and SQUARE_WEBHOOK_URL: if not signature_header: print( f"[ERROR] Missing x-square-hmacsha256-signature header from {request.client.host}" ) return JSONResponse( status_code=status.HTTP_401_UNAUTHORIZED, content={"error": "Missing signature header"}, ) # Debug signature verification inputs print(f"[DEBUG] Signature header: {signature_header[:20]}...") print( f"[DEBUG] Signature key (first 10 chars): {SQUARE_WEBHOOK_SIGNATURE_KEY[:10]}..." ) print(f"[DEBUG] Notification URL: {SQUARE_WEBHOOK_URL}") print(f"[DEBUG] Body (first 100 chars): {body_str[:100]}...") try: is_valid = verify_signature( request_body=body_str, signature_header=signature_header, signature_key=SQUARE_WEBHOOK_SIGNATURE_KEY, notification_url=SQUARE_WEBHOOK_URL, ) if not is_valid: print( f"[ERROR] Invalid webhook signature from {request.client.host}" ) print(f"[ERROR] This usually means:") print( f"[ERROR] 1. SQUARE_WEBHOOK_URL doesn't match Square Dashboard exactly" ) print(f"[ERROR] 2. SQUARE_WEBHOOK_SIGNATURE_KEY is incorrect") print(f"[ERROR] 3. Request body was modified in transit") return JSONResponse( status_code=status.HTTP_401_UNAUTHORIZED, content={"error": "Invalid signature"}, ) print(f"[INFO] ✓ Webhook signature verified from {request.client.host}") except Exception as e: print(f"[ERROR] Signature verification exception: {e}") return JSONResponse( status_code=status.HTTP_401_UNAUTHORIZED, content={"error": f"Signature verification failed: {str(e)}"}, ) else: print( f"[WARN] Webhook signature verification DISABLED - configure SQUARE_WEBHOOK_SIGNATURE_KEY" ) # Parse JSON data try: data = json.loads(body_str) except json.JSONDecodeError as e: print(f"[ERROR] Invalid JSON from {request.client.host}: {e}") return JSONResponse( status_code=status.HTTP_400_BAD_REQUEST, content={"error": "Invalid JSON"}, ) # Check idempotency (event_id) event_id = data.get("event_id") if event_id: if event_id in processed_event_ids: print( f"[INFO] Duplicate event {event_id} - already processed (idempotency)" ) return JSONResponse( status_code=status.HTTP_200_OK, content={ "status": "success", "message": "Event already processed", "event_id": event_id, }, ) # Add to processed set processed_event_ids.add(event_id) # Prevent memory growth if len(processed_event_ids) > MAX_STORED_EVENT_IDS: processed_event_ids.pop() # Get event info for quick response webhook_type = data.get("type", "unknown") event_id_val = data.get("event_id", "none") merchant_id = data.get("merchant_id", "unknown") # RESPOND IMMEDIATELY - Square requires fast response (< 2 seconds) response_data = { "status": "success", "message": "Webhook received", "event_id": event_id_val, } # Process webhook in background (async task) asyncio.create_task( process_webhook( webhook_type, event_id_val, merchant_id, data, request.client.host ) ) # Return immediate response print(f"[INFO] 200 OK response sent for event {event_id_val}") return JSONResponse(status_code=status.HTTP_200_OK, content=response_data) except Exception as e: # Catch-all for any unexpected errors - still respond quickly print(f"[ERROR] Unexpected error in webhook handler: {e}") import traceback traceback.print_exc() return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"error": "Internal server error"}, ) # Return immediate response return JSONResponse(status_code=status.HTTP_200_OK, content=response_data) async def process_webhook( webhook_type: str, event_id: str, merchant_id: str, data: Dict[str, Any], client_ip: str, ): """ Process webhook data asynchronously after responding to Square This runs in the background and doesn't block the HTTP response """ try: print("\n" + "=" * 60) print(f"[WEBHOOK] Received at {asyncio.get_event_loop().time()}") print(f"[WEBHOOK] From: {client_ip}") print(f"[WEBHOOK] Type: {webhook_type}") print(f"[WEBHOOK] Event ID: {event_id}") print(f"[WEBHOOK] Merchant: {merchant_id}") print(f"[WEBHOOK] Full payload:") print(json.dumps(data, indent=2)) print("=" * 60 + "\n") # Process based on webhook type if webhook_type == "inventory.count.updated": print(f"[PROCESS] Processing inventory count update...") inventory_counts = ( data.get("data", {}).get("object", {}).get("inventory_counts", []) ) # Import here to avoid circular imports import src.asquare as asquare import src.glue as glue for count in inventory_counts: variation_id = count.get("catalog_object_id") quantity = count.get("quantity") state = count.get("state") print( f"[PROCESS] Catalog: {variation_id}, Qty: {quantity}, State: {state}" ) try: # Get parent item from variation ID item = await asquare.get_item_by_variation_id(variation_id) item_id = item.get("id") item_name = item.get("name", "") category_id = item.get("category_id") print(f"[PROCESS] Item: {item_id} - {item_name}") print(f"[PROCESS] Category: {category_id}") # Only process LEGO category items if category_id == SQUARE_LEGO_CAT_ID: print( f"[PROCESS] ✓ LEGO item detected - triggering enrichment" ) # Trigger enrichment in background with item data to avoid redundant lookup asyncio.create_task( glue.enrich_single_item(item_id, item_data=item) ) else: print(f"[PROCESS] ✗ Not a LEGO item - skipping enrichment") except Exception as e: print(f"[ERROR] Failed to process variation {variation_id}: {e}") elif webhook_type == "catalog.version.updated": print(f"[PROCESS] Processing catalog version update...") # Import here to avoid circular imports import src.asquare as asquare import src.glue as glue catalog_object = data.get("data", {}).get("object", {}) object_type = catalog_object.get("type") if object_type == "ITEM": item_id = catalog_object.get("id") print(f"[PROCESS] Item updated: {item_id}") try: # Get item details to check category item = await asquare.get_item_by_id(item_id) item_name = item.get("name", "") category_id = item.get("category_id") print(f"[PROCESS] Item: {item_id} - {item_name}") print(f"[PROCESS] Category: {category_id}") # Only process LEGO category items if category_id == SQUARE_LEGO_CAT_ID: print( f"[PROCESS] ✓ LEGO item detected - triggering enrichment" ) # Trigger enrichment in background with item data to avoid redundant lookup asyncio.create_task( glue.enrich_single_item(item_id, item_data=item) ) else: print(f"[PROCESS] ✗ Not a LEGO item - skipping enrichment") except Exception as e: print(f"[ERROR] Failed to process item {item_id}: {e}") else: print(f"[PROCESS] Object type '{object_type}' - skipping") else: print(f"[PROCESS] Webhook type '{webhook_type}' - no handler configured") print(f"[PROCESS] ✓ Webhook processing complete for {event_id}") except Exception as e: print(f"[ERROR] Failed to process webhook {event_id}: {e}") import traceback traceback.print_exc() if __name__ == "__main__": # For development/testing - use uvicorn directly import uvicorn print("\n" + "=" * 60) print("Studspider webhook boot dev") print("=" * 60) print( f"Signature verification: {'ENABLED' if SQUARE_WEBHOOK_SIGNATURE_KEY else 'DISABLED'}" ) print(f"Webhook URL: {SQUARE_WEBHOOK_URL or 'NOT CONFIGURED'}") print("\nFor production, use Gunicorn:") print( " gunicorn conf/studspider:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:7433" ) print("=" * 60 + "\n") uvicorn.run(app, host="0.0.0.0", port=7433, log_level="info")