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, } # 2. Refactorizar 'extract_answers' para manejar todos los tipos de esquemas def extract_answers(answers_obj): """ Toma el objeto Pydantic devuelto por un extractor y lo formatea en un único string delimitado por |^. """ answer_formated = "" list_items = [] # Lista genérica de items (ej. [Accion, Accion]) item_attr_name = "" # Atributo a extraer de cada item (ej. "accion") # --- Manejo de todos los esquemas que son 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" # Si se encontró un tipo de lista estándar, procesarla if list_items and item_attr_name: for item in list_items: # Obtener el texto (ej. item.accion, item.tema) de forma segura item_text = getattr(item, item_attr_name, "") if item_text: answer_formated += f"{item_text}{DELIMITER}" return answer_formated # --- Manejo del caso especial: Competencias --- if isinstance(answers_obj, Competencias): comp_texts = [] # El esquema de Competencias usa listas de strings, no listas de objetos if answers_obj.basicas: comp_texts.extend([f"Básica: {c}" for c in answers_obj.basicas]) if answers_obj.socioemocionales: comp_texts.extend( [f"Socioemocional: {c}" for c in answers_obj.socioemocionales] ) if answers_obj.ciudadanas: comp_texts.extend([f"Ciudadana: {c}" for c in answers_obj.ciudadanas]) if answers_obj.siglo_xxi: comp_texts.extend([f"Siglo XXI: {c}" for c in answers_obj.siglo_xxi]) if not comp_texts: return "" # Devuelve vacío si el objeto Competencias está vacío # Unir todos los textos de competencias answer_formated = DELIMITER.join(comp_texts) + DELIMITER return answer_formated # Si el tipo no coincide con nada (o era una lista vacía), devuelve un string vacío return answer_formated def format_answer(dataframe, function): dataframe["respuestas_formato"] = None for index, row in dataframe.iterrows(): answers = function(row["respuesta"]) answer_to_insert = extract_answers(answers) dataframe.loc[index, "respuestas_formato"] = answer_to_insert def format_all_answers(Dic_questions): for key, value in Dic_questions.items(): question_dataframe = pd.read_csv(INPUT_FOLDER + "/" + key) format_answer(question_dataframe, value) question_dataframe.to_csv(OUTPUT_FOLDER + "/" + key) format_all_answers(DIC_QUESTIONS)