Upload 3 files

Dockerfile

@@ -0,0 +1,25 @@
+# Base image using Python 3.9
+FROM python:3.9
+
+# Create a new user to run the app
+RUN useradd -m -u 1000 user
+USER user
+
+# Set environment variables
+ENV PATH="/home/user/.local/bin:$PATH"
+
+# Set the working directory
+WORKDIR /app
+
+# Copy the requirements and install dependencies
+COPY --chown=user ./requirements.txt requirements.txt
+RUN pip install --no-cache-dir --upgrade -r requirements.txt
+
+# Copy the rest of the application
+COPY --chown=user . /app
+
+# Expose port 7860 for the application
+EXPOSE 7860
+
+# Command to run the FastAPI app using uvicorn
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

main.py

@@ -0,0 +1,272 @@
+import os
+import io
+import base64
+import asyncio
+from fastapi import FastAPI, UploadFile, File, Body, Form, HTTPException
+from fastapi.responses import JSONResponse, Response
+from pymongo import MongoClient
+from cartesia import Cartesia
+from groq import Groq
+from dotenv import load_dotenv
+
+# Load environment variables from a .env file
+load_dotenv()
+
+# ---------------------------
+# API Client Setup for Groq and Cartesia
+# ---------------------------
+GROQ_API_KEY = os.getenv("GROQ_API_KEY")
+groq_client = Groq(api_key=GROQ_API_KEY)
+
+CARTESIA_API_KEY = os.getenv("CARTESIA_API_KEY")
+cartesia_client = Cartesia(api_key=CARTESIA_API_KEY)
+
+# ---------------------------
+# OpenAI Chat Client Setup (using langchain_openai)
+# ---------------------------
+OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+from langchain_openai import ChatOpenAI
+
+llm = ChatOpenAI(
+    model="gpt-4o",
+    temperature=1,
+    max_tokens=1024,
+    api_key=OPENAI_API_KEY
+)
+
+# ---------------------------
+# FastAPI and MongoDB Setup
+# ---------------------------
+app = FastAPI()
+
+MONGO_DETAILS = "mongodb://localhost:27017"
+mongo_client = MongoClient(MONGO_DETAILS)
+database = mongo_client["contacts_db"]
+contacts_collection = database["contacts"]
+chat_history_collection = database["chat_history"]
+
+# ---------------------------
+# Pydantic Models
+# ---------------------------
+from pydantic import BaseModel, EmailStr, Field
+from typing import List, Optional
+
+class Contact(BaseModel):
+    name: str = Field(..., example="John Doe")
+    phone: str = Field(..., example="+1234567890")
+    email: Optional[EmailStr] = Field(None, example="john.doe@example.com")
+
+class IncomingMessage(BaseModel):
+    phone: str = Field(..., example="+1234567890")
+    message: str = Field(..., example="Hello, can we chat?")
+
+# ---------------------------
+# Chat History Helper Functions
+# ---------------------------
+def get_chat_history(caller_number: str) -> List[dict]:
+    """
+    Retrieve the conversation history for a given caller from MongoDB.
+    Each entry is a dict with 'role' and 'content' keys.
+    """
+    doc = chat_history_collection.find_one({"caller_number": caller_number})
+    if doc and "messages" in doc:
+        return doc["messages"]
+    return []
+
+def update_chat_history(caller_number: str, role: str, content: str):
+    """
+    Append a new message (with role and content) to the chat history for the given caller.
+    If no document exists, one is created (upsert).
+    """
+    chat_history_collection.update_one(
+        {"caller_number": caller_number},
+        {"$push": {"messages": {"role": role, "content": content}}},
+        upsert=True
+    )
+
+# ---------------------------
+# Conversation Simulation Functions
+# ---------------------------
+def simulate_text_conversation(caller_number: str, initial_message: str) -> str:
+    """
+    For incoming text messages from unknown numbers, use the LLM with a humorous detective prompt.
+    The LLM wastes the spammer's time with a witty, drawn-out conversation.
+    """
+    chat_history = get_chat_history(caller_number)
+    messages = [
+        {
+            "role": "system",
+            "content": (
+                "You are Detective Quip, a witty, sarcastic detective with a flair for humor. "
+                "Your mission is to waste the time of potential spammers by engaging them in an overly elaborate, "
+                "investigative conversation. Ask funny and overly detailed questions, and use your detective skills to slowly uncover "
+                "their dubious intentions—all while making humorous remarks. Your goal is to keep the spammer engaged for at least 30 seconds with "
+                "a minimum of three back-and-forth exchanges. Answers shouls be concise, humorous, and investigative."
+            )
+        }
+    ]
+    # Append previous conversation history if available
+    for entry in chat_history:
+        messages.append({"role": entry["role"], "content": entry["content"]})
+    messages.append({"role": "user", "content": initial_message})
+    
+    # Call the LLM for a response and convert the response to string
+    assistant_response = llm.invoke(messages)
+    if hasattr(assistant_response, "content"):
+        assistant_response = assistant_response.content
+    else:
+        assistant_response = str(assistant_response)
+    
+    # Log the conversation
+    update_chat_history(caller_number, "user", initial_message)
+    update_chat_history(caller_number, "assistant", assistant_response)
+    
+    return assistant_response
+
+def simulate_call_conversation(caller_number: str, initial_message: str) -> str:
+    """
+    For voice calls from unknown numbers, use the LLM with a detective-on-the-phone prompt.
+    The LLM must waste the spammer's time by engaging in a drawn-out, humorous, detective-style conversation.
+    """
+    chat_history = get_chat_history(caller_number)
+    messages = [
+        {
+            "role": "system",
+            "content": (
+                "You are Detective Quip on a phone call. Your tone is a mix of gritty detective seriousness and playful humor. "
+                "A spammer has just called and said something suspicious. Engage them in a lengthy, multi-turn conversation filled with sarcastic remarks, "
+                "overly detailed questions, and witty banter designed to waste their time for at least 30 seconds and at least three conversational turns. "
+                "Make sure your responses are concise, humorous and investigative."
+            )
+        }
+    ]
+    for entry in chat_history:
+        messages.append({"role": entry["role"], "content": entry["content"]})
+    messages.append({"role": "user", "content": initial_message})
+    
+    assistant_response = llm.invoke(messages)
+    if hasattr(assistant_response, "content"):
+        assistant_response = assistant_response.content
+    else:
+        assistant_response = str(assistant_response)
+    
+    update_chat_history(caller_number, "user", initial_message)
+    update_chat_history(caller_number, "assistant", assistant_response)
+    
+    return assistant_response
+
+# ---------------------------
+# Endpoints for Contacts, Texts, and Call Forwarding Setup
+# ---------------------------
+@app.post("/contacts", response_model=List[Contact])
+def create_contacts(contacts: List[Contact]):
+    """
+    Save a list of contacts into MongoDB.
+    """
+    contacts_to_insert = [contact.dict() for contact in contacts]
+    result = contacts_collection.insert_many(contacts_to_insert)
+    if not result.inserted_ids:
+        raise HTTPException(status_code=500, detail="Error inserting contacts")
+    return contacts
+
+@app.post("/incoming-message")
+def process_incoming_message(incoming: IncomingMessage):
+    """
+    Process an incoming text message:
+      - If the sender's number is in contacts, forward it normally.
+      - Otherwise, simulate a multi-turn conversation using the AI bot.
+    """
+    contact = contacts_collection.find_one({"phone": incoming.phone})
+    if contact:
+        return {
+            "status": "contact_found",
+            "detail": f"Primary: {incoming.phone} – '{incoming.message}'"
+        }
+    else:
+        conversation_result = simulate_text_conversation(incoming.phone, incoming.message)
+        return {
+            "status": "not_in_contacts",
+            "conversation_result": conversation_result
+        }
+
+@app.post("/setup-call-forwarding")
+def setup_call_forwarding():
+    """
+    Simulate call forwarding setup.
+    """
+    forwarding_number = "+1-555-123-4567"
+    return {"status": "success", "message": f"Setup done! Calls forwarded to {forwarding_number}"}
+
+# ---------------------------
+# STT and TTS Functions for Voice Calls
+# ---------------------------
+def transcribe_audio(audio_file: bytes) -> str:
+    """
+    Convert incoming audio to text using Groq Whisper v3 (STT).
+    """
+    response = groq_client.audio.transcriptions.create(
+        file=("audio.m4a", audio_file),
+        model="whisper-large-v3",
+        response_format="verbose_json"
+    )
+    return response.text
+
+def text_to_speech(text: str) -> bytes:
+    """
+    Convert text to speech using Cartesia TTS.
+    """
+    audio_bytes = cartesia_client.tts.bytes(
+        model_id="sonic",
+        transcript=text,
+        voice_id="694f9389-aac1-45b6-b726-9d9369183238",  # Example voice
+        output_format={
+            "container": "wav",
+            "encoding": "pcm_f32le",
+            "sample_rate": 44100,
+        },
+    )
+    return audio_bytes
+
+# ---------------------------
+# Endpoint for Processing Voice Calls
+# ---------------------------
+@app.post("/process-call")
+async def process_call(caller_number: str = Form(...), audio: UploadFile = File(...)):
+    """
+    Process an incoming voice call:
+      - If the caller is in contacts, return a normal "ringing" message.
+      - Otherwise, transcribe the audio (STT), simulate a multi-turn conversation using the AI bot (with detective humor), 
+        log the conversation, and return a TTS audio response.
+    """
+    # Check if caller is in contacts
+    contact = contacts_collection.find_one({"phone": caller_number})
+    if contact:
+        ringing_text = f"Call from {caller_number} – Ringing"
+        response_audio = text_to_speech(ringing_text)
+        return Response(content=response_audio, media_type="audio/wav")
+    
+    try:
+        audio_bytes = await audio.read()
+    except Exception as e:
+        raise HTTPException(status_code=400, detail=f"Error reading audio file: {str(e)}")
+    
+    try:
+        transcription = transcribe_audio(audio_bytes)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error during transcription: {str(e)}")
+    
+    try:
+        conversation_result = simulate_call_conversation(caller_number, transcription)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error during AI conversation: {str(e)}")
+    
+    try:
+        response_audio = text_to_speech(conversation_result)
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Error during TTS conversion: {str(e)}")
+    
+    return Response(content=response_audio, media_type="audio/wav")
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

requirements.txt

@@ -0,0 +1,9 @@
+fastapi 
+uvicorn 
+pymongo 
+python-dotenv
+python-multipart
+twilio 
+websockets 
+openai
+langchain_openai