src/procesador_de_cvs_con_llm.py

import os
import pandas as pd
import json
import textwrap
from scipy import spatial
from datetime import datetime
from openai import OpenAI

class ProcesadorCV:

    def __init__(self, api_key, cv_text, job_text, ner_pre_prompt, system_prompt, user_prompt, ner_schema, response_schema,
                inference_model="gpt-4o-mini", embeddings_model="text-embedding-3-small"):
        """
        Inicializa una instancia de la clase con los parámetros proporcionados.

        Args:
            api_key (str): La clave de API para autenticar con el cliente OpenAI.
            cv_text (str): contenido del CV en formato de texto.
            job_text (str): título de la oferta de trabajo a evaluar.
            ner_pre_prompt (str): instrucción de "reconocimiento de entidades nombradas" (NER) para el modelo en lenguaje natural.
            system_prompt (str): instrucción en lenguaje natural para la salida estructurada final.
            user_prompt (str): instrucción con los parámetros y datos calculados en el preprocesamiento.
            ner_schema (dict): esquema para la llamada con "structured outputs" al modelo de OpenAI para NER.
            response_schema (dict): esquema para la respuesta final de la aplicación.
            inference_model (str, opcional): El modelo de inferencia a utilizar. Por defecto es "gpt-4o-mini".
            embeddings_model (str, opcional): El modelo de embeddings a utilizar. Por defecto es "text-embedding-3-small".

        Atributos:
            inference_model (str): Almacena el modelo de inferencia seleccionado.
            embeddings_model (str): Almacena el modelo de embeddings seleccionado.
            client (OpenAI): Instancia del cliente OpenAI inicializada con la clave de API proporcionada.
            cv (str): Almacena el texto del currículum vitae proporcionado.

        """
        self.inference_model = inference_model
        self.embeddings_model = embeddings_model
        self.ner_pre_prompt = ner_pre_prompt
        self.user_prompt = user_prompt
        self.system_prompt = system_prompt
        self.ner_schema = ner_schema
        self.response_schema = response_schema
        self.client = OpenAI(api_key=api_key)
        self.cv = cv_text
        self.job_text = job_text
        print("Cliente inicializado como",self.client)

    def extraer_datos_cv(self, temperature=0.5):
        """
        Extrae datos estructurados de un CV con OpenAI API.
        Args:
            pre_prompt (str): instrucción para el modelo en lenguaje natural.
            schema (dict): esquema de los parámetros que se espera extraer del CV.
            temperature (float, optional): valor de temperatura para el modelo de lenguaje. Por defecto es 0.5.
        Returns:
            pd.DataFrame: DataFrame con los datos estructurados extraídos del CV.
        Raises:
            ValueError: si no se pueden extraer datos estructurados del CV.
        """
        response = self.client.chat.completions.create(
            model=self.inference_model,
            temperature=temperature,
            messages=[
                {"role": "system", "content": self.ner_pre_prompt},
                {"role": "user", "content": self.cv}
            ],
            functions=[
                {
                    "name": "extraer_datos_cv",
                    "description": "Extrae tabla con títulos de puesto de trabajo, nombres de empresa y períodos de un CV.",
                    "parameters": self.ner_schema
                }
            ],
            function_call="auto"
        )

        if response.choices[0].message.function_call:
            function_call = response.choices[0].message.function_call
            structured_output = json.loads(function_call.arguments)
            if structured_output.get("experiencia"):
                df_cv = pd.DataFrame(structured_output["experiencia"]) 
                return df_cv
            else:
                raise ValueError(f"No se han podido extraer datos estructurados: {response.choices[0].message.content}")
        else:
            raise ValueError(f"No se han podido extraer datos estructurados: {response.choices[0].message.content}")
        

    def procesar_periodos(self, df):    
        """
        Procesa los períodos en un DataFrame y añade columnas con las fechas de inicio, fin y duración en meses. 
        Si no hay fecha de fin, se considera la fecha actual.
        Args:
            df (pandas.DataFrame): DataFrame que contiene una columna 'periodo' con períodos en formato 'YYYYMM-YYYYMM' o 'YYYYMM'.
        Returns:
            pandas.DataFrame: DataFrame con columnas adicionales 'fec_inicio', 'fec_final' y 'duracion'.
                - 'fec_inicio' (datetime.date): Fecha de inicio del período.
                - 'fec_final' (datetime.date): Fecha de fin del período.
                - 'duracion' (int): Duración del período en meses.
        """
        # Función lambda para procesar el período
        def split_periodo(periodo):
            dates = periodo.split('-')
            start_date = datetime.strptime(dates[0], "%Y%m")
            if len(dates) > 1:
                end_date = datetime.strptime(dates[1], "%Y%m")
            else:
                end_date = datetime.now()
            return start_date, end_date

        df[['fec_inicio', 'fec_final']] = df['periodo'].apply(lambda x: pd.Series(split_periodo(x)))

        # Formateamos las fechas para mostrar mes, año, y el primer día del mes (dado que el día es irrelevante y no se suele especificar)
        df['fec_inicio'] = df['fec_inicio'].dt.date
        df['fec_final'] = df['fec_final'].dt.date

        # Añadimos una columna con la duración en meses
        df['duracion'] = df.apply(
            lambda row: (row['fec_final'].year - row['fec_inicio'].year) * 12 + 
                        row['fec_final'].month - row['fec_inicio'].month, 
            axis=1
        )

        return df


    def calcular_embeddings(self, df, column='puesto', model_name='text-embedding-3-small'):
        """
        Calcula los embeddings de una columna de un dataframe con OpenAI API.
        Args:
            cv_df (pandas.DataFrame): DataFrame con los datos de los CV.
            column (str, optional): Nombre de la columna que contiene los datos a convertir en embeddings. Por defecto es 'puesto'.
            model_name (str, optional): Nombre del modelo de embeddings. Por defecto es 'text-embedding-3-small'.
        """
        df['embeddings'] = df[column].apply(
            lambda puesto: self.client.embeddings.create(
                input=puesto, 
                model=model_name
            ).data[0].embedding
        )
        return df


    def calcular_distancias(self, df, column='embeddings', model_name='text-embedding-3-small'):
        """
        Calcula la distancia coseno entre los embeddings del texto y los incluidos en una columna del dataframe.
        Params:
        df (pandas.DataFrame): DataFrame que contiene los embeddings.
        column (str, optional): nombre de la columna del DataFrame que contiene los embeddings. Por defecto, 'embeddings'.
        model_name (str, optional): modelo de embeddings de la API de OpenAI. Por defecto "text-embedding-3-small".
        Returns:
        pandas.DataFrame: DataFrame ordenado de menor a mayor distancia, con las distancias en una nueva columna.
        """
        response = self.client.embeddings.create(
            input=self.job_text,
            model=model_name
        )
        emb_compare = response.data[0].embedding

        df['distancia'] = df[column].apply(lambda emb: spatial.distance.cosine(emb, emb_compare))
        df.drop(columns=[column], inplace=True)
        df.sort_values(by='distancia', ascending=True, inplace=True)
        return df


    def calcular_puntuacion(self, df, req_experience, positions_cap=4, dist_threshold_low=0.6, dist_threshold_high=0.7):
        """
        Calcula la puntuación de un CV a partir de su tabla de distancias (con respecto a un puesto dado) y duraciones. 

        Params:
        df (pandas.DataFrame): datos de un CV incluyendo diferentes experiencias incluyendo duracies y distancia previamente calculadas sobre los embeddings de un puesto de trabajo
        req_experience (float): experiencia requerida en meses para el puesto de trabajo (valor de referencia para calcular una puntuación entre 0 y 100 en base a diferentes experiencias)
        positions_cap (int, optional): Maximum number of positions to consider for scoring. Defaults to 4.
        dist_threshold_low (float, optional): Distancia entre embeddings a partir de la cual el puesto del CV se considera "equivalente" al de la oferta.
        max_dist_threshold (float, optional): Distancia entre embeddings a partir de la cual el puesto del CV no puntúa.
        
        Returns:
        pandas.DataFrame: DataFrame original añadiendo una columna con las puntuaciones individuales contribuidas por cada puesto.
        float: Puntuación total entre 0 y 100.
        """
        # A efectos de puntuación, computamos para cada puesto como máximo el número total de meses de experiencia requeridos
        df['duration_capped'] = df['duracion'].apply(lambda x: min(x, req_experience))
        # Normalizamos la distancia entre 0 y 1, siendo 0 la distancia mínima y 1 la máxima
        df['adjusted_distance'] = df['distancia'].apply(
            lambda x: 0 if x <= dist_threshold_low else (
                1 if x >= dist_threshold_high else (x - dist_threshold_low) / (dist_threshold_high - dist_threshold_low)
            )
        )
        # Cada puesto puntúa en base a su duración y a la inversa de la distancia (a menor distancia, mayor puntuación)
        df['position_score'] = round(((1 - df['adjusted_distance']) * (df['duration_capped']/req_experience) * 100), 2)
        # Descartamos puestos con distancia superior al umbral definido (asignamos puntuación 0), y ordenamos por puntuación
        df.loc[df['distancia'] >= dist_threshold_high, 'position_score'] = 0
        df = df.sort_values(by='position_score', ascending=False)
        # Nos quedamos con los puestos con mayor puntuación (positions_cap)
        df.iloc[positions_cap:, df.columns.get_loc('position_score')] = 0
        # Totalizamos (no debería superar 100 nunca, pero ponemos un límite para asegurar) y redondeamos a dos decimales
        total_score = round(min(df['position_score'].sum(), 100), 2)
        return df, total_score
    
    def filtra_experiencia_relevante(self, df):
        """
        Filtra las experiencias relevantes del dataframe y las devuelve en formato diccionario.
        Args:
            df (pandas.DataFrame): DataFrame con la información completa de experiencia.
        Returns:
            dict: Diccionario con las experiencias relevantes.
        """
        df_experiencia =  df[df['position_score'] > 0].copy()
        df_experiencia.drop(columns=['periodo', 'fec_inicio', 'fec_final', 
                                     'distancia', 'duration_capped', 'adjusted_distance'], inplace=True)
        experiencia_dict = df_experiencia.to_dict(orient='list')
        return experiencia_dict
    
    def llamada_final(self, req_experience, puntuacion, dict_experiencia):
        """
        Realiza la llamada final al modelo de lenguaje para generar la respuesta final.
        Args:
        req_experience (int): Experiencia requerida en meses para el puesto de trabajo.
        puntuacion (float): Puntuación total del CV.
        dict_experiencia (dict): Diccionario con las experiencias relevantes.
        Returns:
        dict: Diccionario con la respuesta final.
        """
        messages = [
            {
                "role": "system",
                "content": self.system_prompt
            },
            {
                "role": "user",
                "content": self.user_prompt.format(job=self.job_text, req_experience=req_experience,puntuacion=puntuacion, exp=dict_experiencia)
            }
        ]

        functions = [
            {
                "name": "respuesta_formateada",
                "description": "Devuelve el objeto con puntuacion, experiencia y descripcion de la experiencia",
                "parameters": self.response_schema
            }
        ]

        response = self.client.chat.completions.create(
            model=self.inference_model,
            temperature=0.5,
            messages=messages,
            functions=functions,
            function_call={"name": "respuesta_formateada"}
        )

        if response.choices[0].message.function_call:
            function_call = response.choices[0].message.function_call
            structured_output = json.loads(function_call.arguments)
            print("Respuesta:\n", json.dumps(structured_output, indent=4, ensure_ascii=False))
            wrapped_description = textwrap.fill(structured_output['descripcion de la experiencia'], width=120)
            print(f"Descripción de la experiencia:\n{wrapped_description}")
            return structured_output
        else:
            raise ValueError(f"Error. No se ha podido generar respuesta:\n {response.choices[0].message.content}")
    
    def procesar_cv_completo(self, req_experience, positions_cap, dist_threshold_low, dist_threshold_high):
        """
        Procesa un CV y calcula la puntuación final.
        Args:
            req_experience (int, optional): Experiencia requerida en meses para el puesto de trabajo.
            positions_cap (int, optional): Número máximo de puestos a considerar para la puntuación.
            dist_threshold_low (float, optional): Distancia límite para considerar un puesto equivalente.
            dist_threshold_high (float, optional): Distancia límite para considerar un puesto no relevante.
        Returns:
            pd.DataFrame: DataFrame con las puntuaciones individuales contribuidas por cada puesto.
            float: Puntuación total entre 0 y 100.
        """
        df_datos_estructurados_cv = self.extraer_datos_cv()
        df_datos_estructurados_cv = self.procesar_periodos(df_datos_estructurados_cv)
        df_con_embeddings = self.calcular_embeddings(df_datos_estructurados_cv)
        df_con_distancias = self.calcular_distancias(df_con_embeddings)
        df_puntuaciones, puntuacion = self.calcular_puntuacion(df_con_distancias,
                                                                req_experience=req_experience,
                                                                positions_cap=positions_cap,
                                                                dist_threshold_low=dist_threshold_low,
                                                                dist_threshold_high=dist_threshold_high)
        dict_experiencia = self.filtra_experiencia_relevante(df_puntuaciones)
        dict_respuesta = self.llamada_final(req_experience, puntuacion, dict_experiencia)
        return dict_respuesta