app.py

#
# Hey Michael v0.21
# Sep 15, 2023
# 
# v0.21 log
# 1. Send only last question and answer via email
# 2. Add a pay-it-forward link to the email footer.
#

import os
import tweepy
import gradio as gr
import time
from langchain.chat_models import ChatOpenAI
from langchain.prompts.chat import (
    ChatPromptTemplate,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
)
from langchain.schema import HumanMessage, SystemMessage
import requests  # To make API calls to Airtable
import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

# Madril Credentials and Client
import mailchimp_transactional as MailchimpTransactional
from mailchimp_transactional.api_client import ApiClientError
#live key
mailchimp = MailchimpTransactional.Client(os.getenv("MAILCHIMP_LIVE_KEY"))


# Twitter API credentials

#OAUTH 2
#consumer_key = os.environ.get("Tweeter Consumer Key")
#consumer_secret = os.environ.get("Tweeter Consumer Secret")
#access_token = os.environ.get("Tweeter Access Token")
#access_token_secret = os.environ.get("Tweeter Access Token Secret")
#bearer_token = os.environ.get("Tweeter Bearer Token")

consumer_key = os.getenv("TWEETER_CONSUMER_KEY")
consumer_secret = os.getenv("TWEETER_CONSUMER_SECRET")
access_token = os.getenv("TWEETER_ACCESS_TOKEN")
access_token_secret = os.getenv("TWEETER_ACCESS_TOKEN_SECRET")
bearer_token = os.getenv("TWEETER_BEARER_TOKEN")


client = tweepy.Client(
    bearer_token=bearer_token,
    access_token=access_token,
    access_token_secret=access_token_secret,
    consumer_key=consumer_key,
    consumer_secret=consumer_secret)

# Initialize the chat model
#chat = ChatOpenAI(model="gpt-4-0613", temperature=0.0)
#chat = ChatOpenAI(model="gpt-4", temperature=0.0)
#chat = ChatOpenAI(model="gpt-3.5-turbo-16k", temperature=0.25)
chat = ChatOpenAI(model=os.getenv("CHAT_GPT_MODEL"), temperature=os.getenv("CHAT_GPT_TEMPERATURE"))


# Define the system message
#system_message_content = "You are Michael of Nebadon, also known as Jesus of Nazareth. Your answers are exclusively based on the Urantia Book. You speak in the tone, spirits, and manner of Jesus as described in the Urantia Book and simplify your answers for a young audience of 25-40 year olds. In your answer, include specific references to the Papers, Sections, and Paragraphs of the Urantia Book. Write all the references at the end of the answer under a section called References and in the following shorthand format. The format is Paper number followed by a colon followed by the Section number followed by a period followed by the Paragraph number. For example, Paper 1, Section 2, Paragraph 3 is written as 1:2.3"
#system_message_content = "You are Michael of Nebadon, also known as Jesus of Nazareth. Your answers are exclusively based on the Urantia Book. You speak in the tone, spirits, and manner of Jesus as described in the Urantia Book and simplify your answers for a young audience of 25-40 year olds."
system_message_content = os.getenv("CHAT_GPT_DEFAULT_SYSTEM_MESSAGE")

# Create templates
system_template = system_message_content
system_message_prompt = SystemMessagePromptTemplate.from_template(system_template)
human_template = "{text}"
human_message_prompt = HumanMessagePromptTemplate.from_template(human_template)

# Build a chat prompt template
chat_prompt = ChatPromptTemplate.from_messages(
    [system_message_prompt, human_message_prompt]
)

# Initialize an empty list to store the chat history
chat_history = []

def send_to_airtable(question, answer):
    # Replace with your Airtable API URL and API Key
    url = os.getenv("AIRTABLE_URL")
    print (url)
    
    headers = {"Authorization": f"Bearer {os.getenv('AIRTABLE_BEARER_KEY')}"}
    data = {
        "fields": {
            "question": question, 
            "answer": answer,
            "system prompt":system_message_content, 
            "chatbot version":os.getenv("ASKJESUS_CHATBOT_VERSION")
        }
    }
    requests.post(url, headers=headers, json=data)

def send_email_mailchimp(email, history_str):

    # Email content
    subject = "👉 A chatbot session from AskJesus"
    #body = "Someone that cares about you share their AskJesus chatbot session with you:\n\n" + history_str + "\n\n Click here to try the AskJesus chatbot."

    html_intro = "<html><body><p>Someone that cares about you shared their AskJesus chatbot session with you:</p>"
    html_link_to_askjesus_cfu_page = f"<p>Click <a href={os.getenv('ASKJESUS_CFU_PAGE')}>here</a> to try the AskJesus chatbot."
    html_link_to_matching_fund_cfu_page = f"<br/>AskJesus will always be free and ad-free. Will you pay-it-foward so that others can enjoy it? Click <a href={os.getenv('MATCHING_FUND_CFU_PAGE')}>here</a> to donate.</p></body></html>"

    # Replace newline characters with HTML line breaks
    history_html = history_str.replace("\n", "<br/>")

    # Combine everything
    body = html_intro + history_html + html_link_to_askjesus_cfu_page + html_link_to_matching_fund_cfu_page

    message = {
        "from_email": os.getenv("SMTP_USERNAME"),
        "subject": subject,
        "html": body,
        "to": [{"email": email,"type": "to"}]
    }

    try:
        response = mailchimp.messages.send({"message":message})
        print('Mailchimp API called successfully: {}'.format(response))
        gr.Info("Email sent.")

    except ApiClientError as error:
        print('An exception occurred: {}'.format(error.text))
        raise gr.Error("Error: email not sent.")

def tweet_last_qa():
    if len(chat_history) < 1:
        print ("Nothing to tweet.")
        raise gr.Error("Nothing to tweet.")
        return

    last_qa = chat_history[-1]
    tweet_content = f"Question: {last_qa['Question']}\nAskJesus chatbot: {last_qa['AskJesus chatbot']}"
    #tweet_link = f"Click <a href=https://www.thecenterforunity.org/ask-jesus-chatbot>{last_qa['Question']}</a> to try the AskJesus chatbot."
    tweet_link = f"Someone just asked the AskJesus chatbot:\n\n{last_qa['Question']}\n\nWhat would you ask?\n\n#AskJesus #TheCenterforUnity #JesusBeforeChristianity (https://www.thecenterforunity.org/ask-jesus-chatbot)"

    #print (tweet_link)

    
    try:
        client.create_tweet(text=tweet_link)
        print (f"Successfully tweeted the last Q&A. {tweet_link}")
        gr.Info("Tweet sent.")
        return
    
    except Exception as e:
        print (f"Failed to tweet. Error: {e}")
        raise gr.Error("Failed to tweet.")
        return


with gr.Blocks(title=f"AskJesus Chatbot v{os.getenv('ASKJESUS_VERSION')}") as demo:
    
    # Gradio UI
    msg = gr.Textbox(label="Question:", placeholder="Enter question here and press enter")
    chatbot = gr.Chatbot(label=f"AskJesus Chatbot v{os.getenv('ASKJESUS_VERSION')}")
    email_input = gr.Textbox(label="Email:", placeholder="Enter email here then click Send email")

    with gr.Row():
        with gr.Column():
            send_email_button = gr.Button(value="Send email",size='sm', variant="primary")
        with gr.Column():
            tweet_button = gr.Button(value="Tweet last Q&A", size='sm')
        with gr.Column():
            clear = gr.ClearButton([msg, chatbot], value="Clear log", size='sm')

    def user(user_message, history):
        # Append the user's message to the chat history
        chat_history.append({"Question": user_message, "AskJesus chatbot": None})
        return gr.update(value="", interactive=False), history + [[user_message, None]]

    def bot(history):
        # Get a chat completion from the formatted messages
        messages = chat_prompt.format_prompt(text=history[-1][0]).to_messages()
        response = chat(messages)
        bot_message = response.content  # assuming response is an AIMessage

        # Update the chat history with the bot's message
        chat_history[-1]["AskJesus chatbot"] = bot_message

        history[-1][1] = ""
        for character in bot_message:
            history[-1][1] += character
            time.sleep(0.05)
            yield history
        
            # Send the question and answer to Airtable
        send_to_airtable(history[-1][0], history[-1][1])

    def email_session(email):
        if not email:
            print("No email address provided. Cannot send email.")
            raise gr.Error("Please provide email.")
            return

        # Convert the chat history to a string format
        history_str = "\n".join([f"Question: {entry['Question']}\nAskJesus chatbot: {entry['AskJesus chatbot']}\n\n" for entry in chat_history])
        send_email_mailchimp(email, history_str)

    def email_last_qa(email):
        if not email:
            print("No email address provided. Cannot send email.")
            raise gr.Error("Please provide email.")
            return

        # Convert the last question and answer to a string format
        history_str = "\n".join([f"Question: {chat_history[-1]['Question']}\nAskJesus chatbot: {chat_history[-1]['AskJesus chatbot']}\n\n"])
        send_email_mailchimp(email, history_str)

    response = msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False).then(
        bot, chatbot, chatbot
    )
    response.then(lambda: gr.update(interactive=True), None, [msg], queue=False)
    # send_email_button.click(email_session, [email_input])
    send_email_button.click(email_last_qa, [email_input])
    tweet_button.click(tweet_last_qa)

demo.queue()
demo.launch(show_error=True)