rishithM/voicecloneBot

1import os2import re3import requests4import json5import gradio as gr6from langchain.chat_models import ChatOpenAI7from langchain import LLMChain, PromptTemplate8from langchain.memory import ConversationBufferMemory9 10OPENAI_API_KEY=os.getenv('OPENAI_API_KEY')11PLAY_HT_API_KEY=os.getenv('PLAY_HT_API_KEY')12PLAY_HT_USER_ID=os.getenv('PLAY_HT_USER_ID')13 14PLAY_HT_VOICE_ID=os.getenv('PLAY_HT_VOICE_ID')15play_ht_api_get_audio_url = "https://play.ht/api/v2/tts"16 17 18template = """You are a helpful assistant to answer user queries.19{chat_history}20User: {user_message}21Chatbot:"""22 23prompt = PromptTemplate(24    input_variables=["chat_history", "user_message"], template=template25)26 27memory = ConversationBufferMemory(memory_key="chat_history")28 29llm_chain = LLMChain(30    llm=ChatOpenAI(temperature='0.5', model_name="gpt-3.5-turbo"),31    prompt=prompt,32    verbose=True,33    memory=memory,34)35 36headers = {37      "accept": "text/event-stream",38      "content-type": "application/json",39      "AUTHORIZATION": "Bearer "+ PLAY_HT_API_KEY,40      "X-USER-ID": PLAY_HT_USER_ID41}42 43 44def get_payload(text):45  return {46    "text": text,47    "voice": PLAY_HT_VOICE_ID,48    "quality": "medium",49    "output_format": "mp3",50    "speed": 1,51    "sample_rate": 24000,52    "seed": None,53    "temperature": None54  }55 56def get_generated_audio(text):57  payload = get_payload(text)58  generated_response = {}59  try:60      response = requests.post(play_ht_api_get_audio_url, json=payload, headers=headers)61      response.raise_for_status()62      generated_response["type"]= 'SUCCESS'63      generated_response["response"] = response.text64  except requests.exceptions.RequestException as e:65      generated_response["type"]= 'ERROR'66      try:67        response_text = json.loads(response.text)68        if response_text['error_message']:69          generated_response["response"] = response_text['error_message']70        else:71          generated_response["response"] = response.text72      except Exception as e:73        generated_response["response"] = response.text74  except Exception as e:75    generated_response["type"]= 'ERROR'76    generated_response["response"] = response.text77  return generated_response78 79def extract_urls(text):80    # Define the regex pattern for URLs81    url_pattern = r'https?://(?:[-\w.]|(?:%[\da-fA-F]{2}))+[/\w\.-]*'82 83    # Find all occurrences of URLs in the text84    urls = re.findall(url_pattern, text)85 86    return urls87 88def get_audio_reply_for_question(text):89  generated_audio_event = get_generated_audio(text)90  #From get_generated_audio, you will get events in a string format, from that we need to extract the url91  final_response = {92      "audio_url": '',93      "message": ''94  }95  if generated_audio_event["type"] == 'SUCCESS':96    audio_urls = extract_urls(generated_audio_event["response"])97    if len(audio_urls) == 0:98      final_response['message'] = "No audio file link found in generated event"99    else:100      final_response['audio_url'] = audio_urls[-1]101  else:102    final_response['message'] = generated_audio_event['response']103  return final_response104 105def download_url(url):106    try:107        # Send a GET request to the URL to fetch the content108        final_response = {109            'content':'',110            'error':''111        }112        response = requests.get(url)113        # Check if the request was successful (status code 200)114        if response.status_code == 200:115            final_response['content'] = response.content116        else:117            final_response['error'] = f"Failed to download the URL. Status code: {response.status_code}"118    except Exception as e:119        final_response['error'] = f"Failed to download the URL. Error: {e}"120    return final_response121 122def get_filename_from_url(url):123    # Use os.path.basename() to extract the file name from the URL124    file_name = os.path.basename(url)125    return file_name126 127def get_text_response(user_message):128    response = llm_chain.predict(user_message = user_message)129    return response130 131def get_text_response_and_audio_response(user_message):132    response = get_text_response(user_message) # Getting the reply from Open AI133    audio_reply_for_question_response = get_audio_reply_for_question(response)134    final_response = {135        'output_file_path': '',136        'message':''137    }138    audio_url = audio_reply_for_question_response['audio_url']139    if audio_url:140      output_file_path=get_filename_from_url(audio_url)141      download_url_response = download_url(audio_url)142      audio_content = download_url_response['content']143      if audio_content:144        with open(output_file_path, "wb") as audio_file:145          audio_file.write(audio_content)146          final_response['output_file_path'] = output_file_path147      else:148          final_response['message'] = download_url_response['error']149    else:150      final_response['message'] = audio_reply_for_question_response['message']151    return final_response152 153def chat_bot_response(message, history):154    text_and_audio_response = get_text_response_and_audio_response(message)155    output_file_path = text_and_audio_response['output_file_path']156    if output_file_path:157      return (text_and_audio_response['output_file_path'],)158    else:159      return text_and_audio_response['message']160 161demo = gr.ChatInterface(chat_bot_response,examples=["How are you doing?","What are your interests?","Which places do you like to visit?"])162 163if __name__ == "__main__":164    demo.launch() #To create a public link, set `share=True` in `launch()`. To enable errors and logs, set `debug=True` in `launch()`.165