app.py

import gradio as gr
import PyPDF2
import os
import re
import vertexai
from vertexai.generative_models import GenerativeModel, Part, SafetySetting
from difflib import SequenceMatcher

# --------------------
# CONFIGURACIÓN GLOBAL
# --------------------
generation_config = {
    "max_output_tokens": 8192,
    "temperature": 0,
    "top_p": 0.8,
}
safety_settings = [
    SafetySetting(
        category=SafetySetting.HarmCategory.HARM_CATEGORY_HATE_SPEECH,
        threshold=SafetySetting.HarmBlockThreshold.OFF
    ),
    SafetySetting(
        category=SafetySetting.HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
        threshold=SafetySetting.HarmBlockThreshold.OFF
    ),
    SafetySetting(
        category=SafetySetting.HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
        threshold=SafetySetting.HarmBlockThreshold.OFF
    ),
    SafetySetting(
        category=SafetySetting.HarmCategory.HARM_CATEGORY_HARASSMENT,
        threshold=SafetySetting.HarmBlockThreshold.OFF
    ),
]

def configurar_credenciales(json_path: str):
    """Configura credenciales de Google Cloud a partir de un archivo JSON."""
    os.environ["GOOGLE_APPLICATION_CREDENTIALS"] = json_path

# -----------
# LECTURA PDF
# -----------
def extraer_texto(pdf_path: str) -> str:
    """
    Extrae el texto de todas las páginas de un PDF con PyPDF2.
    Retorna un string con todo el texto concatenado.
    """
    texto_total = ""
    with open(pdf_path, "rb") as f:
        lector = PyPDF2.PdfReader(f)
        for page in lector.pages:
            texto_total += page.extract_text() or ""
    return texto_total

# -----------
# PARSEO DE TEXTO
# -----------
def split_secciones(texto: str) -> (str, str):
    """
    Separa el texto en dos partes: la sección 'Preguntas' y la sección 'RESPUESTAS'.
    Busca las palabras 'Preguntas' y 'RESPUESTAS' ignorando mayúsculas y espacios al inicio.
    """
    match_preg = re.search(r'(?im)^\s*preguntas', texto)
    match_resp = re.search(r'(?im)^\s*respuestas', texto)
    
    if not match_preg or not match_resp:
        return (texto, "")
    
    start_preg = match_preg.end()  # Fin de "Preguntas"
    start_resp = match_resp.start()  # Inicio de "RESPUESTAS"
    
    texto_preguntas = texto[start_preg:start_resp].strip()
    texto_respuestas = texto[match_resp.end():].strip()
    return (texto_preguntas, texto_respuestas)

def parsear_enumeraciones(texto: str) -> dict:
    """
    Dado un texto que contiene enumeraciones de preguntas (por ejemplo, "1. 1- RTA1" o "2- RTA2"),
    separa cada número y su contenido.
    Retorna un dict: {"Pregunta 1": "contenido", "Pregunta 2": "contenido", ...}.
    Este patrón es flexible y tolera espacios al inicio y formatos creativos.
    Además, elimina duplicados al inicio de la respuesta (por ejemplo, "Durante Durante ...").
    """
    # Se utiliza un lookahead para dividir cada bloque cuando se encuentre una línea que empiece con un número,
    # un punto o guión y, opcionalmente, otro número con punto o guión.
    bloques = re.split(r'(?=^\s*\d+[\.\-]\s*(?:\d+[\.\-])?\s*)', texto, flags=re.MULTILINE)
    resultado = {}
    for bloque in bloques:
        bloque = bloque.strip()
        if not bloque:
            continue
        # Extraemos el número de la pregunta y el contenido.
        match = re.match(r'^\s*(\d+)[\.\-]\s*(?:\d+[\.\-])?\s*(.*)', bloque)
        if match:
            numero = match.group(1)
            contenido = match.group(2)
            # Si hay múltiples líneas, unimos las líneas adicionales.
            lineas = bloque.split("\n")
            if len(lineas) > 1:
                contenido_completo = " ".join([linea.strip() for linea in lineas[1:]])
                if contenido_completo:
                    contenido = contenido + " " + contenido_completo
            # Eliminar duplicados al inicio (por ejemplo, "Durante Durante ..." se convierte en "Durante ...")
            contenido = re.sub(r'^(\S+)(\s+\1)+\s+', r'\1 ', contenido)
            resultado[f"Pregunta {numero}"] = contenido.strip()
    return resultado

# ------------
# COMPARACIÓN Y ANÁLISIS
# ------------
def similar_textos(texto1: str, texto2: str) -> float:
    """Calcula la similitud entre dos textos (valor entre 0 y 1)."""
    return SequenceMatcher(None, texto1, texto2).ratio()

def comparar_preguntas_respuestas(dict_docente: dict, dict_alumno: dict) -> (str, list):
    """
    Compara las respuestas del docente (correctas) con las del alumno.
    Para cada pregunta:
      - Si no fue asignada se indica "No fue asignada".
      - Si fue asignada se calcula la similitud y se evalúa:
          * Correcta: ratio >= 0.85
          * Incompleta: 0.5 <= ratio < 0.85
          * Incorrecta: ratio < 0.5
    Devuelve:
      - Un string con la retroalimentación por pregunta.
      - Una lista de diccionarios con el análisis por pregunta (solo para las asignadas).
    """
    feedback = []
    analisis = []
    for pregunta, resp_correcta in dict_docente.items():
        correct_clean = " ".join(resp_correcta.split())
        resp_alumno_raw = dict_alumno.get(pregunta, "").strip()
        
        if not resp_alumno_raw:
            feedback.append(
                f"**{pregunta}**\n"
                f"Respuesta del alumno: No fue asignada.\n"
                f"Respuesta correcta: {correct_clean}\n"
            )
            analisis.append({"pregunta": pregunta, "asignada": False})
        else:
            alumno_clean = " ".join(resp_alumno_raw.split())
            ratio = similar_textos(alumno_clean.lower(), correct_clean.lower())
            if ratio >= 0.85:
                eval_text = "La respuesta es correcta."
                resultado = "correcta"
            elif ratio >= 0.5:
                eval_text = "La respuesta es incompleta. Se observa que faltan conceptos clave."
                resultado = "incompleta"
            else:
                eval_text = "La respuesta es incorrecta. No se refleja el mecanismo o concepto correcto."
                resultado = "incorrecta"
            feedback.append(
                f"**{pregunta}**\n"
                f"Respuesta del alumno: {alumno_clean}\n"
                f"Respuesta correcta: {correct_clean}\n"
                f"{eval_text}\n"
            )
            analisis.append({"pregunta": pregunta, "asignada": True, "resultado": resultado})
    return "\n".join(feedback), analisis

# -----------
# FUNCIÓN PRINCIPAL
# -----------
def revisar_examen(json_cred, pdf_docente, pdf_alumno):
    """
    Función generadora que:
      1. Configura credenciales.
      2. Extrae y parsea el contenido de los PDFs.
      3. Separa las secciones 'Preguntas' y 'RESPUESTAS'.
      4. Parsea las enumeraciones de cada sección (soportando formatos creativos).
      5. Compara las respuestas del alumno con las correctas.
      6. Llama a un LLM para generar un resumen final con retroalimentación.
    """
    yield "Cargando credenciales..."
    try:
        configurar_credenciales(json_cred.name)
        
        yield "Inicializando Vertex AI..."
        vertexai.init(project="deploygpt", location="us-central1")
        
        yield "Extrayendo texto del PDF del docente..."
        texto_docente = extraer_texto(pdf_docente.name)
        
        yield "Extrayendo texto del PDF del alumno..."
        texto_alumno = extraer_texto(pdf_alumno.name)
        
        yield "Dividiendo secciones (docente)..."
        preguntas_doc, respuestas_doc = split_secciones(texto_docente)
        
        yield "Dividiendo secciones (alumno)..."
        preguntas_alum, respuestas_alum = split_secciones(texto_alumno)
        
        yield "Parseando enumeraciones (docente)..."
        dict_preg_doc = parsear_enumeraciones(preguntas_doc)
        dict_resp_doc = parsear_enumeraciones(respuestas_doc)
        # Unir las respuestas correctas del docente
        dict_docente = {}
        for key in dict_preg_doc:
            dict_docente[key] = dict_resp_doc.get(key, "")
        
        yield "Parseando enumeraciones (alumno)..."
        dict_preg_alum = parsear_enumeraciones(preguntas_alum)
        dict_resp_alum = parsear_enumeraciones(respuestas_alum)
        # Unir las respuestas del alumno
        dict_alumno = {}
        for key in dict_preg_alum:
            dict_alumno[key] = dict_resp_alum.get(key, "")
        
        yield "Comparando preguntas y respuestas..."
        feedback_text, analisis = comparar_preguntas_respuestas(dict_docente, dict_alumno)
        if len(feedback_text.strip()) < 5:
            yield "No se encontraron preguntas o respuestas válidas."
            return
        
        # Generar resumen global utilizando el LLM (solo para preguntas asignadas)
        analisis_asignadas = [a for a in analisis if a.get("asignada")]
        resumen_prompt = f"""
A continuación se presenta el análisis por pregunta de un examen sobre la regulación del colesterol, considerando solo las preguntas asignadas al alumno:
{analisis_asignadas}
Con base en este análisis, genera un resumen del desempeño del alumno en el examen que incluya:
- Puntos fuertes: conceptos que el alumno ha comprendido correctamente.
- Puntos a reforzar: preguntas en las que la respuesta fue incompleta o incorrecta, indicando qué conceptos clave faltaron o se confundieron.
- Una recomendación general sobre si el alumno demuestra comprender los fundamentos o si necesita repasar el tema.
No incluyas en el análisis las preguntas que no fueron asignadas.
"""
        yield "Generando resumen final con LLM..."
        model = GenerativeModel(
            "gemini-1.5-pro-001",
            system_instruction=["Eres un profesor experto en bioquímica. Evalúa el desempeño del alumno basándote en los conceptos clave, sin inventar elementos adicionales."]
        )
        summary_part = Part.from_text(resumen_prompt)
        summary_resp = model.generate_content(
            [summary_part],
            generation_config=generation_config,
            safety_settings=safety_settings,
            stream=False
        )
        resumen_final = summary_resp.text.strip()
        final_result = f"{feedback_text}\n\n**Resumen del desempeño:**\n{resumen_final}"
        yield final_result
    except Exception as e:
        yield f"Error al procesar: {str(e)}"

# -----------------
# INTERFAZ DE GRADIO
# -----------------
interface = gr.Interface(
    fn=revisar_examen,
    inputs=[
        gr.File(label="Credenciales JSON"),
        gr.File(label="PDF del Docente"),
        gr.File(label="PDF del Alumno")
    ],
    outputs="text",
    title="Revisión de Exámenes (Preguntas/Respuestas enumeradas)",
    description=(
        "Sube las credenciales, el PDF del docente (con las preguntas y respuestas correctas) y el PDF del alumno. "
        "El sistema separa las secciones 'Preguntas' y 'RESPUESTAS', parsea las enumeraciones (soportando formatos creativos) "
        "y luego compara las respuestas. Se evalúa si el alumno comprende los conceptos fundamentales: si la respuesta está incompleta se indica qué falta, "
        "si es incorrecta se comenta por qué, y se omiten las preguntas no asignadas. Finalmente, se genera un resumen con recomendaciones."
    )
)

interface.launch(debug=True)