src/alpine_bits_python/api.py

@@ -573,6 +573,61 @@ async def handle_wix_form_test(
         raise HTTPException(status_code=500, detail="Error processing test data")
 
 
+@api_router.post("/webhook/generic")
+@webhook_limiter.limit(WEBHOOK_RATE_LIMIT)
+async def handle_generic_webhook(request: Request, data: dict[str, Any]):
+    """Handle generic webhook endpoint for receiving JSON payloads.
+
+    Logs the data to file for later analysis. Does not process the data
+    or save to database since the structure is not yet known.
+
+    No authentication required for this endpoint.
+    """
+    try:
+        timestamp = datetime.now().isoformat()
+        _LOGGER.info("Received generic webhook data at %s", timestamp)
+
+        # Create log entry with metadata
+        log_entry = {
+            "timestamp": timestamp,
+            "client_ip": request.client.host if request.client else "unknown",
+            "headers": dict(request.headers),
+            "data": data,
+            "origin_header": request.headers.get("origin"),
+        }
+
+        # Create logs directory if it doesn't exist
+        logs_dir = Path("logs/generic_webhooks")
+        if not logs_dir.exists():
+            logs_dir.mkdir(parents=True, mode=0o755, exist_ok=True)
+            _LOGGER.info("Created directory: %s", logs_dir)
+
+        # Generate log filename with timestamp
+        log_filename = (
+            logs_dir / f"webhook_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json"
+        )
+
+        # Write log file
+        with log_filename.open("w", encoding="utf-8") as f:
+            json.dump(log_entry, f, indent=2, default=str, ensure_ascii=False)
+
+        _LOGGER.info("Generic webhook data logged to: %s", log_filename)
+
+    except Exception as e:
+        _LOGGER.exception("Error in handle_generic_webhook")
+        raise HTTPException(
+            status_code=500, detail="Error processing generic webhook data"
+        ) from e
+    else:
+        return {
+            "status": "success",
+            "message": "Generic webhook data received successfully",
+            "data_logged_to": str(log_filename),
+            "timestamp": timestamp,
+            "note": "Data logged for later analysis",
+        }
+
+
 @api_router.put("/hoteldata/conversions_import/{filename:path}")
 @limiter.limit(DEFAULT_RATE_LIMIT)
 async def handle_xml_upload(

tests/test_api.py

@@ -264,6 +264,54 @@ class TestWixWebhookEndpoint:
         assert data["status"] == "success"
 
 
+class TestGenericWebhookEndpoint:
+    """Test generic webhook endpoint."""
+
+    def test_generic_webhook_success(self, client):
+        """Test successful generic webhook submission."""
+        test_data = {
+            "event_type": "test_event",
+            "data": {
+                "key1": "value1",
+                "key2": "value2",
+                "nested": {"foo": "bar"},
+            },
+            "metadata": {"source": "test_system"},
+        }
+
+        response = client.post("/api/webhook/generic", json=test_data)
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "success"
+        assert "timestamp" in data
+        assert "data_logged_to" in data
+        assert "generic_webhooks" in data["data_logged_to"]
+        assert data["note"] == "Data logged for later analysis"
+
+    def test_generic_webhook_empty_payload(self, client):
+        """Test generic webhook with empty payload."""
+        response = client.post("/api/webhook/generic", json={})
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "success"
+
+    def test_generic_webhook_complex_nested_data(self, client):
+        """Test generic webhook with complex nested data structures."""
+        complex_data = {
+            "arrays": [1, 2, 3],
+            "nested": {"level1": {"level2": {"level3": "deep"}}},
+            "mixed": [{"a": 1}, {"b": 2}],
+        }
+
+        response = client.post("/api/webhook/generic", json=complex_data)
+
+        assert response.status_code == 200
+        data = response.json()
+        assert data["status"] == "success"
+
+
 class TestAlpineBitsServerEndpoint:
     """Test AlpineBits server endpoint."""