app.py

from dotenv import load_dotenv
import os
import json

from langchain_openai import ChatOpenAI

#from langchain.schema import AIMessage, HumanMessage, SystemMessage
from langchain_core.messages import HumanMessage, AIMessage, SystemMessage, ToolMessage

from langchain_core.tools import tool
import gradio as gr

load_dotenv()

credentials = {}
credentials[os.getenv("USERNAME")] = os.getenv("PASSWORD")
i = 1
while os.getenv(f"USERNAME_{i}") and os.getenv(f"PASSWORD_{i}"):
    credentials[os.getenv(f"USERNAME_{i}")] = os.getenv(f"PASSWORD_{i}")
    i += 1

#print(credentials)

MAX_HISTORY = 20

system_message = """
Sei l'assistente virtuale di Forfè, un software di fatturazione e gestione fiscale 
specifico per il Regime Forfettario, la Gestione Separata INPS e la Cassa Artigiani e Commercianti. 
Il tuo compito è aiutare gli utenti nella gestione della loro attività, fornendo supporto nella creazione 
di fatture, gestione clienti e prodotti, e altre operazioni contabili.

🔹 **Le tue funzionalità principali includono:**
- Creazione di nuovi clienti, sia persone fisiche che società.
- Creazione di nuovi prodotti o servizi da fatturare.
- Generazione di fatture con i dati dei clienti e dei prodotti registrati.

📌 **Regole di interazione:**
- Chiedi all'utente solo i dati strettamente necessari per l'operazione richiesta.
- Mantieni sempre un linguaggio chiaro, professionale e amichevole.
- Se il numero di informazioni da richiedere all'utente sono troppe, dividi l'operazione in più passaggi, chiedendo dapprima i parametri obbligatori e poi quelli opzionali.
- Rispondi in italiano e guida l'utente passo dopo passo nel processo.

🚀 **Obiettivo:** Aiutare i professionisti e le piccole imprese a gestire la loro attività in modo semplice ed efficace con Forfè.
"""

@tool
def create_product(
    name: str,  
    price: float  
) -> str:
    """Crea un nuovo prodotto o servizio e restituisce una conferma.

    Parametri:
    - name (str): Nome del prodotto o del servizio offerto.
    - price (float): Importo in euro (€) del prodotto o servizio.

    Ritorna:
    - str: Un messaggio di conferma con i dettagli del prodotto o servizio creato.
    """

    print("\n--- Nuovo Prodotto/Servizio Creato ---")
    print(f"Nome: {name}")
    print(f"Prezzo: {price:.2f} EUR")

    return f"Il nuovo prodotto o servizio '{name}' con un importo di {price:.2f}€ è stato generato con successo!"

@tool
def create_customer_type(customer_type: str) -> str:
    """Seleziona il tipo di cliente da creare.

    Parametri:
    - customer_type (str): Il tipo di cliente da creare. Deve essere 'individual' (persona fisica) o 'company' (società).

    Ritorna:
    - str: Un messaggio di conferma con il tipo di cliente selezionato.
    """

    if customer_type not in ["individual", "company"]:
        return "Errore: Il tipo di cliente deve essere 'individual' (persona fisica) o 'company' (società)."

    return f"Hai selezionato {customer_type}. Ora puoi procedere con l'inserimento dei dati."

@tool
def create_individual(
    first_name: str, 
    last_name: str, 
    tax_code: str, 
    address: str, 
    street_number: str, 
    postal_code: str, 
    city: str, 
    province: str, 
    country: str, 
    vat_number: str = None,  # Opzionale
    email: str = None, 
    pec: str = None, 
    phone: str = None, 
    recipient_code: str = None
) -> str:
    """Crea un cliente di tipo persona fisica.

    Parametri:
    - first_name (str): Nome del cliente.
    - last_name (str): Cognome del cliente.
    - tax_code (str): Codice fiscale.
    - address (str): Indirizzo di residenza.
    - street_number (str): Numero civico.
    - postal_code (str): CAP.
    - city (str): Città.
    - province (str): Provincia.
    - country (str): Nazione.
    - vat_number (str, opzionale): Partita IVA, se presente.
    - email (str, opzionale): Indirizzo email.
    - pec (str, opzionale): Indirizzo PEC.
    - phone (str, opzionale): Numero di telefono.
    - recipient_code (str, opzionale): Codice destinatario.

    Ritorna:
    - str: Un messaggio di conferma con i dati della persona fisica creata.
    """

    print("\n--- Nuova Persona Fisica Creata ---")
    print(f"Nome: {first_name} {last_name}")
    print(f"Codice Fiscale: {tax_code}")
    print(f"Indirizzo: {address}, {street_number}, {postal_code}, {city}, {province}, {country}")

    if vat_number:
        print(f"Partita IVA: {vat_number}")
    if email:
        print(f"Email: {email}")
    if pec:
        print(f"PEC: {pec}")
    if phone:
        print(f"Telefono: {phone}")
    if recipient_code:
        print(f"Codice Destinatario: {recipient_code}")

    return f"La persona fisica {first_name} {last_name} è stata creata con successo!"

@tool
def create_company_mode(company_type: str) -> str:
    """Permette di scegliere se inserire i dati societari tramite Partiva IVA o Codice Fiscale.

    Parametri:
    - company_type (str): Modalità di inserimento delle informazioni societarie. Deve essere 'vat_number' (Partiva IVA) o 'tax_code' (Codice Fiscale).

    Ritorna:
    - str: Un messaggio di conferma con il tipo di selezionato.
    """

    if customer_type not in ["vat_number", "tax_code"]:
        return "Errore: Il tipo di modalità deve essere 'vat_number' (Partiva IVA) o 'tax_code' (Codice Fiscale)."

    return f"Hai selezionato {company_type}. Ora puoi procedere con l'inserimento dei dati."

@tool
def create_company_tax_code(
    company_name: str, 
    tax_code: str, 
    address: str, 
    street_number: str, 
    postal_code: str, 
    city: str, 
    province: str, 
    country: str, 
    vat_number: str = None,
    email: str = None, 
    pec: str = None, 
    phone: str = None, 
    recipient_code: str = None
) -> str:
    """Crea un cliente di tipo società tramite Codice Fiscale.

    Parametri:
    - company_name (str): Nome della società (ragione sociale).
    - tax_code (str): Codice fiscale della società.
    - address (str): Indirizzo della sede legale.
    - street_number (str): Numero civico.
    - postal_code (str): CAP.
    - city (str): Città.
    - province (str): Provincia.
    - country (str): Nazione.
    - vat_number (str, opzionale): Partita IVA.
    - email (str, opzionale): Indirizzo email.
    - pec (str, opzionale): Indirizzo PEC.
    - phone (str, opzionale): Numero di telefono.
    - recipient_code (str, opzionale): Codice destinatario.

    Ritorna:
    - str: Un messaggio di conferma con i dati della società creata.
    """

    print("\n--- Nuova Società Creata ---")
    print(f"Ragione Sociale: {company_name}")
    print(f"Codice Fiscale: {tax_code}")
    print(f"Indirizzo: {address}, {street_number}, {postal_code}, {city}, {province}, {country}")

    if vat_number:
        print(f"Partita IVA: {vat_number}")
    if email:
        print(f"Email: {email}")
    if pec:
        print(f"PEC: {pec}")
    if phone:
        print(f"Telefono: {phone}")
    if recipient_code:
        print(f"Codice Destinatario: {recipient_code}")

    return f"La società {company_name} è stata creata con successo!"

@tool
def create_company_vat_number(
    vat_number: str = None,
) -> str:
    """Crea un cliente di tipo società tramite Partita IVA, non vanno richiesti altri dati, solo confermare la Partita IVA.

    Parametri:
    - vat_number (str): Partita IVA.

    Ritorna:
    - str: Un messaggio di conferma con i dati della Partita IVA della società creata.
    """

    print("\n--- Nuova Società Creata ---")
    print(f"Partita IVA: {vat_number}")

    return f"La società con Partita IVA {vat_number} è stata creata con successo!"


tools = [
    create_product, 
    create_customer_type, 
    create_individual, 
    create_company_mode, 
    create_company_tax_code, 
    create_company_vat_number
]

llm = ChatOpenAI(
    model="gpt-4o-mini", 
    streaming=True,
)
llm_with_tools = llm.bind_tools(tools)

tool_name = ""
tool_args = ""

tool_mapping = {
    #"create_customer_type": create_customer_type,
    "create_individual": create_individual,
    "create_company_tax_code": create_company_tax_code,
    "create_company_vat_number": create_company_vat_number,
    "create_product": create_product
}

label_buttons = ["Procedi", "Annulla"]

def _check_response(ai_msg):

    if "persona fisica" in ai_msg.lower() and "società" in ai_msg.lower():

        context_analysis_prompt = f"""
        Questa è la risposta dell'LLM a un utente che vuole creare un cliente:

        \"{ai_msg}\"

        Devi solo rispondere con "create_client" se la risposta chiede all'utente di selezionare tra "Persona Fisica" e "Società", 
        oppure "SPIEGAZIONE" se si tratta solo di una descrizione generale dei tipi di clienti senza richiedere un'azione.
        """
        analysis_response = llm.invoke(context_analysis_prompt).content.strip()
        print(analysis_response)

        if analysis_response == "create_client":
            return True, analysis_response

    elif "partita iva" in ai_msg.lower() and "codice fiscale" in ai_msg.lower():

        context_analysis_prompt = f"""
        Questa è la risposta dell'LLM a un utente che deve scegliere la modalità con cui vuole inserire i dati societari

        \"{ai_msg}\"

        Devi solo rispondere con "select_mode" se la risposta chiede all'utente di selezionare tra "Partita IVA" e "Codice Fiscale", 
        oppure "SPIEGAZIONE" se si tratta solo di una descrizione generale senza richiedere un'azione.
        """
        analysis_response = llm.invoke(context_analysis_prompt).content.strip()
        print(analysis_response)

        if analysis_response == "select_mode":
            return True, analysis_response

    return False, ""

def _get_confirmation_message(tool_name: str, tool_args: dict) -> str:
    """Genera un messaggio descrittivo per confermare l'azione che sta per essere eseguita."""
    
    if tool_name == "create_individual":
        return f"**ACTION** Stai per creare un cliente di tipo Persona Fisica: {tool_args.get('first_name', 'N/D')} {tool_args.get('last_name', 'N/D')}."

    elif tool_name == "create_company_tax_code":
        return f"**ACTION** Stai per creare una Società: {tool_args.get('company_name', 'N/D')}."

    elif tool_name == "create_company_vat_number":
        return f"**ACTION** Stai per creare una Società con Partita IVA: {tool_args.get('vat_number', 'N/D')}."

    elif tool_name == "create_product":
        return f"**ACTION** Stai per creare un nuovo prodotto o servizio: '{tool_args.get('name', 'N/D')}' al prezzo di {tool_args.get('price', 0):.2f}€."

    return "**ACTION** Stai per eseguire un'operazione sconosciuta."

def _format_tool_args(tool_name: str, tool_args: dict) -> str:
    """Trasforma i parametri tecnici in una descrizione leggibile per l'utente, gestendo i valori opzionali."""
    
    translations = {
        "first_name": "Nome",
        "last_name": "Cognome",
        "tax_code": "Codice Fiscale",
        "address": "Indirizzo",
        "street_number": "Numero Civico",
        "postal_code": "CAP",
        "city": "Città",
        "province": "Provincia",
        "country": "Nazione",
        "vat_number": "Partita IVA",
        "email": "Email",
        "pec": "PEC",
        "phone": "Telefono",
        "recipient_code": "Codice Destinatario",
        "company_name": "Ragione Sociale",
        "name": "Nome del Prodotto",
        "price": "Prezzo",
        "amount": "Importo"
    }
    
    _formatted_args = "\n".join([
        f"- {translations.get(key, key)}: {value if value else 'Non specificato'}" 
        for key, value in tool_args.items()
    ])
    
    return _formatted_args

def _get_company_info(vat_number: str) -> dict:
    print(f"Getting company info for vat_number: {vat_number}")

    company_info = {
        "company_name": "NewCo",
        "tax_code": "abcdefghilmnopqr",
        "vat_number": vat_number,
        "address": "via Roma",
        "street_number": "12",
        "postal_code": "00100",
        "city": "Roma",
        "province": "RM",
        "country": "IT",
        "email": "[email protected]",
        "pec": "[email protected]",
        "phone": "1234567890",
        "recipient_code": "abcabcabc"
    }

    return company_info

def chatbot_response(message, history):
    
    global tool_name, tool_args

    history = history or []

    if message.startswith("**OPERAZIONE ANNULLATA**"):
        history.append({"role": "user", "content": "**OPERAZIONE ANNULLATA**"})
    else:
        history.append({"role": "user", "content": message})

    messages = [SystemMessage(content=system_message)]
    truncated_history = history[-MAX_HISTORY:] if len(history) > MAX_HISTORY else history

    #print(f"Truncated history: {truncated_history}")

    for entry in truncated_history:
        if entry["role"] == "user":
            messages.append(HumanMessage(content=entry["content"]))
        elif entry["role"] == "assistant":
            messages.append(AIMessage(content=entry["content"]))

    if message is not None:
        messages.append(HumanMessage(content=message))
        
        #_response=llm_with_tools.stream(history_langchain_format)
        ai_msg=llm_with_tools.invoke(messages)
        print(ai_msg)

        choice, analysis_response = _check_response(ai_msg.content)

        if choice and analysis_response == "create_client":
            history.append({"role": "assistant", "content": f"**CREAZIONE CLIENTE**\n{ai_msg.content}"})

        elif choice and analysis_response == "select_mode":
            history.append({"role": "assistant", "content": f"**MODALITA' INSERIMENTO DATI SOCIETARI**\n{ai_msg.content}"})

        elif hasattr(ai_msg, "tool_calls") and ai_msg.tool_calls:
    
            for tool_call in ai_msg.tool_calls:
                tool_name = tool_call["name"].lower()
                tool_args = tool_call["args"]
                selected_tool = tool_mapping.get(tool_name)

                if selected_tool:
                    if selected_tool==create_company_vat_number:
                        tool_args=_get_company_info(tool_args.get('vat_number', 'N/D'))
                        if tool_args is None:
                            history.append({"role": "user", "content": "Errore: Dati societari non trovati."})
                            return history

                    confirmation_message = _get_confirmation_message(tool_name, tool_args)
                    formatted_args = _format_tool_args(tool_name, tool_args)
                    ai_msg= f"{confirmation_message}\n\n**Ecco i dettagli inseriti:**\n{formatted_args}"
                    history.append({"role": "assistant", "content": ai_msg})                            

        else:
            history.append({"role": "assistant", "content": ai_msg.content})

        return history

def reset_textbox():
    """Clears the textbox after sending a message."""
    return gr.update(value="")

def show_buttons(history):
    global label_buttons

    if history and history[-1]["content"].startswith("**ACTION**"):
        label_buttons = ["Procedi", "Annulla"]
        return (
            gr.update(visible=False), 
            gr.update(visible=True, value=label_buttons[0]), 
            gr.update(visible=True, value=label_buttons[1])
        )
    elif history and history[-1]["content"].startswith("**CREAZIONE CLIENTE**"):
        label_buttons = ["Persona Fisica", "Società"]
        return (
            gr.update(visible=False), 
            gr.update(visible=True, value=label_buttons[0]), 
            gr.update(visible=True, value=label_buttons[1])
        )
    elif history and history[-1]["content"].startswith("**MODALITA' INSERIMENTO DATI SOCIETARI**"):
        #print("SONO QUI")
        label_buttons = ["Partita IVA", "Codice Fiscale"]
        return (
            gr.update(visible=False), 
            gr.update(visible=True, value=label_buttons[0]), 
            gr.update(visible=True, value=label_buttons[1])
        )
    return (
        gr.update(visible=True), 
        gr.update(visible=False), 
        gr.update(visible=False)
    )

def hide_buttons():
    """Hides buttons and shows the textbox after clicking a button."""
    return gr.update(visible=True), gr.update(visible=False), gr.update(visible=False)

def show_or_hide_buttons(history):
    global label_buttons

    if history and history[-1]["content"].startswith("**ACTION**"):
        label_buttons = ["Procedi", "Annulla"]
        return (
            gr.update(visible=False), 
            gr.update(visible=True, value=label_buttons[0]), 
            gr.update(visible=True, value=label_buttons[1]),
            gr.update(visible=False)
        )
    elif history and history[-1]["content"].startswith("**CREAZIONE CLIENTE**"):
        label_buttons = ["Persona Fisica", "Società"]
        return (
            gr.update(visible=False), 
            gr.update(visible=True, value=label_buttons[0]), 
            gr.update(visible=True, value=label_buttons[1]),
            gr.update(visible=False)
        )
    elif history and history[-1]["content"].startswith("**MODALITA' INSERIMENTO DATI SOCIETARI**"):
        #print("SONO QUI")
        label_buttons = ["Partita IVA", "Codice Fiscale"]
        return (
            gr.update(visible=False), 
            gr.update(visible=True, value=label_buttons[0]), 
            gr.update(visible=True, value=label_buttons[1]),
            gr.update(visible=False)
        )
    return (
        gr.update(visible=True), 
        gr.update(visible=False), 
        gr.update(visible=False),
        gr.update(visible=True)
    )

def button_clicked(option, history):
    """Handles button clicks and updates chat history."""

    global tool_name, tool_args

    print(f"Button clicked: {option}")

    if (option == "Procedi"):
        history.append({"role": "user", "content": option})
        if tool_name and tool_args:
            selected_tool = tool_mapping.get(tool_name)
            if selected_tool:
                tool_output = selected_tool.invoke(tool_args)
                history.append({"role": "assistant", "content": tool_output})
            else:
                history.append({"role": "user", "content": "Operazione annullata"})
    elif option == "Annulla":
        history.append({"role": "user", "content": option})
        llm_message = (
            "**OPERAZIONE ANNULLATA**\n"
            "Ho annullato l'operazione corrente.\n"
            "Dobbiamo modificare alcuni parametri oppure passare a una nuova operazione.\n"
        )
        history = chatbot_response(llm_message, history)
    elif option == "Persona Fisica":
        llm_message = "Persona Fisica"
        history = chatbot_response(llm_message, history)
    elif option == "Società":
        llm_message = "Società"
        history = chatbot_response(llm_message, history)
    elif option == "Partita IVA":
        llm_message = "Partita IVA"
        history = chatbot_response(llm_message, history)
    elif option == "Codice Fiscale":
        llm_message = "Codice Fiscale"
        history = chatbot_response(llm_message, history)
    
    tool_name = ""
    tool_args = ""
    
    return history

# Authentication function
def authenticate(username, password):
    if username in credentials and credentials[username] == password:
        print("🔑 Login successful!")
        return gr.update(visible=False), gr.update(visible=True), gr.update(value="", visible=False)  # Hide login, show chatbot, clear error
    else:
        print("❌ Incorrect username or password")
        return gr.update(visible=True), gr.update(visible=False), gr.update(value="❌ Incorrect username or password", visible=True)  # Show error


with gr.Blocks() as demo:

    # 🔒 Login Section (Initially Visible)
    with gr.Column(visible=True) as login_section:
        gr.Markdown("### 🔒 Login Required")
        username_input = gr.Textbox(label="Username")
        password_input = gr.Textbox(label="Password", type="password")
        login_button = gr.Button("Login")
        error_message = gr.Text("", visible=False)

    # 🧠 Chatbot Section (Initially Hidden)
    with gr.Column(visible=False) as chatbot_section:

        chatbot = gr.Chatbot(
            label="Assistente Forfè",
            type="messages"
        )  

        user_input = gr.Textbox(label="Utente",placeholder="Cosa vuoi chiedere al tuo assistente Forfè?")

        with gr.Row():
            btn1 = gr.Button("", visible=False)
            btn2 = gr.Button("", visible=False)

        send_btn = gr.Button("Invia")

        # When user submits text
        user_input.submit(chatbot_response, [user_input, chatbot], chatbot) \
                .then(reset_textbox, None, user_input) \
                .then(show_or_hide_buttons, chatbot, [user_input, btn1, btn2, send_btn])
                #.then(show_buttons, chatbot, [user_input, btn1, btn2]) 

        # When user clicks send
        send_btn.click(chatbot_response, [user_input, chatbot], chatbot) \
                .then(reset_textbox, None, user_input) \
                .then(show_or_hide_buttons, chatbot, [user_input, btn1, btn2, send_btn])
                #.then(show_buttons, chatbot, [user_input, btn1, btn2]) 

        # Button clicks: Show textbox, hide buttons
        btn1.click(lambda h: button_clicked(label_buttons[0], h), chatbot, chatbot) \
            .then(show_or_hide_buttons, chatbot, [user_input, btn1, btn2, send_btn])
            #.then(hide_buttons, None, [user_input, btn1, btn2])
        
        btn2.click(lambda h: button_clicked(label_buttons[1], h), chatbot, chatbot) \
            .then(show_or_hide_buttons, chatbot, [user_input, btn1, btn2, send_btn])
            #.then(hide_buttons, None, [user_input, btn1, btn2])

    # 🔑 Login Button Action (Now updates visibility correctly)
    login_button.click(
        authenticate, 
        [username_input, password_input], 
        [login_section, chatbot_section, error_message]
    )



demo.launch(
    debug=True, 
    #share=True
)