app.py

import random
from enum import Enum

import gradio as gr
from dotenv import load_dotenv

from client import AppClient

load_dotenv(verbose=True)


class ErrorMessage(Enum):
    unauthorized = "Unfortunately, you do not have access to the project. Please reach out to the admin to get access"
    not_signed_in = "Sign in with your hugging face account to get started!"
    agent_error = "error in agent processing: {error}"
    agent_error_unknown = "error in agent processing, contact admin for support"


def respond_dummy():
    return random.choice(["How are you?", "I love you", "I'm very hungry"])


def make_visible():
    return gr.update(visible=True)


def make_invisible():
    return gr.update(visible=False)


def error_response(error_message: ErrorMessage):
    return {
        error_bar: gr.update(value=error_message.value, visible=True),
        start_button: make_invisible(),
        logged_in_components_group: make_invisible()
    }


with gr.Blocks() as demo:
    ################################
    # Define the UI
    ################################

    chat_id = gr.State(None)
    interview_options = gr.State(None)
    interview_selected_index = gr.State(None)

    with gr.Row():
        login_button = gr.LoginButton()
        logout_button = gr.LogoutButton()

    start_button = gr.Button(value="Let's get started!!", visible=True)
    error_bar = gr.Textbox(visible=False)

    logged_in_components_group = gr.Group(visible=False)
    with logged_in_components_group:

        interview_selector_group = gr.Group(visible=True)
        chat_interface_group = gr.Group(visible=False)

        # make dropdown for selecting interview
        with interview_selector_group:
            # Note: DO NOT try to use type="index" as there's a gradio bug which will break the interface
            interview_dropdown = gr.Dropdown(type="value", label="Select the interview", show_label=True, interactive=True)
            interview_selection_submit = gr.Button(value="Start!")

        with chat_interface_group:
            chatbot = gr.Chatbot()
            user_input = gr.Textbox(label="Candidate response")

    ################################
    # Setup user response logic
    ################################

    def user(user_message, history):
        return "", history + [[user_message, None]]

    def bot(chat_id_val, history):
        user_message = history[-1][0]
        # bot_message = respond_dummy()
        response = AppClient.get_agent_response(user_message=user_message, chat_id=chat_id_val)
        if response.agent_message is not None:
            bot_message = response.agent_message
            history[-1][1] = bot_message
            if response.end_interview:
                return {user_input: make_invisible(), chatbot: history}
            return {chatbot: history}
        if response.error_message is not None:
            return {
                error_bar: gr.update(value=ErrorMessage.agent_error.value.format(error=response.error_message), visible=True),
                user_input: gr.update(visible=False)
            }

    user_input.submit(user, [user_input, chatbot], [user_input, chatbot], queue=False).then(
        bot, [chat_id, chatbot], [user_input, chatbot, error_bar]
    )

    ################################
    # Define the auth / interview control logic
    ################################

    # login_button.click(lambda: start_button.update(visible=True), None, [start_button])
    logout_button.click(lambda: logged_in_components_group.update(visible=False),
                        None, [logged_in_components_group])

    # initialize the dropdown for interview selection
    def get_chat_options(profile: gr.OAuthProfile | None):
        # grant access case
        if profile is not None:
            hf_id = profile["preferred_username"]
            available_chats = AppClient.get_chats(user_hf_id=hf_id)
            chats_dict = {f"{r.title} (by {r.creator_name})": r.id for r in available_chats}
            if available_chats is not None:
                print(f"granting access to user {hf_id} for interviews: {available_chats}")
                return {
                    logged_in_components_group: make_visible(),
                    interview_selector_group: make_visible(),
                    start_button: make_invisible(),
                    interview_dropdown: gr.Dropdown.update(choices=list(chats_dict.keys())),
                    interview_options: chats_dict
                }
            else:
                error = ErrorMessage.unauthorized
        else:
            error = ErrorMessage.not_signed_in
        # case: access not granted
        return error_response(error)


    start_button.click(
        get_chat_options, None,
        [logged_in_components_group, error_bar, start_button, interview_selector_group, interview_dropdown, interview_options])

    # process dropdown selection and start interview
    def start_interview(profile: gr.OAuthProfile, dropdown_text, options: dict[str, str]):
        if profile is None:
            return error_response(ErrorMessage.unauthorized)
        # initialize chat
        hf_id = profile["preferred_username"]
        response = AppClient.start_chat(user_hf_id=hf_id, interview_id=options[dropdown_text])
        if response is None:
            return error_response(ErrorMessage.agent_error_unknown)

        return {
            chat_id: response.chat_id,
            chatbot: gr.update(value=[(None, response.first_message)]),
            logged_in_components_group: make_visible(),
            start_button: make_invisible(),
            interview_selector_group: make_invisible(),
            chat_interface_group: make_visible()
        }


    interview_selection_submit.click(
        start_interview,
        [interview_dropdown, interview_options],
        [chat_id, chatbot, interview_selector_group, chat_interface_group, logged_in_components_group, error_bar, start_button])

# demo.queue()
demo.launch()