""" ======================================================================== Diff semántico entre temarios Buscador Multisearch — Intervención 4 (extracción del temario) ======================================================================== Compara dos temarios del mismo cuerpo (publicaciones distintas, distinto año) y clasifica cada tema como: - identico : el tema ya estaba en el anterior, mismo título - reformulado : el tema ya estaba, pero el título cambió (paráfrasis, actualización de norma citada, etc.) - nuevo : el tema no estaba en el anterior - eliminado : el tema estaba en el anterior pero no en el nuevo El criterio es semántico (no textual): se calculan embeddings con sentence-transformers y se comparan por similaridad coseno. similaridad >= 0.95 → idéntico similaridad >= 0.70 → reformulado similaridad < 0.70 → son temas distintos Uso típico: from diff_temarios import DiffTemarios diff = DiffTemarios() resultado = diff.comparar( temas_anterior=[Tema(...), Tema(...), ...], temas_nuevo=[Tema(...), Tema(...), ...], ) print(f"Nuevos: {len(resultado.nuevos)}") print(f"Eliminados: {len(resultado.eliminados)}") print(f"Reformulados: {len(resultado.reformulados)}") print(f"Idénticos: {len(resultado.identicos)}") print(f"% de cambio: {resultado.porcentaje_cambio:.1f}%") Dependencias: pip install sentence-transformers numpy """ from dataclasses import dataclass, field from typing import Optional # La importación de sentence_transformers es lazy (solo si se usa) # para que el script pueda inspeccionarse sin tenerlo instalado. # ======================================================================== # Estructura de datos # ======================================================================== @dataclass class Tema: """Replicada aquí para que el módulo sea autónomo. En producción se importa desde parser_bases.py.""" numero: int titulo: str parte: str = "general" embedding: Optional[list] = None # vector de floats @dataclass class TemaReformulado: """Par de temas que se consideran 'el mismo' pero con cambios.""" numero_anterior: int titulo_anterior: str parte_anterior: str numero_nuevo: int titulo_nuevo: str parte_nuevo: str similaridad: float @dataclass class ResultadoDiff: """Resultado completo de comparar dos temarios.""" nuevos: list = field(default_factory=list) # list[Tema] eliminados: list = field(default_factory=list) # list[Tema] reformulados: list = field(default_factory=list) # list[TemaReformulado] identicos: list = field(default_factory=list) # list[Tema] (los del nuevo) @property def total_temas_nuevo(self) -> int: return len(self.nuevos) + len(self.reformulados) + len(self.identicos) @property def total_temas_anterior(self) -> int: return len(self.eliminados) + len(self.reformulados) + len(self.identicos) @property def porcentaje_cambio(self) -> float: """Porcentaje de cambio (nuevos + eliminados + reformulados) sobre el total del temario anterior.""" if self.total_temas_anterior == 0: return 0.0 cambios = len(self.nuevos) + len(self.eliminados) + len(self.reformulados) return (cambios / self.total_temas_anterior) * 100 @property def es_reescritura_total(self) -> bool: """True si más del 70% del temario ha cambiado: probablemente es un cuerpo distinto o reformulación completa.""" return self.porcentaje_cambio > 70 # ======================================================================== # Diff principal # ======================================================================== class DiffTemarios: """Compara semánticamente dos temarios.""" MODELO_DEFAULT = 'sentence-transformers/paraphrase-multilingual-MiniLM-L12-v2' UMBRAL_IDENTICO = 0.95 UMBRAL_REFORMULADO = 0.70 def __init__(self, modelo: str = None, umbral_identico: float = None, umbral_reformulado: float = None): self.modelo_nombre = modelo or self.MODELO_DEFAULT self.umbral_identico = umbral_identico or self.UMBRAL_IDENTICO self.umbral_reformulado = umbral_reformulado or self.UMBRAL_REFORMULADO self._modelo = None # se carga en lazy def _cargar_modelo(self): """Carga el modelo de embeddings la primera vez.""" if self._modelo is None: from sentence_transformers import SentenceTransformer self._modelo = SentenceTransformer(self.modelo_nombre) return self._modelo def _calcular_embedding(self, texto: str): """Calcula el embedding de un texto.""" import numpy as np modelo = self._cargar_modelo() emb = modelo.encode(texto, convert_to_numpy=True, normalize_embeddings=True) return emb def _calcular_embeddings_batch(self, textos: list): """Calcula embeddings de varios textos a la vez (más eficiente).""" import numpy as np modelo = self._cargar_modelo() embs = modelo.encode(textos, convert_to_numpy=True, normalize_embeddings=True, show_progress_bar=False) return embs @staticmethod def _similaridad_coseno(a, b) -> float: """Producto escalar de vectores ya normalizados = similaridad coseno.""" import numpy as np return float(np.dot(a, b)) # ---------------------------------------------------------------- # Comparación principal # ---------------------------------------------------------------- def comparar(self, temas_anterior: list, temas_nuevo: list) -> ResultadoDiff: """Punto de entrada principal. Args: temas_anterior: lista de Tema del temario antiguo temas_nuevo: lista de Tema del temario nuevo Returns: ResultadoDiff con la clasificación de cada tema """ import numpy as np resultado = ResultadoDiff() # Casos triviales if not temas_anterior and not temas_nuevo: return resultado if not temas_anterior: resultado.nuevos = list(temas_nuevo) return resultado if not temas_nuevo: resultado.eliminados = list(temas_anterior) return resultado # Calcular embeddings de todos los temas (batch para eficiencia) textos_anterior = [self._texto_para_embedding(t) for t in temas_anterior] textos_nuevo = [self._texto_para_embedding(t) for t in temas_nuevo] embs_anterior = self._calcular_embeddings_batch(textos_anterior) embs_nuevo = self._calcular_embeddings_batch(textos_nuevo) # Matriz de similaridad entre todos los temas # similaridad[i][j] = sim(anterior[i], nuevo[j]) similaridad = np.dot(embs_anterior, embs_nuevo.T) # Matching greedy: para cada tema nuevo, buscar el anterior más # parecido (que no esté ya emparejado). Es un asignamiento bipartito # simple; para temarios pequeños (<100 temas) es suficiente. emparejados_anterior = set() emparejados_nuevo = set() # Lista de (similaridad, i_anterior, j_nuevo), ordenada descendente pares = [] for i in range(len(temas_anterior)): for j in range(len(temas_nuevo)): pares.append((float(similaridad[i][j]), i, j)) pares.sort(reverse=True, key=lambda x: x[0]) # Tomar pares por orden de similaridad si ambos extremos están libres for sim, i, j in pares: if sim < self.umbral_reformulado: break # ya no vamos a encontrar nada útil if i in emparejados_anterior or j in emparejados_nuevo: continue # Tenemos un emparejamiento tema_ant = temas_anterior[i] tema_nue = temas_nuevo[j] if sim >= self.umbral_identico: resultado.identicos.append(tema_nue) else: resultado.reformulados.append(TemaReformulado( numero_anterior=tema_ant.numero, titulo_anterior=tema_ant.titulo, parte_anterior=tema_ant.parte, numero_nuevo=tema_nue.numero, titulo_nuevo=tema_nue.titulo, parte_nuevo=tema_nue.parte, similaridad=sim, )) emparejados_anterior.add(i) emparejados_nuevo.add(j) # Los que quedan sin emparejar son nuevos o eliminados for i, tema in enumerate(temas_anterior): if i not in emparejados_anterior: resultado.eliminados.append(tema) for j, tema in enumerate(temas_nuevo): if j not in emparejados_nuevo: resultado.nuevos.append(tema) return resultado def _texto_para_embedding(self, tema: Tema) -> str: """Construye el texto que se va a codificar para un tema.""" # Usar título; si hay texto completo, también if hasattr(tema, 'texto_completo') and tema.texto_completo: return f"{tema.titulo}. {tema.texto_completo[:500]}" return tema.titulo # ---------------------------------------------------------------- # Generación de reporte humano # ---------------------------------------------------------------- def generar_reporte(self, resultado: ResultadoDiff, nombre_anterior: str = "Temario anterior", nombre_nuevo: str = "Temario nuevo") -> str: """Genera un reporte legible del diff.""" lineas = [] lineas.append(f"DIFF: {nombre_anterior} → {nombre_nuevo}") lineas.append("=" * 70) lineas.append(f"Total temas anterior: {resultado.total_temas_anterior}") lineas.append(f"Total temas nuevo: {resultado.total_temas_nuevo}") lineas.append(f"") lineas.append(f" ✅ Idénticos: {len(resultado.identicos):3d}") lineas.append(f" 🔀 Reformulados: {len(resultado.reformulados):3d}") lineas.append(f" ➕ Nuevos: {len(resultado.nuevos):3d}") lineas.append(f" ➖ Eliminados: {len(resultado.eliminados):3d}") lineas.append(f"") lineas.append(f" Porcentaje de cambio: {resultado.porcentaje_cambio:.1f}%") if resultado.es_reescritura_total: lineas.append(f" ⚠️ REESCRITURA TOTAL: revisar si es realmente " "el mismo cuerpo") if resultado.nuevos: lineas.append("") lineas.append("TEMAS NUEVOS:") lineas.append("-" * 70) for t in resultado.nuevos: lineas.append(f" [+] {t.parte} · {t.numero}. {t.titulo[:100]}") if resultado.eliminados: lineas.append("") lineas.append("TEMAS ELIMINADOS:") lineas.append("-" * 70) for t in resultado.eliminados: lineas.append(f" [-] {t.parte} · {t.numero}. {t.titulo[:100]}") if resultado.reformulados: lineas.append("") lineas.append("TEMAS REFORMULADOS:") lineas.append("-" * 70) for r in resultado.reformulados: lineas.append(f" [~] (sim={r.similaridad:.2f})") lineas.append(f" antes: {r.parte_anterior} · " f"{r.numero_anterior}. {r.titulo_anterior[:90]}") lineas.append(f" ahora: {r.parte_nuevo} · " f"{r.numero_nuevo}. {r.titulo_nuevo[:90]}") return "\n".join(lineas) def serializar_a_json(self, resultado: ResultadoDiff) -> dict: """Devuelve dict serializable para guardar en BD (columna JSONB).""" return { 'nuevos': [ {'numero': t.numero, 'titulo': t.titulo, 'parte': t.parte} for t in resultado.nuevos ], 'eliminados': [ {'numero': t.numero, 'titulo': t.titulo, 'parte': t.parte} for t in resultado.eliminados ], 'reformulados': [ { 'numero_anterior': r.numero_anterior, 'titulo_anterior': r.titulo_anterior, 'parte_anterior': r.parte_anterior, 'numero_nuevo': r.numero_nuevo, 'titulo_nuevo': r.titulo_nuevo, 'parte_nuevo': r.parte_nuevo, 'similaridad': r.similaridad, } for r in resultado.reformulados ], 'identicos': [ {'numero': t.numero, 'titulo': t.titulo, 'parte': t.parte} for t in resultado.identicos ], 'resumen': { 'total_anterior': resultado.total_temas_anterior, 'total_nuevo': resultado.total_temas_nuevo, 'porcentaje_cambio': resultado.porcentaje_cambio, 'es_reescritura_total': resultado.es_reescritura_total, } } # ======================================================================== # Ejemplo de uso (con datos sintéticos para test) # ======================================================================== if __name__ == '__main__': # Temarios sintéticos para probar sin tener PDFs reales temas_anterior = [ Tema(numero=1, titulo="La Constitución española de 1978. Estructura y " "contenido. Principios generales."), Tema(numero=2, titulo="El Tribunal Constitucional. Composición. " "Atribuciones."), Tema(numero=3, titulo="La organización territorial del Estado. " "Las Comunidades Autónomas."), Tema(numero=4, titulo="La Administración General del Estado. " "Estructura."), Tema(numero=5, titulo="El procedimiento administrativo común. " "Ley 30/1992."), Tema(numero=6, titulo="La Unión Europea. Instituciones."), ] temas_nuevo = [ Tema(numero=1, titulo="La Constitución española de 1978. Estructura y " "contenido. Principios generales."), Tema(numero=2, titulo="El Tribunal Constitucional. Composición, " "funcionamiento y atribuciones."), Tema(numero=3, titulo="La organización territorial del Estado. " "Las Comunidades Autónomas."), Tema(numero=4, titulo="El procedimiento administrativo común. " "Ley 39/2015."), # cambia ley Tema(numero=5, titulo="La Unión Europea. Instituciones y procesos " "decisorios."), # ampliado Tema(numero=6, titulo="Protección de datos personales. RGPD y LOPDGDD."), # nuevo Tema(numero=7, titulo="Transparencia y acceso a la información pública."), # nuevo ] print("Calculando diff... (puede tardar unos segundos la primera vez)") diff_calculador = DiffTemarios() try: resultado = diff_calculador.comparar(temas_anterior, temas_nuevo) reporte = diff_calculador.generar_reporte( resultado, nombre_anterior="Convocatoria 2024", nombre_nuevo="Convocatoria 2026" ) print(reporte) except ImportError as e: print(f"Para ejecutar el ejemplo necesitas: pip install sentence-transformers") print(f"Error: {e}")