File size: 5,044 Bytes
c58df45 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 |
from .database_init import get_sql_containers
from datetime import datetime, timezone
import logging
import bcrypt
import uuid
logger = logging.getLogger(__name__)
def get_user(username, role=None):
user_container, _, _ = get_sql_containers()
try:
query = f"SELECT * FROM c WHERE c.id = '{username}'"
if role:
query += f" AND c.role = '{role}'"
items = list(user_container.query_items(query=query, enable_cross_partition_query=True))
return items[0] if items else None
except Exception as e:
logger.error(f"Error al obtener usuario {username}: {str(e)}")
return None
def get_admin_user(username):
return get_user(username, role='Administrador')
def get_student_user(username):
return get_user(username, role='Estudiante')
def get_teacher_user(username):
return get_user(username, role='Profesor')
def create_user(username, password, role, additional_info=None):
user_container, _, _ = get_sql_containers()
try:
hashed_password = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
user_data = {
'id': username,
'password': hashed_password,
'role': role,
'timestamp': datetime.now(timezone.utc).isoformat(),
'additional_info': additional_info or {}
}
user_container.create_item(body=user_data)
logger.info(f"Usuario {role} creado: {username}")
return True
except Exception as e:
logger.error(f"Error al crear usuario {role}: {str(e)}")
return False
def create_student_user(username, password, additional_info=None):
return create_user(username, password, 'Estudiante', additional_info)
def create_teacher_user(username, password, additional_info=None):
return create_user(username, password, 'Profesor', additional_info)
def create_admin_user(username, password, additional_info=None):
return create_user(username, password, 'Administrador', additional_info)
###########################################################
def update_student_user(username, new_info):
user_container, _, _ = get_sql_containers()
try:
user = get_student_user(username)
if user:
user['additional_info'].update(new_info)
user_container.upsert_item(body=user)
logger.info(f"Informaci贸n del estudiante actualizada: {username}")
return True
else:
logger.warning(f"Intento de actualizar estudiante no existente: {username}")
return False
except Exception as e:
logger.error(f"Error al actualizar informaci贸n del estudiante {username}: {str(e)}")
return False
def delete_student_user(username):
user_container, _, _ = get_sql_containers()
try:
user = get_student_user(username)
if user:
user_container.delete_item(item=user['id'], partition_key=username)
logger.info(f"Estudiante eliminado: {username}")
return True
else:
logger.warning(f"Intento de eliminar estudiante no existente: {username}")
return False
except Exception as e:
logger.error(f"Error al eliminar estudiante {username}: {str(e)}")
return False
def store_application_request(name, lastname, email, institution, current_role, desired_role, reason):
_, application_requests_container, _ = get_sql_containers()
try:
application_request = {
"id": str(uuid.uuid4()),
"name": name,
"lastname": lastname,
"email": email,
"institution": institution,
"current_role": current_role,
"desired_role": desired_role,
"reason": reason,
"requestDate": datetime.utcnow().isoformat()
}
application_requests_container.create_item(body=application_request)
logger.info(f"Solicitud de aplicaci贸n almacenada para el email: {email}")
return True
except Exception as e:
logger.error(f"Error al almacenar la solicitud de aplicaci贸n: {str(e)}")
return False
def store_student_feedback(username, name, email, feedback):
_, _, user_feedback_container = get_sql_containers()
try:
feedback_item = {
"id": str(uuid.uuid4()),
"username": username,
"name": name,
"email": email,
"feedback": feedback,
"role": "Estudiante",
'timestamp': datetime.now(timezone.utc).isoformat(),
}
result = user_feedback_container.create_item(body=feedback_item)
logger.info(f"Feedback de estudiante almacenado con ID: {result['id']} para el usuario: {username}")
return True
except Exception as e:
logger.error(f"Error al almacenar el feedback del estudiante {username}: {str(e)}")
return False |