handler_best.py

from pydantic import BaseModel
import openai
from environs import Env
from typing import List, Dict, Any


import requests

def download_env_file(url: str, local_path: str):
    response = requests.get(url)
    response.raise_for_status()  # Ensure we notice bad responses
    with open(local_path, 'wb') as f:
        f.write(response.content)

# Download the .env file
env_file_url = "https://www.dropbox.com/scl/fi/21ldek2cdsak2v3mhyy5x/openai.env?rlkey=nxdkd8l8esdy8npa3vfgvqkhp&st=s2f2zzwl&dl=1"  # Adjusted URL for direct download
local_env_path = "openai.env"
download_env_file(env_file_url, local_env_path)

# Load environment variables
env = Env()
env.read_env("openai.env")
openai.api_key = env.str("OPENAI_API_KEY")

# Constants
MODEL = env.str("MODEL", "gpt-3.5-turbo")
AI_RESPONSE_TIMEOUT = env.int("AI_RESPONSE_TIMEOUT", 20)

class EndpointHandler:
    def __init__(self, model_dir=None):
        self.model_dir = model_dir

    def __call__(self, data: Dict[str, Any]) -> Dict[str, Any]:
        try:
            if "inputs" in data:  # Check if data is in Hugging Face JSON format
                return self.process_hf_input(data)
            else:
                return self.process_json_input(data)
        except ValueError as e:
            return {"error": str(e)}
        except Exception as e:
            return {"error": str(e)}

    def process_json_input(self, json_data):
        if "FromUserKavasQuestions" in json_data and "Chatmood" in json_data:
            prompt = self.create_conversation_starter_prompt(
                json_data["FromUserKavasQuestions"],
                json_data["Chatmood"]
            )
            starter_suggestion = self.generate_conversation_starters(prompt)
            return {"conversation_starter": starter_suggestion}
        elif "LastChatMessages" in json_data:
            last_chat_messages = json_data["LastChatMessages"][-4:]
            response = {
                "version": "1.0.0-alpha",
                "suggested_responses": self.get_conversation_suggestions(last_chat_messages)
            }
            return response
        else:
            raise ValueError("Invalid JSON structure.")

    def process_hf_input(self, hf_data):
        print("Received HF Data:", hf_data)  # Debugging line
        if "inputs" in hf_data:
            actual_data = hf_data["inputs"]
            print("Processing actual data:", actual_data)  # Debugging line
            return self.process_json_input(actual_data)
        else:
            return {"error": "Invalid Hugging Face JSON structure."}


    def create_conversation_starter_prompt(self, user_questions, chatmood):
        formatted_info = " ".join([f"{qa['Question']} - {qa['Answer']}" for qa in user_questions if qa['Answer']])
        prompt = (f"Based on user profile info and a {chatmood} mood, "
                  f"generate 3 subtle and very short conversation starters. "
                  f"Explore various topics like travel, hobbies, movies, and not just culinary tastes. "
                  f"\nProfile Info: {formatted_info}")
        return prompt

    def generate_conversation_starters(self, prompt):
        try:
            response = openai.ChatCompletion.create(
                model=MODEL,
                messages=[{"role": "system", "content": prompt}],
                temperature=0.7,
                max_tokens=100,
                n=1,
                request_timeout=AI_RESPONSE_TIMEOUT
            )
            return response.choices[0].message["content"]
        except openai.error.OpenAIError as e:
            raise Exception(f"OpenAI API error: {str(e)}")
        except Exception as e:
            raise Exception(f"Unexpected error: {str(e)}")

    def transform_messages(self, last_chat_messages):
        t_messages = []
        for chat in last_chat_messages:
            if "fromUser" in chat:
                from_user = chat['fromUser']
                message = chat.get('touser', '')
                t_messages.append(f"{from_user}: {message}")
            elif "touser" in chat:
                to_user = chat['touser']
                message = chat.get('fromUser', '')
                t_messages.append(f"{to_user}: {message}")
        
        if t_messages and "touser" in last_chat_messages[-1]:
            latest_message = t_messages[-1]
            latest_message = f"Q: {latest_message}"
            t_messages[-1] = latest_message
        
        return t_messages

    def generate_system_prompt(self, last_chat_messages, fromusername, tousername, zodiansign=None, chatmood=None):
        prompt = ""
        if not last_chat_messages or ("touser" not in last_chat_messages[-1]):
            prompt = f"Suggest a casual and friendly message for {fromusername} to start a conversation with {tousername} or continue naturally, as if talking to a good friend. Strictly avoid replying to messages from {fromusername} or answering their questions."
        else:
            prompt = f"Suggest a warm and friendly reply for {fromusername} to respond to the last message from {tousername}, as if responding to a dear friend. Strictly avoid replying to messages from {fromusername} or answering their questions."
        
        if zodiansign:
            prompt += f" Keep in mind {tousername}'s {zodiansign} zodiac sign."
        
        if chatmood:
            if chatmood == "Casual Vibes":
                prompt += " Keep the conversation relaxed and informal."
            elif chatmood == "Flirty Fun":
                prompt += " Add a playful and teasing tone to the conversation."
            elif chatmood == "Deep and Thoughtful":
                prompt += " Encourage reflective and introspective responses."
            elif chatmood == "Humor Central":
                prompt += " Incorporate witty and humorous elements into the conversation."
            elif chatmood == "Romantic Feels":
                prompt += " Express affection and use sweet and romantic language."
            elif chatmood == "Intellectual Banter":
                prompt += " Engage in thought-provoking discussions on topics like books and movies."
            elif chatmood == "Supportive Mode":
                prompt += " Offer empathy, support, and encouragement in the conversation."
            elif chatmood == "Curiosity Unleashed":
                prompt += " Show eagerness to learn and explore interests by asking questions."
            elif chatmood == "Chill and Easygoing":
                prompt += " Maintain a relaxed and laid-back tone in the conversation."
            elif chatmood == "Adventurous Spirit":
                prompt += " Share travel stories and plans with enthusiasm and energy."
        
        return prompt

    def get_conversation_suggestions(self, last_chat_messages):
        fromusername = last_chat_messages[-1].get("fromusername", "")
        tousername = last_chat_messages[-1].get("tousername", "")
        zodiansign = last_chat_messages[-1].get("zodiansign", "")
        chatmood = last_chat_messages[-1].get("Chatmood", "")
        
        messages = self.transform_messages(last_chat_messages)
        
        system_prompt = self.generate_system_prompt(last_chat_messages, fromusername, tousername, zodiansign, chatmood)
        messages_final = [{"role": "system", "content": system_prompt}]
        
        if messages:
            messages_final.extend([{"role": "user", "content": m} for m in messages])
        else:
            # If there are no messages, add a default message to ensure a response is generated
            default_message = f"{tousername}: Hi there!"
            messages_final.append({"role": "user", "content": default_message})
        
        try:
            response = openai.ChatCompletion.create(
                model=MODEL,
                messages=messages_final,
                temperature=0.7,
                max_tokens=150,
                n=3,
                request_timeout=AI_RESPONSE_TIMEOUT
            )
            
            formatted_replies = []
            for idx, choice in enumerate(response.choices):
                formatted_replies.append({
                    "type": "TEXT",
                    "body": choice.message['content'],
                    "title": f"AI Reply {idx + 1}",
                    "confidence": 1,
                })
            
            return formatted_replies
        
        except openai.error.Timeout as e:
            formatted_reply = [{
                "type": "TEXT",
                "body": "Request to the AI response generator has timed out. Please try again later.",
                "title": "AI Response Error",
                "confidence": 1
            }]
            return formatted_reply