Spaces:
Runtime error
Runtime error
| from utils.csv_manager import CSVManager | |
| from datetime import datetime | |
| import os | |
| import logging | |
| class DataManager(CSVManager): | |
| def __init__(self, csv_file='data/clientes.csv'): | |
| # Asegurar que el directorio data existe | |
| os.makedirs('data', exist_ok=True) | |
| super().__init__( | |
| csv_file=csv_file, | |
| headers=['Nombre', 'Teléfono', 'Edad', 'Empleo', 'Fecha', 'Modo'] | |
| ) | |
| self.data_collection_state = None | |
| self.customer_data = { | |
| 'nombre': None, | |
| 'telefono': None, | |
| 'edad': None, | |
| 'empleo': None, | |
| 'modo': None | |
| } | |
| # Configurar logging | |
| self.logger = logging.getLogger('data_manager') | |
| self.logger.setLevel(logging.INFO) | |
| handler = logging.FileHandler('logs/data_collection.log') | |
| handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')) | |
| self.logger.addHandler(handler) | |
| def handle_data_collection(self, response, mode, get_data_collection_steps): | |
| """Manejar la recolección de datos según el modo""" | |
| if not response: | |
| return "No pude entender eso. ¿Podría repetirlo?" | |
| try: | |
| current_state = self.data_collection_state | |
| next_prompts = get_data_collection_steps(mode) | |
| self.logger.info(f"Procesando dato para {current_state}: {response}") | |
| # Validar el dato actual | |
| is_valid, value = self.validate_data(current_state, response) | |
| if not is_valid: | |
| self.logger.warning(f"Dato inválido para {current_state}: {response}") | |
| return f"{value}. Por favor, inténtelo de nuevo." | |
| # Guardar el dato validado | |
| self.customer_data[current_state] = value | |
| self.customer_data['modo'] = mode | |
| self.logger.info(f"Dato guardado para {current_state}: {value}") | |
| # Obtener siguiente estado y prompt | |
| next_state, next_prompt = next_prompts[current_state] | |
| if next_state is None: | |
| self.logger.info("Finalizando recolección de datos") | |
| success = self.save_customer_data() | |
| self.data_collection_state = None | |
| return "Error al guardar los datos. Por favor, intente de nuevo." if not success else next_prompt | |
| else: | |
| self.data_collection_state = next_state | |
| self.logger.info(f"Siguiente estado: {next_state}") | |
| return next_prompt | |
| except Exception as e: | |
| self.logger.error(f"Error en handle_data_collection: {e}") | |
| return "Hubo un error procesando su información. ¿Podríamos comenzar de nuevo?" | |
| def save_customer_data(self): | |
| """Guardar datos del cliente en CSV""" | |
| try: | |
| row_data = [ | |
| self.customer_data['nombre'], | |
| self.customer_data['telefono'], | |
| self.customer_data['edad'], | |
| self.customer_data['empleo'], | |
| datetime.now().strftime("%Y-%m-%d %H:%M:%S"), | |
| self.customer_data['modo'] | |
| ] | |
| self.logger.info(f"Guardando datos: {row_data}") | |
| success = self.write_row(row_data) | |
| if success: | |
| self.logger.info("Datos guardados exitosamente") | |
| self.customer_data = {key: None for key in self.customer_data} | |
| else: | |
| self.logger.error("Error al guardar datos en CSV") | |
| return success | |
| except Exception as e: | |
| self.logger.error(f"Error en save_customer_data: {e}") | |
| return False | |
| def start_data_collection(self): | |
| """Iniciar recolección de datos""" | |
| self.data_collection_state = 'nombre' | |
| self.logger.info("Iniciando recolección de datos") | |
| def is_collecting_data(self): | |
| """Verificar si está en modo recolección""" | |
| return self.data_collection_state is not None | |
| def get_current_state(self): | |
| """Obtener estado actual de recolección""" | |
| return self.data_collection_state | |
| def validate_data(self, data_type, value): | |
| """Validar datos según su tipo""" | |
| if not value: | |
| return False, "El valor no puede estar vacío" | |
| try: | |
| if data_type == 'nombre': | |
| # Validar que solo contenga letras y espacios | |
| value = value.strip() | |
| if not all(c.isalpha() or c.isspace() for c in value): | |
| return False, "El nombre solo debe contener letras" | |
| if len(value.split()) < 2: | |
| return False, "Por favor proporcione nombre y apellido" | |
| return True, value.title() # Convertir a formato título | |
| elif data_type == 'telefono': | |
| # Limpiar y validar número telefónico | |
| clean_number = ''.join(filter(str.isdigit, value)) | |
| if len(clean_number) != 10: | |
| return False, "El número debe tener 10 dígitos" | |
| # Formatear número: (123) 456-7890 | |
| formatted_number = f"({clean_number[:3]}) {clean_number[3:6]}-{clean_number[6:]}" | |
| return True, formatted_number | |
| elif data_type == 'edad': | |
| # Validar edad | |
| try: | |
| edad = int(''.join(filter(str.isdigit, value))) | |
| if not (18 <= edad <= 99): | |
| return False, "La edad debe estar entre 18 y 99 años" | |
| return True, str(edad) | |
| except ValueError: | |
| return False, "La edad debe ser un número" | |
| elif data_type == 'empleo': | |
| # Validar empleo | |
| value = value.strip() | |
| if len(value) < 3: | |
| return False, "Por favor proporcione un empleo válido" | |
| # Capitalizar primera letra de cada palabra | |
| return True, value.title() | |
| return True, value | |
| except Exception as e: | |
| self.logger.error(f"Error validando dato {data_type}: {e}") | |
| return False, "Error validando el dato. Por favor, inténtelo de nuevo." |