import os from fastapi import FastAPI, Request, HTTPException, Form # Keep Form for now, though not used in webhook import uvicorn from gradio_client import Client from fastapi.responses import Response import json import re import asyncio from pydantic import BaseModel # Import BaseModel # Define Pydantic model for incoming Twilio webhook data # Twilio typically sends 'From', 'Body', 'MessageSid', etc. as form data, # but the previous log showed application/json. We'll adapt to expect JSON for now. # If Twilio is configured to send JSON, the keys should match exactly. # Assuming the keys are 'From' and 'Body' based on the previous error. class TwilioWebhookData(BaseModel): From: str Body: str # Add other fields you expect in the JSON payload if needed, e.g.: # MessageSid: str = None # AccountSid: str = None # To: str = None # Connect to your hosted Gradio Space (Futuresony/Mr.Events) # This client is used by BOTH the /chat and /webhook endpoints to interact with the core chatbot try: client = Client("Futuresony/ABSA_Test_Space") print("Gradio Client for 'Futuresony/ABSA_Test_Space' initialized.") except Exception as e: print(f"Error initializing Gradio Client for 'Futuresony/Mr.Events': {e}") print("Ensure the Space name is correct and it is accessible.") client = None # Get your secure API key for THIS FastAPI application and the hosted Space from environment VALID_API_KEY = os.getenv("APP_API_KEY") print(f"APP_API_KEY loaded: {'Yes' if VALID_API_KEY else 'No'}") if not VALID_API_KEY: print("Warning: APP_API_KEY secret not set. API key validation and calls to hosted space may fail.") app = FastAPI() # --- Chat Endpoint (Existing Functionality) --- @app.post("/chat") async def chat(request: Request): """ Handles chat requests via a JSON payload, validates API key, and calls the hosted Gradio chatbot. """ print("\n--- Received POST request at /chat ---") data = await request.json() # API Key Check for THIS FastAPI application api_key = request.headers.get("X-API-Key") print(f"API Key from header: {api_key[:4]}...") if api_key else "No API Key in header" if not VALID_API_KEY or api_key != VALID_API_KEY: print("API Key validation failed.") raise HTTPException(status_code=403, detail="Invalid API Key") print("API Key validation successful.") user_message = data.get("message") if not user_message: print("Error: 'message' is required in the request body.") raise HTTPException(status_code=400, detail="Message is required") print(f"User message: {user_message}") if client is None: print("Error: Gradio Client not initialized. Cannot call chatbot.") raise HTTPException(status_code=500, detail="Chatbot service not available.") try: print(f"Calling hosted Gradio Space 'Futuresony/ABSA_Test_Space' /chat endpoint from /chat...") result = await client.predict( query=user_message, api_key=VALID_API_KEY, api_name="/chat" ) print(f"Received raw result from hosted Space: {result}") assistant_response = result if not isinstance(assistant_response, str): print(f"Warning: Hosted Space returned unexpected result type: {type(assistant_response)}. Raw result: {result}") assistant_response = str(assistant_response) print(f"Formatted assistant response: {assistant_response}") except Exception as e: print(f"Error calling hosted Gradio Space from /chat: {e}") raise HTTPException(status_code=500, detail=f"Error communicating with chatbot model: {e}") return {"response": assistant_response} # --- Twilio Webhook Endpoint --- # In-memory dictionary to store history per sender (NOT for production!) conversation_histories = {} @app.post("/webhook") async def webhook( # Receive JSON data using the Pydantic model data: TwilioWebhookData, request: Request = None ): """ Handles incoming Twilio webhook requests (expecting JSON), processes them with the chatbot, and returns TwiML. Note: This implementation uses in-memory history (NOT for production). """ print("\n--- Received POST request at /webhook from Twilio (expecting JSON) ---") # Access incoming message and sender number from the Pydantic model instance incoming_message = data.Body sender_number = data.From print(f"Parsed Incoming Message: '{incoming_message}' from {sender_number}") # --- Conversation History Management (In-Memory - NOT Persistent!) --- chat_history = conversation_histories.get(sender_number, []) print(f"Retrieved in-memory history for {sender_number}: {chat_history}") # --- Call Chatbot Logic --- if client is None: print("Error: Gradio Client not initialized. Cannot call chatbot from webhook.") bot_response = "Error: Chatbot service is not available." else: try: print(f"Calling hosted Gradio Space 'Futuresony/ABSA_Test_Space' /chat endpoint from /webhook...") print(f" Query: {incoming_message}") result = await client.predict( query=incoming_message, api_key=VALID_API_KEY, api_name="/chat" ) print(f"Received raw result from hosted Space for webhook: {result}") bot_response = result if not isinstance(bot_response, str): print(f"Warning: Hosted Space returned unexpected result type for webhook: {type(bot_response)}. Raw result: {result}") bot_response = str(bot_response) print(f"Formatted chatbot response for webhook: {bot_response}") except Exception as e: print(f"Error calling hosted Gradio Space from /webhook: {e}") bot_response = f"An error occurred while processing your request: {e}" # --- Update and Store History (In-Memory - NOT Persistent!) --- chat_history.append([incoming_message, bot_response]) conversation_histories[sender_number] = chat_history print(f"Updated in-memory history for {sender_number}: {conversation_histories[sender_number]}") # --- Generate TwiML Response --- twiml_response = f'''{bot_response}''' print(f"Generated TwiML response: {twiml_response}") return Response(content=twiml_response, media_type="application/xml") if __name__ == "__main__": print("Starting FastAPI application with Uvicorn...") uvicorn.run(app, host="0.0.0.0", port=7860)