Spaces:

efaraneda
/

olivia

Sleeping

olivia / app.py

Update app.py

e7c673d verified over 2 years ago

6.41 kB

import os
import boto3
import ast
import traceback
import time

a = os.environ.get("AWS_ACCESS_KEY_ID")
b = os.environ.get("AWS_SECRET_ACCESS_KEY")

def dynamo(username, message):
    try:
        # Create a DynamoDB client
        dynamodb = boto3.client('dynamodb',aws_access_key_id=b,
                                    aws_secret_access_key=a,
                                    region_name='us-east-1'
                                    )

        # Specify the table name
        table_name = 'PROD_historial_olivia'
    
        # Define the key you want to filter by
        key = 'usuario'
        value = username
    
        # Create the key condition expression as a string
        key_condition_expression = f"{key} = :val"

        # Define the expression attribute values
        expression_attribute_values = {
            ":val": {"S": value}  # Assuming 'phone' is of string type
        }

        # Query the table
        response = dynamodb.query(
            TableName=table_name,
            KeyConditionExpression=key_condition_expression,
            ExpressionAttributeValues=expression_attribute_values
        )
    
        # Retrieve the filtered items
        items = response['Items']
        
        sorted_items = sorted(items, key=lambda x: x['fecha']['S'])
        
        mensajes = []
        if len(sorted_items) > 0:
                if message == ast.literal_eval(sorted_items[-1]['interaccion']['S'])['Estudiante']:
                    return ast.literal_eval(sorted_items[-1]['interaccion']['S'])['Olivia']
                else:
                    return 'esperar'
        else:
            return 'esperar'               
        
    except Exception as e:
        print(f"An error occurred: {str(e)}")
        # Handle the error as needed
        # You can raise the error again, log it, or return a specific error response
        traceback_info = traceback.format_exc()
        print(traceback_info)
        return {'statusCode': 500, 'body': 'An error occurred during item query.'}

def activar_olivia(username, current_message):
        sns_client = boto3.client('sns',
                                aws_access_key_id=b,
                                aws_secret_access_key=a,
                                region_name='us-east-1'
                                )

        topic_arn = os.environ.get("ARN_OLIVIA")

        attributes = {
            'username': {'DataType': 'String', 'StringValue': username},
            'message': {'DataType': 'String', 'StringValue': current_message}}

        r = sns_client.publish(
                            TopicArn=topic_arn,
                            Message='huggingface',
                            MessageAttributes=attributes
                        )   




def lookup_users(username, password):
    print(username)
    try:
        dynamodb = boto3.client('dynamodb',aws_access_key_id=b,
                                    aws_secret_access_key=a,
                                    region_name='us-east-1'
                                    )

        # Specify the table name
        table_name = 'PROD_usuario_olivia'
    
        # Query the table
        response = dynamodb.scan(
            TableName=table_name
        )
    
        # Retrieve the filtered items
        items = response['Items']
        if len(items)>0:
            for i in items:
                if username == i['usuario']['S'] and password == i['contraseña']['S']:
                    os.environ['user'] = username
                    return True
    
        return False
                
    except Exception as e:
        print(f"An error occurred: {str(e)}")
        return False

def greet(name, password):
    r=lookup_users(name, password)
    if r == False:
        mensaje = f"""<div style="display: flex; align-items: center;"><img src="Red.png" width="10" height="10">&nbsp; No Conectado: &nbsp;<b>No se encontró el usuario</b></div>"""
    elif r == True:
        mensaje = f"""<div style="display: flex; align-items: center;"><img src="Green.png" width="10" height="10">&nbsp; Conectado como: &nbsp;<b>{name}</b></div>"""
    return mensaje

def gradio_interface():
    import gradio as gr
    
    with gr.Blocks(theme=gr.themes.Soft()) as demo:

        with gr.Row():
            username = gr.Textbox(scale=3,label="Usuario",info='Ingresa tu nombre de usuario')

            password = gr.Textbox(scale=3,label="Contraseña",type="password",info='Ingresa tu contraseña')
            greet_btn = gr.Button("Conectar",scale=1,interactive=True)
        output = gr.Markdown()
        greet_btn.click(fn=greet, inputs=[username,password], outputs=output, api_name="greet")
        print("username:",username)        
        chatbot = gr.Chatbot(avatar_images=["avatar.jpg", "olivia.jpeg"],
                             label="Esta conversación es anónima y personal. Habla con Olivia para ayudar a mejorar a tu colegio y curso ❤️",
                             bubble_full_width=False
                             )
        msg = gr.Textbox(label='Escribe tu respuesta y pulsa "Enter"',interactive=True, info="Nota: Tu respuesta puede ser tan larga como tu quieras")
    
        def respond(current_message, chat_history):
            username = os.getenv('user')
            activar_olivia(username, current_message)

            t = 0
            while t < 100:
                respuesta = dynamo(username, current_message)
                print('RESPUESTA:', str(respuesta))
                
                if respuesta != 'esperar':
                    break
                else:
                    time.sleep(0.5)
                    t+=0.5
            
            if '.' in respuesta:
                b=respuesta.split('.')
                chat_history.append(("<b>Estudiante:</b> \n"+current_message, "<b>Olivia:</b> \n"+b[0]))
                chat_history.append((None,"<b>Olivia:</b> \n"+b[1]))
            else:
                chat_history.append(("<b>Estudiante:</b> \n"+current_message, "<b>Olivia:</b> \n"+respuesta))
        
            # operaciones para guardar la interaccion
    
            return "", chat_history

        msg.submit(respond, [msg, chatbot], [msg, chatbot])
        demo.queue(max_size=90)
        demo.launch(max_threads=60)     
    


if __name__ == "__main__":
    # Start the HTTP server in a separate thread
    gradio_interface()

	import os
	import boto3
	import ast
	import traceback
	import time

	a = os.environ.get("AWS_ACCESS_KEY_ID")
	b = os.environ.get("AWS_SECRET_ACCESS_KEY")

	def dynamo(username, message):
	try:
	# Create a DynamoDB client
	dynamodb = boto3.client('dynamodb',aws_access_key_id=b,
	aws_secret_access_key=a,
	region_name='us-east-1'
	)

	# Specify the table name
	table_name = 'PROD_historial_olivia'

	# Define the key you want to filter by
	key = 'usuario'
	value = username

	# Create the key condition expression as a string
	key_condition_expression = f"{key} = :val"

	# Define the expression attribute values
	expression_attribute_values = {
	":val": {"S": value} # Assuming 'phone' is of string type
	}

	# Query the table
	response = dynamodb.query(
	TableName=table_name,
	KeyConditionExpression=key_condition_expression,
	ExpressionAttributeValues=expression_attribute_values
	)

	# Retrieve the filtered items
	items = response['Items']

	sorted_items = sorted(items, key=lambda x: x['fecha']['S'])

	mensajes = []
	if len(sorted_items) > 0:
	if message == ast.literal_eval(sorted_items[-1]['interaccion']['S'])['Estudiante']:
	return ast.literal_eval(sorted_items[-1]['interaccion']['S'])['Olivia']
	else:
	return 'esperar'
	else:
	return 'esperar'

	except Exception as e:
	print(f"An error occurred: {str(e)}")
	# Handle the error as needed
	# You can raise the error again, log it, or return a specific error response
	traceback_info = traceback.format_exc()
	print(traceback_info)
	return {'statusCode': 500, 'body': 'An error occurred during item query.'}

	def activar_olivia(username, current_message):
	sns_client = boto3.client('sns',
	aws_access_key_id=b,
	aws_secret_access_key=a,
	region_name='us-east-1'
	)

	topic_arn = os.environ.get("ARN_OLIVIA")

	attributes = {
	'username': {'DataType': 'String', 'StringValue': username},
	'message': {'DataType': 'String', 'StringValue': current_message}}

	r = sns_client.publish(
	TopicArn=topic_arn,
	Message='huggingface',
	MessageAttributes=attributes
	)




	def lookup_users(username, password):
	print(username)
	try:
	dynamodb = boto3.client('dynamodb',aws_access_key_id=b,
	aws_secret_access_key=a,
	region_name='us-east-1'
	)

	# Specify the table name
	table_name = 'PROD_usuario_olivia'

	# Query the table
	response = dynamodb.scan(
	TableName=table_name
	)

	# Retrieve the filtered items
	items = response['Items']
	if len(items)>0:
	for i in items:
	if username == i['usuario']['S'] and password == i['contraseña']['S']:
	os.environ['user'] = username
	return True

	return False

	except Exception as e:
	print(f"An error occurred: {str(e)}")
	return False

	def greet(name, password):
	r=lookup_users(name, password)
	if r == False:
	mensaje = f"""<div style="display: flex; align-items: center;"><img src="Red.png" width="10" height="10">  No Conectado:  <b>No se encontró el usuario</b></div>"""
	elif r == True:
	mensaje = f"""<div style="display: flex; align-items: center;"><img src="Green.png" width="10" height="10">  Conectado como:  <b>{name}</b></div>"""
	return mensaje

	def gradio_interface():
	import gradio as gr

	with gr.Blocks(theme=gr.themes.Soft()) as demo:

	with gr.Row():
	username = gr.Textbox(scale=3,label="Usuario",info='Ingresa tu nombre de usuario')

	password = gr.Textbox(scale=3,label="Contraseña",type="password",info='Ingresa tu contraseña')
	greet_btn = gr.Button("Conectar",scale=1,interactive=True)
	output = gr.Markdown()
	greet_btn.click(fn=greet, inputs=[username,password], outputs=output, api_name="greet")
	print("username:",username)
	chatbot = gr.Chatbot(avatar_images=["avatar.jpg", "olivia.jpeg"],
	label="Esta conversación es anónima y personal. Habla con Olivia para ayudar a mejorar a tu colegio y curso ❤️",
	bubble_full_width=False
	)
	msg = gr.Textbox(label='Escribe tu respuesta y pulsa "Enter"',interactive=True, info="Nota: Tu respuesta puede ser tan larga como tu quieras")

	def respond(current_message, chat_history):
	username = os.getenv('user')
	activar_olivia(username, current_message)

	t = 0
	while t < 100:
	respuesta = dynamo(username, current_message)
	print('RESPUESTA:', str(respuesta))

	if respuesta != 'esperar':
	break
	else:
	time.sleep(0.5)
	t+=0.5

	if '.' in respuesta:
	b=respuesta.split('.')
	chat_history.append(("<b>Estudiante:</b> \n"+current_message, "<b>Olivia:</b> \n"+b[0]))
	chat_history.append((None,"<b>Olivia:</b> \n"+b[1]))
	else:
	chat_history.append(("<b>Estudiante:</b> \n"+current_message, "<b>Olivia:</b> \n"+respuesta))

	# operaciones para guardar la interaccion

	return "", chat_history

	msg.submit(respond, [msg, chatbot], [msg, chatbot])
	demo.queue(max_size=90)
	demo.launch(max_threads=60)



	if __name__ == "__main__":
	# Start the HTTP server in a separate thread
	gradio_interface()