cap.py

import os
import asyncio
import json
import base64
import logging

from dotenv import load_dotenv
from quart import Quart, Response, request, websocket
import websockets
import google.generativeai as genai
from pinecone import Pinecone, ServerlessSpec

# Configure logging
logging.basicConfig(level=logging.INFO, 
                    format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Load environment variables
load_dotenv()

# API keys
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
GENAI_API_KEY = os.getenv("GOOGLE_API_KEY")
PINECONE_API_KEY = os.getenv("PINECONE_API_KEY")

# Verify required keys
missing_keys = []
if not OPENAI_API_KEY: missing_keys.append("OPENAI_API_KEY")
if not GENAI_API_KEY: missing_keys.append("GOOGLE_API_KEY")
if not PINECONE_API_KEY: missing_keys.append("PINECONE_API_KEY")

if missing_keys:
    raise ValueError(f"Missing required API keys: {', '.join(missing_keys)}")

# Initialize Gemini API
genai.configure(api_key=GENAI_API_KEY)     

# Initialize Pinecone
pc = Pinecone(api_key=PINECONE_API_KEY)
INDEX_NAME = "voice-bot-gemini-embedding-004-index"

# Check if index exists, create if needed
try:
    index_names = pc.list_indexes().names()
    if INDEX_NAME not in index_names:
        logger.info(f"Creating Pinecone index: {INDEX_NAME}")
        pc.create_index(
            name=INDEX_NAME,
            dimension=768,
            metric="cosine",
            spec=ServerlessSpec(cloud="aws", region="us-east-1"),
        )
    index = pc.Index(INDEX_NAME)
    logger.info(f"Connected to Pinecone index: {INDEX_NAME}")
except Exception as e:
    logger.error(f"Pinecone setup error: {e}")
    raise

# Initialize Quart app
app = Quart(__name__)
PORT = int(os.getenv("PORT", "5000"))

# Default product data namespace
DEFAULT_NAMESPACE = "Tec Nviirons Sample data testing.xlsx 2025-05-02 09:05:24"

# System message for OpenAI
SYSTEM_MESSAGE = """You are a helpful voice assistant for a tech product company.
Whenever a user asks a product-related question, you must always call the function `search_product_database` using the given query.
Don't try to answer from memory.
Keep responses concise and conversational."""

# Vector search function
async def search_product_database(query, namespace=DEFAULT_NAMESPACE):
    """Search product information in vector database"""
    try:
        # Logs the query and namespace so developers can see what's being searched.
        logger.info(f"Searching for: '{query}' in namespace '{namespace}'")
        
        # Converts the user's natural language query into a vector embedding (a list of numbers).
        # The embedding captures the semantic meaning of the query.
        # It uses Gemini's text embedding model text-embedding-004.
        embed_response = genai.embed_content(
            model="models/text-embedding-004", 
            content=query
        )
        embedding = embed_response["embedding"]
        
        # Searches Pinecone using the generated embedding.
        # Finds the top 3 most relevant documents (vectors) from the specified namespace.
        # Also fetches associated metadata (e.g., text description of the product).
        results = index.query(
            vector=embedding,
            namespace=namespace,
            top_k=3,
            include_metadata=True
        )

        # If Pinecone returns no matches, the assistant politely replies that it found nothing.
        
        if not results["matches"]:
            return "I couldn't find information about that product in our database."
            
        # Extracts the actual product information text from the top matches.
        # Each match may contain a metadata["text"] field — that's the useful part.
        
        contexts = []
        for match in results["matches"]:
            if "text" in match.get("metadata", {}):
                contexts.append(match["metadata"]["text"])
        
        # Sometimes matches are returned but don't contain actual info (e.g., missing text field).
        if not contexts:
            return "I found some matches but they don't contain usable information."
            

        # Combines all the relevant product texts into one block.
# Forms a new prompt asking Gemini: "Here's the info — now answer this question."
# Uses the gemini-1.5-flash model to generate a fast, summarized, human-like answer
# Returns the summary text to the user.

        context_text = "\n---\n".join(contexts)
        summary_prompt = f"""
        Based on these product details:
        {context_text}
        
        Answer this question concisely: {query}
        """
        
        summary = genai.GenerativeModel("gemini-1.5-flash").generate_content(summary_prompt)
        return summary.text
        

        # If anything breaks (API failure, timeout, etc.), logs the error and gives the user a graceful fallback response.

    except Exception as e:
        logger.error(f"Vector search error: {e}")
        return f"Sorry, I encountered an error searching our product database."


# This defines an asynchronous HTTP endpoint at /webhook.

# It handles incoming HTTP GET or POST requests (likely from Plivo, a voice call service).

# This is the entry point for incoming voice calls from Plivo.


@app.route("/webhook", methods=["GET", "POST"])
async def webhook():
    """Handle incoming Plivo calls"""

    # Reads the caller's phone number from the From query parameter in the request URL.

# For example: /webhook?From=+1234567890

# If From is not provided, it defaults to "Unknown".

# Logs the incoming call with caller info for tracking/debugging
    caller = request.args.get("From", "Unknown")
    logger.info(f"Incoming call from: {caller}")
    
    # Create WebSocket URL for streaming
    # Grabs the current server's host (e.g., localhost:5000 or your deployed domain).

# Constructs a WebSocket URL that Plivo will connect to for streaming audio.

# The app must handle this at /media-stream elsewhere.

# This is where the audio will be sent in real time
    host = request.host
    stream_url = f"ws://{host}/media-stream"
    
    # Generate Plivo XML response
    response = f"""<?xml version="1.0" encoding="UTF-8"?>
    <Response>
        <Speak voice="Polly.Amy">Welcome to our product help line. How can I assist you today?</Speak>

        
        <Stream streamTimeout="60" keepCallAlive="true" bidirectional="true" contentType="audio/x-mulaw;rate=8000">
            {stream_url}
        </Stream>
    </Response>
    """
    # This tells Plivo to:

# Stream the audio of the call to your server (at /media-stream).

# Use the audio format: audio/x-mulaw with 8000 Hz sample rate (standard for telephony).

# Keep the call alive while streaming (keepCallAlive="true").

# Allow two-way streaming (bidirectional="true").

# Automatically stop streaming after 60 seconds (streamTimeout="60").

    return Response(response, mimetype="application/xml")



# Registers an asynchronous WebSocket endpoint at /media-stream.
# Plivo will stream live call audio here using a WebSocket.
# This is where your app listens and processes real-time audio from callers.

@app.websocket("/media-stream")

# Defines an async function to manage 2-way audio streaming.
# This is where audio is received from Plivo and forwarded to OpenAI, and where responses come back.
async def media_stream():
    """WebSocket handler for bidirectional audio streaming"""

# Logs that a client (Plivo) has connected to this WebSocket.
# Stores the Quart WebSocket object (websocket) in a local variable plivo_ws for clarity.
    logger.info("WebSocket connection opened")
    plivo_ws = websocket
    
    # active: A flag to control the loop or stream state later (not used yet).
    # stream_id: Placeholder for a unique stream ID (if Plivo provides one or you assign one later).
    active = True
    stream_id = None
    
    # Specifies the OpenAI WebSocket endpoint for real-time streaming using GPT-4o (October 2024 preview).
    # Sets the Authorization header with your API key.
    # Uses the beta header to enable real-time functionality: realtime=v1
    openai_url = "wss://api.openai.com/v1/realtime?model=gpt-4o-realtime-preview-2024-10-01"
    headers = {
        "Authorization": f"Bearer {OPENAI_API_KEY}",
        "OpenAI-Beta": "realtime=v1"
    }
    
    try:

        # Establishes a WebSocket connection to OpenAI.
        # openai_ws will be used to send and receive messages (audio + text) to/from GPT-4o in real time.
        async with websockets.connect(openai_url, extra_headers=headers) as openai_ws:
            logger.info("Connected to OpenAI WebSocket")
            
            # Configure OpenAI session
            # This block informs OpenAI how you want to interact with the assistant. Here's what each part does:
            # "type": "session.update" – tells OpenAI you're configuring a session.
            # "turn_detection": "server_vad" – uses server-side voice activity detection to know when the user is done speaking.
            # "input_audio_format": "g711_ulaw" – this is the audio format Plivo uses (standard for calls).
            # "output_audio_format": "g711_ulaw" – ensures the bot responds in a format that Plivo can play.
            # "voice": "alloy" – picks a pre-defined AI voice for responses.
            # "instructions" – system message to guide assistant behavior (like "you are a helpful voice assistant").
            # "modalities" – supports both text and audio interaction.
            # "tools" – integrates your custom function search_product_database, so GPT can call it during conversation.

            session_config = {
                "type": "session.update",
                "session": {
                    "turn_detection": {"type": "server_vad"},
                    "input_audio_format": "g711_ulaw",
                    "output_audio_format": "g711_ulaw",
                    "voice": "alloy",
                    "instructions": SYSTEM_MESSAGE,
                    "modalities": ["text", "audio"],
                    "tools": [{
                        "type": "function",
                        "function": {
                            "name": "search_product_database",
                            "description": "Search for product information",
                            "parameters": {
                                "type": "object",
                                "properties": {
                                    "query": {
                                        "type": "string",
                                        "description": "Product query"
                                    },
                                    "namespace": {
                                        "type": "string",
                                        "description": "Database namespace",
                                        "default": DEFAULT_NAMESPACE
                                    }
                                },
                                "required": ["query"]
                            }
                        }
                    }]
                }
            }
            
            # Sends the session config to OpenAI as a JSON message to initialize the assistant session with your settings and tools.
            await openai_ws.send(json.dumps(session_config))
        #    is session config correct we will test it.
            # Handle messages from Plivo
            # Defines an async function that runs in a loop to process Plivo's WebSocket messages.
            # nonlocal means it modifies the stream_id and active variables declared in the outer media_stream() scope.
            async def handle_plivo():
                nonlocal stream_id, active
                
            # Keeps running while the active flag is True — essentially as long as the call is live and streaming.
                while active:
                    try:
                        # Receive message from Plivo
                        # Waits for a message from the Plivo WebSocket (caller audio/events) with a 5-second timeout.
                        # If a message is received, it's parsed as JSON into data
                        message = await asyncio.wait_for(plivo_ws.receive(), timeout=5.0)
                        data = json.loads(message)
                        
                        # Handle stream start
                        # This block is triggered when Plivo sends a start event.
                        # Extracts and stores the stream ID (unique per call).
                        # Logs that the audio stream has begun.
                        if data.get("event") == "start":
                            stream_id = data["start"]["streamId"]
                            logger.info(f"Stream started, ID: {stream_id}")
                        
                        # Forward audio to OpenAI
                        # This block handles audio packets sent by Plivo during the call.
                        # Plivo sends audio in small base64-encoded chunks via media events.
                        # Converts it into a message that OpenAI's WebSocket expects:
                        # type: input_audio_buffer.append
                        # Sends the audio chunk to OpenAI in real-time.
                        #⚠️ Format of the audio is g711_ulaw, which was configured earlier.
                        elif data.get("event") == "media" and "media" in data:
                            audio_data = {
                                "type": "input_audio_buffer.append",
                                "audio": data["media"]["payload"]
                            }
                            await openai_ws.send(json.dumps(audio_data))
                        
                        # When the Plivo stream sends a stop event:
                        # Logs that the stream ended.
                        # Sets active = False to exit the loop.
                        # Breaks the loop immediately.
                        elif data.get("event") == "stop":
                            logger.info("Stream stopped by Plivo")
                            active = False
                            break
                    
                    # If no data is received in 5 seconds, just skip this loop and continue checking.
                    except asyncio.TimeoutError:
                        continue

                    # If any other exception occurs (e.g. malformed message, connection drop), logs it and exits the loop.
                    except Exception as e:
                        logger.error(f"Plivo handler error: {e}")
                        break
            
            # Handle messages from OpenAI
            # Defines an async coroutine.

            # Uses the nonlocal keyword to reference and potentially change the active flag from the outer scope, which controls the lifetime of the streaming session.
            # Listens for responses from OpenAI (text, audio, tool calls).
            # If audio is received, sends it to the caller via Plivo.
            # If a tool call is made (like search_product_database), executes it and sends results back to OpenAI.
            
            async def handle_openai():
                nonlocal active
                
                try:

                    # Listens continuously to the OpenAI WebSocket connection (openai_ws).
                    # Breaks the loop and stops processing if active becomes False.
                    async for message in openai_ws:
                        if not active:
                            break
                            
                        # Parses the received WebSocket message from OpenAI (assumed to be JSON).
                        # Determines the type of message, such as audio, tool calls, etc.
                        response = json.loads(message)
                        response_type = response.get("type", "")
                        
                        # Handle audio responses (TTS output)
                        # OpenAI is sending audio output back (like TTS = text-to-speech).
                        if response_type == "response.audio.delta":
                            try:

                                # Constructs a playAudio event for Plivo to play audio to the caller.
                                # The payload is already base64-encoded audio in G.711 µ-law format (audio/x-mulaw at 8000Hz), which is compatible with Plivo.
                                # Sends it back to the plivo_ws WebSocket so the caller hears the AI's voice.
                                audio_payload = response["delta"]
                                audio_message = {
                                    "event": "playAudio",
                                    "media": {
                                        "contentType": "audio/x-mulaw",
                                        "sampleRate": 8000,
                                        "payload": audio_payload  # Already base64 encoded
                                    }
                                }
                                await plivo_ws.send(json.dumps(audio_message))
                            except Exception as e:
                                logger.error(f"Error sending audio to Plivo: {e}")
                        
                        # This means OpenAI decided to call an external function (tool) — in this case, search_product_database.
                        elif response_type == "response.tool_calls":

                            # Loops through one or more tool call entries (in case multiple are triggered simultaneously).
                            for tool_call in response.get("tool_calls", []):

                                # Confirms the tool is the one we defined.
                                # Parses arguments like query and namespace.
                                if tool_call.get("function", {}).get("name") == "search_product_database":
                                    try:
                                        # Parse arguments
                                        args = json.loads(tool_call["function"]["arguments"])
                                        query = args["query"]
                                        namespace = args.get("namespace", DEFAULT_NAMESPACE)
                                        
                                        # Get product info
                                        logger.info(f"Product search request: {query}")
                                        # Calls the async function to query the Pinecone vector DB and get matching product info.
                                        product_info = await search_product_database(query, namespace)
                                        
                                        # Sends the result of the tool call back to OpenAI so it can include the info in its next response.
                                        result = {
                                            "type": "response.tool_calls.result",
                                            "id": tool_call["id"],
                                            "result": product_info
                                        }
                                        await openai_ws.send(json.dumps(result))

                                    except Exception as e:
                                        logger.error(f"Tool call error: {e}")
                                        # Sends an appropriate fallback message back to OpenAI.
                                        error_result = {
                                            "type": "response.tool_calls.result",
                                            "id": tool_call["id"],
                                            "result": "I had trouble finding that information."
                                        }
                                        await openai_ws.send(json.dumps(error_result))
                
                # Handles a clean disconnection from OpenAI.
                except websockets.ConnectionClosed:
                    logger.info("OpenAI WebSocket closed")
                    active = False

                    # Handles any error in the loop and sets active = False to stop the session.
                except Exception as e:
                    logger.error(f"OpenAI handler error: {e}")
                    active = False
            
            #This runs handle_plivo() and handle_openai() concurrently, so the app can:
            # Stream audio to OpenAI.
            # Receive responses and play them back to the caller.
            # Handle tool invocations live.
            await asyncio.gather(
                handle_plivo(),
                handle_openai()
            )
    
    except websockets.ConnectionClosed:
        logger.info("WebSocket connection closed")
    except Exception as e:
        logger.error(f"WebSocket error: {e}")
    finally:

        # Logs that the WebSocket session is being cleaned up — ensures graceful shutdown.
        logger.info("Closing WebSocket connection")

if __name__ == "__main__":
    logger.info(f"Starting server on port {PORT}")
    app.run(host="0.0.0.0", port=PORT)