""" ======================================================================== Algoritmo de generación de id_proceso Buscador Multisearch — Intervención 3 (vinculación de fases) ======================================================================== Genera un identificador determinista para vincular las distintas fases de un mismo proceso selectivo a lo largo del tiempo. Formato del id: ______ Ejemplo: aeat__inspectores_hacienda__2026__libre Uso típico: from id_proceso import IdentificadorProceso idp = IdentificadorProceso( catalogo_administraciones_path='03_catalogo_administraciones.yaml', catalogo_cuerpos_path='03_catalogo_cuerpos.yaml' ) resultado = idp.identificar( fragmento_completo=texto_del_boletin, boletin='BOE', fecha_publicacion=date(2026, 5, 7) ) # resultado = { # 'id_proceso': 'aeat__inspectores_hacienda__2026__libre', # 'administracion': {'codigo': 'aeat', 'nombre': '...', 'confianza': 0.95}, # 'cuerpo': {'codigo': 'inspectores_hacienda', 'nombre': '...', 'confianza': 0.98}, # 'anio': 2026, # 'turno': 'libre', # 'confianza_global': 0.93 # } Dependencias: pip install pyyaml unidecode python-Levenshtein """ import re import yaml import unicodedata from dataclasses import dataclass, field from datetime import date from typing import Optional from difflib import SequenceMatcher # ======================================================================== # Estructuras de datos # ======================================================================== @dataclass class ResultadoIdentificacion: """Resultado de identificar una entidad (administración o cuerpo).""" codigo: Optional[str] = None nombre: Optional[str] = None confianza: float = 0.0 metodo: str = "" # 'exacto', 'fuzzy', 'codigo_oficial', 'desconocido' @dataclass class ResultadoIdProceso: """Resultado completo de identificar un proceso.""" id_proceso: Optional[str] = None administracion: ResultadoIdentificacion = field( default_factory=ResultadoIdentificacion) cuerpo: ResultadoIdentificacion = field( default_factory=ResultadoIdentificacion) anio: Optional[int] = None turno: str = "libre" confianza_global: float = 0.0 candidatos_alternativos: list = field(default_factory=list) # ======================================================================== # Utilidades de normalización # ======================================================================== def normalizar(texto: str) -> str: """Normaliza texto para comparación: minúsculas, sin acentos, espacios simples.""" if not texto: return "" # Quitar acentos texto = unicodedata.normalize('NFD', texto) texto = ''.join(c for c in texto if unicodedata.category(c) != 'Mn') # Minúsculas y espacios normalizados texto = texto.lower().strip() texto = re.sub(r'\s+', ' ', texto) return texto def similaridad(a: str, b: str) -> float: """Ratio de similaridad entre dos strings normalizados (0.0 a 1.0).""" a_n = normalizar(a) b_n = normalizar(b) return SequenceMatcher(None, a_n, b_n).ratio() # ======================================================================== # Identificador principal # ======================================================================== class IdentificadorProceso: """Identifica el proceso selectivo al que pertenece un hit del buscador.""" UMBRAL_FUZZY = 0.85 LONGITUD_CABECERA_DOCUMENTO = 1500 # primeros chars a inspeccionar def __init__(self, catalogo_administraciones_path: str, catalogo_cuerpos_path: str): with open(catalogo_administraciones_path, encoding='utf-8') as f: self.cat_admin = yaml.safe_load(f)['administraciones'] with open(catalogo_cuerpos_path, encoding='utf-8') as f: self.cat_cuerpos = yaml.safe_load(f)['cuerpos'] # Pre-construir índices de búsqueda exacta (normalizados) self._indice_admin = self._construir_indice( self.cat_admin, campo_nombres='nombres') self._indice_cuerpos = self._construir_indice( self.cat_cuerpos, campo_nombres='nombres') # Índice por código oficial (cuerpos) self._indice_codigo_oficial = { c['codigo_oficial']: c for c in self.cat_cuerpos if c.get('codigo_oficial') } # Pista por boletín: qué administración suele publicar en cada uno self._boletin_a_admin = self._construir_pista_boletines() def _construir_indice(self, catalogo: list, campo_nombres: str) -> dict: """Crea un dict {nombre_normalizado: entrada_completa}.""" indice = {} for entrada in catalogo: for nombre in entrada.get(campo_nombres, []): indice[normalizar(nombre)] = entrada return indice def _construir_pista_boletines(self) -> dict: """Mapea código de boletín a administración matriz probable. Es una pista, no determinante.""" return { 'BOE': 'age', 'BOJA': 'andalucia', 'BOA': 'aragon', 'BOPA': 'asturias', 'BOIB': 'baleares', 'BOC': 'canarias', 'BOCYL': 'castilla_y_leon', 'DOCM': 'castilla_la_mancha', 'DOGC': 'cataluna', 'BOCCE': 'ceuta', 'DOGV': 'c_valenciana', 'DOE': 'extremadura', 'DOG': 'galicia', 'BOR': 'la_rioja', 'BOCM': 'madrid_ccaa', 'BON': 'navarra', 'BOPV': 'pais_vasco', # Provinciales: el padre depende del BOP concreto # BOPB Barcelona -> Cataluña, BOP Sevilla -> Andalucía, etc. } # -------------------------------------------------------------- # Identificación de administración # -------------------------------------------------------------- def identificar_administracion( self, fragmento_completo: str, boletin: str = None ) -> ResultadoIdentificacion: """Identifica la administración convocante.""" # Inspeccionar la cabecera del documento (donde suele aparecer # el emisor: "Ayuntamiento de X", "Conselleria de Y", etc.) cabecera = fragmento_completo[:self.LONGITUD_CABECERA_DOCUMENTO] cabecera_norm = normalizar(cabecera) # 1. Match exacto for nombre_norm, entrada in self._indice_admin.items(): if nombre_norm in cabecera_norm: return ResultadoIdentificacion( codigo=entrada['codigo'], nombre=entrada['nombres'][0], confianza=0.95, metodo='exacto' ) # 2. Match fuzzy sobre líneas de cabecera lineas_cabecera = [l.strip() for l in cabecera.split('\n') if l.strip()] mejor_match = (None, 0.0) for linea in lineas_cabecera[:20]: # solo primeras 20 líneas for entrada in self.cat_admin: for nombre in entrada['nombres']: s = similaridad(linea, nombre) if s > mejor_match[1] and s >= self.UMBRAL_FUZZY: mejor_match = (entrada, s) if mejor_match[0]: entrada, score = mejor_match return ResultadoIdentificacion( codigo=entrada['codigo'], nombre=entrada['nombres'][0], confianza=score, metodo='fuzzy' ) # 3. Pista por boletín (poco específica pero algo es algo) if boletin and boletin in self._boletin_a_admin: codigo_admin_pista = self._boletin_a_admin[boletin] entrada = next( (a for a in self.cat_admin if a['codigo'] == codigo_admin_pista), None ) if entrada: return ResultadoIdentificacion( codigo=entrada['codigo'], nombre=entrada['nombres'][0], confianza=0.40, # confianza baja, es solo una pista metodo='boletin' ) # 4. Nada coincide return ResultadoIdentificacion(metodo='desconocido') # -------------------------------------------------------------- # Identificación de cuerpo # -------------------------------------------------------------- def identificar_cuerpo( self, fragmento_completo: str ) -> ResultadoIdentificacion: """Identifica el cuerpo/escala/categoría convocada.""" texto_norm = normalizar(fragmento_completo) # 1. Código oficial (XXXX entre comillas o seguido del nombre) # Patrones típicos: "0011 SUPERIOR DE INSPECTORES..." o "Código 1166" match_codigo = re.search(r'\b(\d{4}[A-Z]?)\s+[A-Z]', fragmento_completo) if match_codigo: codigo_oficial = match_codigo.group(1) if codigo_oficial in self._indice_codigo_oficial: entrada = self._indice_codigo_oficial[codigo_oficial] return ResultadoIdentificacion( codigo=entrada['codigo'], nombre=entrada['nombres'][0], confianza=0.99, metodo='codigo_oficial' ) # 2. Match exacto sobre nombres for nombre_norm, entrada in self._indice_cuerpos.items(): if nombre_norm in texto_norm: return ResultadoIdentificacion( codigo=entrada['codigo'], nombre=entrada['nombres'][0], confianza=0.95, metodo='exacto' ) # 3. Match fuzzy sobre frases que contienen palabras-clave # ("cuerpo", "escala", "categoría", "subescala", "plaza", "puesto") regex_frase = re.compile( r'(cuerpo|escala|categoria|subescala|plaza|puesto|funcionarios?\s+de)' r'\s+([\w\s,]+?)(?:\.|\n|;)', re.IGNORECASE ) candidatos = regex_frase.findall(fragmento_completo[:5000]) mejor_match = (None, 0.0) for _, frase in candidatos: for entrada in self.cat_cuerpos: for nombre in entrada['nombres']: s = similaridad(frase, nombre) if s > mejor_match[1] and s >= self.UMBRAL_FUZZY: mejor_match = (entrada, s) if mejor_match[0]: entrada, score = mejor_match return ResultadoIdentificacion( codigo=entrada['codigo'], nombre=entrada['nombres'][0], confianza=score, metodo='fuzzy' ) return ResultadoIdentificacion(metodo='desconocido') # -------------------------------------------------------------- # Identificación de año y turno # -------------------------------------------------------------- def identificar_anio( self, fragmento_completo: str, fecha_publicacion: date ) -> int: """Identifica el año de la OEP/convocatoria. Prefiere lo que diga el texto, fallback al año de publicación.""" # Patrones típicos: "oferta de empleo público para 2026", # "ejercicio 2026", "OEP 2026" patrones = [ r'oferta de empleo p[uú]blico[\s\w]*?(\d{4})', r'ejercicio\s+(\d{4})', r'OEP[\s\w]*?(\d{4})', r'convocatoria[\s\w]*?(\d{4})', r'a[ñn]o\s+(\d{4})', ] for p in patrones: m = re.search(p, fragmento_completo, re.IGNORECASE) if m: anio = int(m.group(1)) # Sanity check: año razonable if 2020 <= anio <= fecha_publicacion.year + 2: return anio return fecha_publicacion.year def identificar_turno(self, fragmento_completo: str) -> str: """Detecta el turno (libre / promoción interna / reserva).""" texto = fragmento_completo.lower() # Patrones para promoción interna if re.search(r'\bpromoci[oó]n\s+interna\b', texto): # Pero hay que distinguir si es SOLO PI o si menciona ambos if re.search(r'\bturno libre\b|\bacceso libre\b', texto): # Mencionan ambos: en este contexto el evento puede ser # de PI o de libre; mirar más contexto, por defecto libre return 'libre' return 'promocion_interna' # Patrones para discapacidad if re.search(r'discapacidad\s+intelectual', texto): return 'discapacidad_intelectual' if re.search(r'\breserva\b.*discapacidad', texto): return 'reserva_discapacidad' return 'libre' # -------------------------------------------------------------- # Identificación completa # -------------------------------------------------------------- def identificar( self, fragmento_completo: str, boletin: str, fecha_publicacion: date ) -> ResultadoIdProceso: """Identifica el proceso completo. Punto de entrada principal.""" resultado = ResultadoIdProceso() # Identificar administración resultado.administracion = self.identificar_administracion( fragmento_completo, boletin) # Identificar cuerpo resultado.cuerpo = self.identificar_cuerpo(fragmento_completo) # Identificar año resultado.anio = self.identificar_anio( fragmento_completo, fecha_publicacion) # Identificar turno resultado.turno = self.identificar_turno(fragmento_completo) # Generar id_proceso si tenemos admin + cuerpo if (resultado.administracion.codigo and resultado.cuerpo.codigo and resultado.administracion.codigo != 'desconocido' and resultado.cuerpo.codigo != 'desconocido'): resultado.id_proceso = ( f"{resultado.administracion.codigo}__" f"{resultado.cuerpo.codigo}__" f"{resultado.anio}__" f"{resultado.turno}" ) # Calcular confianza global como media ponderada # (administración cuenta 0.4, cuerpo 0.6: el cuerpo es más # determinante para el proceso concreto) resultado.confianza_global = ( resultado.administracion.confianza * 0.4 + resultado.cuerpo.confianza * 0.6 ) return resultado # -------------------------------------------------------------- # Funciones auxiliares # -------------------------------------------------------------- def vincular_evento_a_proceso( self, id_proceso: str, fase: str, fecha: date, url_pdf: str, # ... resto de campos del evento ) -> None: """Crea el evento en BD y actualiza el proceso correspondiente. Implementación real con psycopg2 omitida. Pseudo-código: with conn.cursor() as cur: # 1. Asegurar que el proceso existe (crear si no) cur.execute(''' INSERT INTO procesos (id_proceso, ...) VALUES (%s, ...) ON CONFLICT (id_proceso) DO NOTHING ''', (id_proceso, ...)) # 2. Insertar el evento cur.execute(''' INSERT INTO eventos_proceso (id_proceso, fase, ...) VALUES (%s, %s, ...) ON CONFLICT (id_proceso, codigo_cve) DO NOTHING ''', (id_proceso, fase, ...)) # 3. Actualizar fechas y estado del proceso según la fase if fase == 'oep': cur.execute(''' UPDATE procesos SET fecha_oep = %s, url_oep = %s, estado = CASE WHEN estado IN ('oep_publicada', NULL) THEN 'oep_publicada' ELSE estado END WHERE id_proceso = %s ''', (fecha, url_pdf, id_proceso)) elif fase == 'bases': cur.execute(''' UPDATE procesos SET fecha_bases = %s, url_bases = %s, estado = 'bases_publicadas' WHERE id_proceso = %s ''', (fecha, url_pdf, id_proceso)) elif fase == 'apertura': cur.execute(''' UPDATE procesos SET fecha_apertura_plazo = %s, url_apertura = %s, estado = 'plazo_abierto' WHERE id_proceso = %s ''', (fecha, url_pdf, id_proceso)) conn.commit() """ pass # ======================================================================== # Ejemplo de uso (para pruebas) # ======================================================================== if __name__ == '__main__': from datetime import date idp = IdentificadorProceso( catalogo_administraciones_path='03_catalogo_administraciones.yaml', catalogo_cuerpos_path='03_catalogo_cuerpos.yaml' ) # Caso 1: OEP estatal 2026 (Inspectores Hacienda) fragmento_1 = """ MINISTERIO PARA LA TRANSFORMACION DIGITAL Y DE LA FUNCION PUBLICA Real Decreto 387/2026, de 6 de mayo, por el que se aprueba la oferta de empleo público correspondiente al ejercicio 2026. ... 0011 SUPERIOR DE INSPECTORES DE HACIENDA DEL ESTADO. 120 10 130 """ r1 = idp.identificar(fragmento_1, boletin='BOE', fecha_publicacion=date(2026, 5, 7)) print(f"Caso 1: {r1.id_proceso} (confianza {r1.confianza_global:.2f})") # Esperado: age__inspectores_hacienda__2026__libre # Caso 2: Bases convocatoria Ayuntamiento de Madrid - TAG fragmento_2 = """ EXCMO. AYUNTAMIENTO DE MADRID Bases reguladoras de la convocatoria para la cobertura mediante concurso-oposición de 15 plazas de Técnico de Administración General (TAG) del Ayuntamiento de Madrid. Año 2025. Sistema selectivo: concurso-oposición libre. Fase de oposición... """ r2 = idp.identificar(fragmento_2, boletin='BOCM', fecha_publicacion=date(2026, 8, 15)) print(f"Caso 2: {r2.id_proceso} (confianza {r2.confianza_global:.2f})") # Esperado: ayto_madrid__tag__2025__libre