app.py

import gradio as gr
import subprocess
import json
import requests
from bs4 import BeautifulSoup

"""
General helper functions
"""

def strip_html_tags(html_text):
    # Use BeautifulSoup to parse and clean HTML content
    soup = BeautifulSoup(html_text, 'html.parser')
    return soup.get_text()

"""
Padlet API Interactions
"""
def api_call(input_text):

    #TODO: Refactor to be one function that can get OR post

    curl_command = [
        'curl', '-s', '--request', 'GET',
        '--url', f"https://api.padlet.dev/v1/boards/{board_id}?include=posts%2Csections",
        '--header', 'X-Api-Key: pdltp_0e380a0de1ff32d77b12dbcc030b1373199b7525681ddc81bd1b9ef3e4e3dd49577a23',
        '--header', 'accept: application/vnd.api+json'
    ]
    
    try:
        response = subprocess.check_output(curl_command, universal_newlines=True)
        response_data = json.loads(response)
        
        # Extract the contents of all posts, stripping HTML tags from bodyHtml
        posts_data = response_data.get("included", [])
        post_contents = []
        for post in posts_data:
            if post.get("type") == "post":
                attributes = post.get("attributes", {}).get("content", {})
                subject = attributes.get("subject", "")
                body_html = attributes.get("bodyHtml", "")
                
                if subject:
                    post_content = f"Subject: {subject}"
                    if body_html:
                        cleaned_body = strip_html_tags(body_html)
                        post_content += f"\nBody Text: {cleaned_body}"
                    
                    post_contents.append(post_content)
        
        return "\n\n".join(post_contents) if post_contents else "No post contents found."
    except subprocess.CalledProcessError:
        return "Error: Unable to fetch data using cURL."

def create_post(board_id, post_content):

    curl_command = [
        'curl', '-s', '--request', 'POST',
        '--url', f"https://api.padlet.dev/v1/boards/{board_id}/posts",
        '--header', 'X-Api-Key: pdltp_0e380a0de1ff32d77b12dbcc030b1373199b7525681ddc81bd1b9ef3e4e3dd49577a23',
        '--header', 'accept: application/vnd.api+json',
        '--header', 'content-type: application/vnd.api+json',
        '--data',
        json.dumps({
            "data": {
                "type": "post",
                "attributes": {
                    "content": {
                        "subject": post_content
                    }
                }
            }
        })
    ]
    
    try:
        response = subprocess.check_output(curl_command, universal_newlines=True)
        response_data = json.loads(response)
        return "Post created successfully."
    except subprocess.CalledProcessError as e:
        return f"Error: Unable to create post - {str(e)}"

"""
LLM Functions
"""
        
#Streaming endpoint 
API_URL = "https://api.openai.com/v1/chat/completions" #os.getenv("API_URL") + "/generate_stream"

#Inference function
def predict(openai_gpt4_key, system_msg, api_result, top_p, temperature, chat_counter, chatbot=[], history=[]): 

    headers = {
    "Content-Type": "application/json",
    "Authorization": f"Bearer {openai_gpt4_key}"  #Users will provide their own OPENAI_API_KEY 
    }
    print(f"system message is ^^ {system_msg}")
    if system_msg.strip() == '':
        initial_message = [{"role": "user", "content": f"{inputs}"},]
        multi_turn_message = []
    else:
        initial_message= [{"role": "system", "content": system_msg},
                   {"role": "user", "content": f"{inputs}"},]
        multi_turn_message = [{"role": "system", "content": system_msg},]
        
    if chat_counter == 0 :
        payload = {
        "model": "gpt-4",
        "messages": initial_message , 
        "temperature" : 1.0,
        "top_p":1.0,
        "n" : 1,
        "stream": True,
        "presence_penalty":0,
        "frequency_penalty":0,
        }
        print(f"chat_counter - {chat_counter}")
    else: #if chat_counter != 0 :
        messages=multi_turn_message # Of the type of - [{"role": "system", "content": system_msg},]
        for data in chatbot:
          user = {}
          user["role"] = "user" 
          user["content"] = data[0] 
          assistant = {}
          assistant["role"] = "assistant" 
          assistant["content"] = data[1]
          messages.append(user)
          messages.append(assistant)
        temp = {}
        temp["role"] = "user" 
        temp["content"] = inputs
        messages.append(temp)
        #messages
        payload = {
        "model": "gpt-4",
        "messages": messages, # Of the type of [{"role": "user", "content": f"{inputs}"}],
        "temperature" : temperature, #1.0,
        "top_p": top_p, #1.0,
        "n" : 1,
        "stream": True,
        "presence_penalty":0,
        "frequency_penalty":0,}

    chat_counter+=1

    history.append(inputs)
    print(f"Logging : payload is - {payload}")
    # make a POST request to the API endpoint using the requests.post method, passing in stream=True
    response = requests.post(API_URL, headers=headers, json=payload, stream=True)
    print(f"Logging : response code - {response}")
    token_counter = 0 
    partial_words = "" 

    counter=0
    for chunk in response.iter_lines():
        #Skipping first chunk
        if counter == 0:
          counter+=1
          continue
        # check whether each line is non-empty
        if chunk.decode() :
          chunk = chunk.decode()
          # decode each line as response data is in bytes
          if len(chunk) > 12 and "content" in json.loads(chunk[6:])['choices'][0]['delta']:
              partial_words = partial_words + json.loads(chunk[6:])['choices'][0]["delta"]["content"]
              if token_counter == 0:
                history.append(" " + partial_words)
              else:
                history[-1] = partial_words
              chat = [(history[i], history[i + 1]) for i in range(0, len(history) - 1, 2) ]  # convert to tuples of list
              token_counter+=1
              yield chat, history, chat_counter, response  # resembles {chatbot: chat, state: history}  
                   
#Resetting to blank
def reset_textbox():
    return gr.update(value='')

#to set a component as visible=False
def set_visible_false():
    return gr.update(visible=False)

#to set a component as visible=True
def set_visible_true():
    return gr.update(visible=True)

# Define the Gradio interface
iface = gr.Interface(
    fn=predict,  # Use 'predict' as the function
    inputs=[
        gr.inputs.Textbox(label="OpenAI GPT4 Key", type="password", placeholder="sk.."),
        gr.inputs.Textbox(label="System Message", default=""),
        gr.inputs.Textbox(label="Input Board ID for api_call"),
        gr.inputs.Textbox(label="Output Board ID for create_post"),
    ],
    outputs=gr.outputs.Textbox(label="Summary"),
    live=True,
    title="Padlet API Caller with cURL and LLM",
    description="Enter OpenAI GPT4 key, system message, input board ID for api_call, and output board ID for create_post",
)

# Add event handlers to call 'api_call' and 'create_post' when the "Generate Summary" and "Post Summary" buttons are clicked
iface.inputs[4].submit(api_call, [gr.inputs.Textbox])
iface.inputs[4].click(api_call, [gr.inputs.Textbox])

iface.inputs[5].submit(create_post, [gr.inputs.Textbox, gr.outputs.Textbox])
iface.inputs[5].click(create_post, [gr.inputs.Textbox, gr.outputs.Textbox])

iface.launch()