import pandas as pd import os import sys from extraccion import agentes_entidades from extraccion.schemas_entidades import ( Acciones, Temas, Estrategias, Factores, Entidades, Competencias, Herramientas, Mecanismos, Opiniones, Expectativas, ) module_path = os.path.abspath(os.path.join("..")) # Add to sys.path if not already present if module_path not in sys.path: sys.path.append(module_path) print(module_path) INPUT_FOLDER = "input/Preguntas Categoricas" OUTPUT_FOLDER = "output/fase1" FILES_TO_PROCESS = os.listdir(INPUT_FOLDER) DELIMITER = "|^" DIC_QUESTIONS = { "Encuesta_MediaG01Q02.csv": agentes_entidades.extractor_pre_1, # "Encuesta_MediaG01Q04.csv": agentes_entidades.extractor_pre_2, # "Encuesta_MediaG01Q10.csv": agentes_entidades.extractor_pre_3, # "Encuesta_MediaG03Q17.csv": agentes_entidades.extractor_pre_4, # "Encuesta_MediaG03Q18.csv": agentes_entidades.extractor_pre_5, # # "Encuesta_MediaG03Q19.csv": agentes_entidades.extractor_pre_6, # "Encuesta_MediaG04Q22.csv": agentes_entidades.extractor_pre_7, # "Encuesta_MediaG04Q23.csv": agentes_entidades.extractor_pre_8, # "Encuesta_MediaG04Q25.csv": agentes_entidades.extractor_pre_9, # "Encuesta_MediaG05Q30.csv": agentes_entidades.extractor_pre_10, # "Encuesta_MediaG05Q34.csv": agentes_entidades.extractor_pre_11, # "Encuesta_MediaG05Q28.csv": agentes_entidades.extractor_pre_12, # "Encuesta_MediaG06Q35.csv": agentes_entidades.extractor_pre_13, # "Encuesta_MediaG06Q38.csv": agentes_entidades.extractor_pre_14, # "Encuesta_MediaG06Q40.csv": agentes_entidades.extractor_pre_15, } def extract_answers(answers_obj): """ Toma el objeto Pydantic devuelto por un extractor y devuelve una LISTA de los textos de respuesta extraídos. Devuelve: list[str]: Una lista de strings (ej. ["Accion 1", "Accion 2"]) """ list_items_text = [] # Lista para guardar solo los strings finales list_items = [] # Lista genérica de objetos item_attr_name = "" # Atributo a extraer # --- Manejo de esquemas de listas de objetos --- if isinstance(answers_obj, Acciones): list_items = answers_obj.acciones item_attr_name = "accion" elif isinstance(answers_obj, Temas): list_items = answers_obj.temas item_attr_name = "tema" elif isinstance(answers_obj, Estrategias): list_items = answers_obj.estrategias item_attr_name = "estrategia" elif isinstance(answers_obj, Factores): list_items = answers_obj.factores item_attr_name = "factor" elif isinstance(answers_obj, Entidades): list_items = answers_obj.entidades item_attr_name = "entidad" elif isinstance(answers_obj, Herramientas): list_items = answers_obj.herramientas item_attr_name = "herramienta" elif isinstance(answers_obj, Mecanismos): list_items = answers_obj.mecanismos item_attr_name = "mecanismo" elif isinstance(answers_obj, Opiniones): list_items = answers_obj.opiniones item_attr_name = "opinion" elif isinstance(answers_obj, Expectativas): list_items = answers_obj.expectativas item_attr_name = "expectativa" if list_items and item_attr_name: for item in list_items: item_text = getattr(item, item_attr_name, "") if item_text: list_items_text.append(item_text) return list_items_text # Devuelve la lista de textos # --- Manejo del caso especial: Competencias --- elif isinstance(answers_obj, Competencias): if answers_obj.basicas: list_items_text.extend([f"Básica: {c}" for c in answers_obj.basicas if c]) if answers_obj.socioemocionales: list_items_text.extend( [f"Socioemocional: {c}" for c in answers_obj.socioemocionales if c] ) if answers_obj.ciudadanas: list_items_text.extend( [f"Ciudadana: {c}" for c in answers_obj.ciudadanas if c] ) if answers_obj.siglo_xxi: list_items_text.extend( [f"Siglo XXI: {c}" for c in answers_obj.siglo_xxi if c] ) return list_items_text # Devuelve la lista de textos de competencias # Si no se encontró nada, devuelve una lista vacía return [] def format_answer(dataframe, function): """ Itera sobre el dataframe, aplica la función extractora y gestiona el contador de IDs global para ESE dataframe. """ dataframe["respuestas_formato"] = None dataframe["respuestas_formato_id"] = None # Inicializar el contador de ID global para este archivo global_id_counter = 1 for index, row in dataframe.iterrows(): # 1. Obtener el objeto Pydantic answers = function(row["respuesta"]) # 2. Obtener la LISTA de textos extraídos list_items_text = extract_answers(answers) # 3. Si la lista está vacía, poner strings vacíos y continuar if not list_items_text: dataframe.loc[index, "respuestas_formato"] = "" dataframe.loc[index, "respuestas_formato_id"] = "" continue # No incrementa el contador # 4. Si hay textos, formatear la salida count = len(list_items_text) # Formatear el texto answer_formated = DELIMITER.join(list_items_text) + DELIMITER # Formatear los IDs usando el contador global id_list = [str(i) for i in range(global_id_counter, global_id_counter + count)] id_formated = ",".join(id_list) # 5. Asignar los valores al DataFrame dataframe.loc[index, "respuestas_formato"] = answer_formated dataframe.loc[index, "respuestas_formato_id"] = id_formated # 6. Actualizar el contador global para la siguiente fila global_id_counter += count def format_all_answers(Dic_questions): """ Función principal que procesa todos los archivos CSV definidos en Dic_questions. """ for key, value in Dic_questions.items(): try: # Construir la ruta completa al archivo de entrada input_file_path = os.path.join(INPUT_FOLDER, key) # Verificar si el archivo existe if not os.path.exists(input_file_path): print(f"Advertencia: El archivo no existe, omitiendo: {key}") continue question_dataframe = pd.read_csv(input_file_path) if question_dataframe.empty: print(f"Archivo vacío, omitiendo: {key}") continue if "respuesta" not in question_dataframe.columns: print(f"No se encontró la columna 'respuesta' en {key}, omitiendo.") continue # Esta función ahora maneja toda la lógica de conteo format_answer(question_dataframe, value) # Limpiar NAs antes de guardar question_dataframe["respuestas_formato"] = question_dataframe[ "respuestas_formato" ].fillna("") question_dataframe["respuestas_formato_id"] = question_dataframe[ "respuestas_formato_id" ].fillna("") # Construir la ruta completa al archivo de salida output_file_path = os.path.join(OUTPUT_FOLDER, key) # Asegurarse de que el directorio de salida exista os.makedirs(OUTPUT_FOLDER, exist_ok=True) question_dataframe.to_csv(output_file_path, index=False) print(f"Procesado y guardado: {key}") except pd.errors.EmptyDataError: print(f"Error: El archivo {key} está vacío o es inválido.") except Exception as e: print(f"Error inesperado procesando {key}: {e}") # --- Ejecutar el proceso --- format_all_answers(DIC_QUESTIONS) print("Proceso de Fase 1 completado.")