""" ======================================================================== Generador de calendarios iCalendar (.ics) suscriptibles Buscador Multisearch — Entrega 5 (sistema de notificaciones) ======================================================================== Genera un archivo .ics dinámico por usuario, suscribible desde Google Calendar, Outlook, Apple Calendar o cualquier cliente compatible con RFC 5545. Cada usuario tiene una URL única firmada del tipo: https://buscador.example/ical/{token}.ics El token es un identificador opaco (no es el id_usuario) para evitar filtración de identidad. Al acceder a esa URL, el servidor: 1. Resuelve el token → id_usuario. 2. Lee de la BD las fechas clave de todos los procesos seguidos. 3. Genera el .ics al vuelo y lo sirve con headers de caché correctos. Frecuencia de refresco: los clientes suelen consultar el .ics cada 15-30 minutos. No hace falta cachear el resultado más allá de eso. Uso típico desde el servidor web (FastAPI ejemplo): @app.get("/ical/{token}.ics") async def servir_calendario(token: str): usuario = resolver_token(token) if not usuario: return Response(status_code=404) ics_content = generar_ics_usuario( id_usuario=usuario.id_usuario, conn=db_connection, ) return Response( content=ics_content, media_type="text/calendar; charset=utf-8", headers={ "Cache-Control": "max-age=900", # 15 minutos "Content-Disposition": "inline; filename=oposiciones.ics", } ) Dependencias: pip install icalendar python-dateutil psycopg2-binary """ import hashlib import secrets from dataclasses import dataclass from datetime import datetime, date, timedelta, timezone from typing import Optional, Iterable from icalendar import Calendar, Event, vDatetime, vDate, vText from dateutil import tz # ======================================================================== # Constantes y configuración # ======================================================================== TZ_MADRID = tz.gettz('Europe/Madrid') PREFIJO_UID = 'buscador-multisearch' DOMINIO_UID = 'oposicionesdeporte.com' # Mapeo de tipo de fecha a emoji para el título del evento EMOJI_POR_TIPO = { 'fin_plazo_solicitudes': '⏰', 'plazo_subsanacion': '✏️', 'lista_provisional': '📋', 'lista_definitiva': '📑', 'primer_ejercicio': '📝', 'segundo_ejercicio': '📝', 'tercer_ejercicio': '📝', 'cuarto_ejercicio': '📝', 'quinto_ejercicio': '📝', 'fin_concurso_meritos': '🎖️', 'fin_reclamaciones': '⚖️', 'publicacion_aprobados': '🎉', 'eleccion_destinos': '📍', 'toma_posesion': '🤝', } NOMBRE_LARGO_POR_TIPO = { 'fin_plazo_solicitudes': 'Fin plazo solicitudes', 'plazo_subsanacion': 'Fin plazo subsanación', 'lista_provisional': 'Lista provisional admitidos', 'lista_definitiva': 'Lista definitiva admitidos', 'primer_ejercicio': '1º Ejercicio', 'segundo_ejercicio': '2º Ejercicio', 'tercer_ejercicio': '3º Ejercicio', 'cuarto_ejercicio': '4º Ejercicio', 'quinto_ejercicio': '5º Ejercicio', 'fin_concurso_meritos': 'Fin concurso de méritos', 'fin_reclamaciones': 'Fin plazo reclamaciones', 'publicacion_aprobados': 'Publicación aprobados', 'eleccion_destinos': 'Elección de destinos', 'toma_posesion': 'Toma de posesión', } # Recordatorios automáticos por tipo (en minutos antes del evento) RECORDATORIOS_POR_TIPO = { 'fin_plazo_solicitudes': [7 * 24 * 60, 1 * 24 * 60], # 7 días y 1 día antes 'plazo_subsanacion': [3 * 24 * 60, 1 * 24 * 60], # 3 días y 1 día 'primer_ejercicio': [30 * 24 * 60, 7 * 24 * 60, 1 * 24 * 60], 'segundo_ejercicio': [7 * 24 * 60, 1 * 24 * 60], 'tercer_ejercicio': [7 * 24 * 60, 1 * 24 * 60], 'fin_reclamaciones': [3 * 24 * 60, 1 * 24 * 60], 'toma_posesion': [7 * 24 * 60, 1 * 24 * 60], } # ======================================================================== # Estructura de datos # ======================================================================== @dataclass class FechaClave: """Una fecha clave de un proceso. Replica el modelo de la tabla fechas_clave del schema SQL.""" id_fecha: str id_proceso: str tipo: str fecha: date hora: Optional[str] = None # 'HH:MM' o None si es todo el día lugar: Optional[str] = None notas: Optional[str] = None fuente_url: Optional[str] = None estimada: bool = False # Datos del proceso asociado (joined en la query) proceso_admin: Optional[str] = None proceso_cuerpo: Optional[str] = None # ======================================================================== # Generación del calendario # ======================================================================== class GeneradorICS: """Genera el archivo .ics para un usuario.""" def __init__(self, nombre_calendario: str = "Mis oposiciones", organizacion: str = "Buscador Multisearch"): self.nombre_calendario = nombre_calendario self.organizacion = organizacion def generar(self, fechas: Iterable[FechaClave], preferencias_ical: Optional[dict] = None) -> bytes: """Genera el contenido completo del .ics. Args: fechas: iterable de FechaClave a incluir preferencias_ical: dict con flags del usuario (qué tipos incluir). Si None, se incluyen todos. Returns: bytes con el contenido del .ics (codificado UTF-8) """ cal = Calendar() cal.add('prodid', f'-//{self.organizacion}//ES') cal.add('version', '2.0') cal.add('x-wr-calname', self.nombre_calendario) cal.add('x-wr-caldesc', 'Fechas clave de tus procesos selectivos en seguimiento') cal.add('x-wr-timezone', 'Europe/Madrid') cal.add('method', 'PUBLISH') cal.add('refresh-interval;value=duration', 'PT15M') for fc in fechas: # Filtrar según preferencias del usuario if preferencias_ical and not self._tipo_incluido(fc.tipo, preferencias_ical): continue evento = self._crear_evento(fc) cal.add_component(evento) return cal.to_ical() def _tipo_incluido(self, tipo: str, preferencias: dict) -> bool: """Decide si incluir este tipo de fecha según las preferencias.""" if tipo in ('fin_plazo_solicitudes', 'plazo_subsanacion'): return preferencias.get('ical_fin_plazo', True) if tipo.startswith('primer_') or tipo.startswith('segundo_') \ or tipo.startswith('tercer_') or tipo.startswith('cuarto_') \ or tipo.startswith('quinto_'): return preferencias.get('ical_examenes', True) if tipo in ('lista_provisional', 'plazo_subsanacion'): return preferencias.get('ical_subsanaciones', True) if tipo == 'toma_posesion': return preferencias.get('ical_toma_posesion', False) return True # por defecto incluir def _crear_evento(self, fc: FechaClave) -> Event: """Crea un Event iCal a partir de una FechaClave.""" evento = Event() # UID único y estable (mismo evento → mismo UID, permite # actualizaciones desde el cliente) uid = self._generar_uid(fc) evento.add('uid', uid) # Título emoji = EMOJI_POR_TIPO.get(fc.tipo, '📅') nombre = NOMBRE_LARGO_POR_TIPO.get(fc.tipo, fc.tipo.replace('_', ' ').title()) titulo = f"{emoji} {nombre}" if fc.proceso_cuerpo: titulo += f" — {fc.proceso_cuerpo}" if fc.proceso_admin: titulo += f" ({fc.proceso_admin})" if fc.estimada: titulo += " (estimado)" evento.add('summary', titulo) # Fecha y hora if fc.hora: # Evento puntual con hora concreta try: h, m = fc.hora.split(':') dt = datetime.combine(fc.fecha, datetime.min.time()) \ .replace(hour=int(h), minute=int(m), tzinfo=TZ_MADRID) evento.add('dtstart', vDatetime(dt)) # Duración por defecto: 2h para exámenes, 1h para resto duracion = timedelta(hours=2 if 'ejercicio' in fc.tipo else 1) evento.add('dtend', vDatetime(dt + duracion)) except ValueError: # Si la hora viene mal formada, usar todo el día evento.add('dtstart', vDate(fc.fecha)) evento.add('dtend', vDate(fc.fecha + timedelta(days=1))) else: # Evento de todo el día evento.add('dtstart', vDate(fc.fecha)) evento.add('dtend', vDate(fc.fecha + timedelta(days=1))) # Lugar if fc.lugar: evento.add('location', fc.lugar) # Descripción / notas descripcion_lineas = [] if fc.notas: descripcion_lineas.append(fc.notas) if fc.estimada: descripcion_lineas.append( "⚠️ Esta fecha es una estimación del sistema, no oficial. " "Se actualizará automáticamente cuando se publique la oficial.") if fc.fuente_url: descripcion_lineas.append(f"Ver publicación oficial: {fc.fuente_url}") descripcion_lineas.append("") descripcion_lineas.append(f"--\nBuscador Multisearch · proceso: {fc.id_proceso}") evento.add('description', '\n'.join(descripcion_lineas)) # URL del proceso en el dashboard if fc.id_proceso: evento.add('url', f"https://buscador.example/procesos/{fc.id_proceso}") # Recordatorios automáticos (alarmas) recordatorios = RECORDATORIOS_POR_TIPO.get(fc.tipo, []) for minutos_antes in recordatorios: from icalendar import Alarm alarm = Alarm() alarm.add('action', 'DISPLAY') alarm.add('description', f"Recordatorio: {nombre}") alarm.add('trigger', timedelta(minutes=-minutos_antes)) evento.add_component(alarm) # Marca de tiempo de creación/última modificación evento.add('dtstamp', vDatetime(datetime.now(tz=timezone.utc))) # Estado: confirmado si es oficial, tentativo si es estimado evento.add('status', 'TENTATIVE' if fc.estimada else 'CONFIRMED') # Color (algunos clientes lo respetan) evento.add('color', self._color_por_tipo(fc.tipo)) return evento def _generar_uid(self, fc: FechaClave) -> str: """UID único y estable para un evento. Misma fecha clave → mismo UID en sucesivas regeneraciones del calendario.""" contenido = f"{fc.id_proceso}|{fc.tipo}|{fc.fecha.isoformat()}" h = hashlib.md5(contenido.encode()).hexdigest()[:16] return f"{PREFIJO_UID}-{h}@{DOMINIO_UID}" def _color_por_tipo(self, tipo: str) -> str: """Color sugerido (RFC 7986).""" colores = { 'fin_plazo_solicitudes': 'orange', 'plazo_subsanacion': 'purple', 'lista_provisional': 'lightblue', 'lista_definitiva': 'blue', 'primer_ejercicio': 'red', 'segundo_ejercicio': 'red', 'tercer_ejercicio': 'red', 'toma_posesion': 'green', } return colores.get(tipo, 'gray') # ======================================================================== # Generación de tokens para URLs únicas # ======================================================================== def generar_token_ical(id_usuario: str) -> str: """Genera un token aleatorio para la URL del .ics del usuario. Este token se guarda en preferencias_notificacion.ical_url_token. No es el id_usuario directamente para evitar filtración de identidad. """ return secrets.token_urlsafe(32) # ======================================================================== # Función de alto nivel: generar .ics completo para un usuario # ======================================================================== def generar_ics_usuario(id_usuario: str, conn) -> bytes: """Genera el contenido del .ics para un usuario consultando la BD. Args: id_usuario: UUID del usuario conn: conexión psycopg2 a PostgreSQL Returns: bytes con el .ics codificado UTF-8 """ # 1. Cargar preferencias del usuario with conn.cursor() as cur: cur.execute(""" SELECT ical_fin_plazo, ical_examenes, ical_subsanaciones, ical_toma_posesion FROM preferencias_notificacion WHERE id_usuario = %s """, (id_usuario,)) row = cur.fetchone() preferencias = { 'ical_fin_plazo': row[0] if row else True, 'ical_examenes': row[1] if row else True, 'ical_subsanaciones': row[2] if row else True, 'ical_toma_posesion': row[3] if row else False, } if row else {} # 2. Cargar las fechas clave de los procesos seguidos with conn.cursor() as cur: cur.execute(""" SELECT fc.id_fecha, fc.id_proceso, fc.tipo, fc.fecha, fc.hora, fc.lugar, fc.notas, fc.fuente_url, fc.estimada, p.administracion_nombre, p.cuerpo_nombre FROM fechas_clave fc JOIN procesos p ON p.id_proceso = fc.id_proceso WHERE fc.id_proceso IN ( -- Procesos seguidos directamente SELECT id_proceso FROM suscripciones_proceso WHERE id_usuario = %s AND fecha_baja IS NULL AND incluir_en_ical = TRUE UNION -- Procesos de categorías a las que está suscrito SELECT p2.id_proceso FROM procesos p2 JOIN suscripciones s ON s.categoria = p2.categoria_tematica WHERE s.id_usuario = %s AND s.activa = TRUE ) -- Solo eventos futuros o muy recientes (últimos 7 días) AND fc.fecha >= CURRENT_DATE - INTERVAL '7 days' -- Limitar a 18 meses adelante para evitar calendarios infinitos AND fc.fecha <= CURRENT_DATE + INTERVAL '18 months' ORDER BY fc.fecha """, (id_usuario, id_usuario)) fechas = [] for r in cur.fetchall(): fechas.append(FechaClave( id_fecha=str(r[0]), id_proceso=r[1], tipo=r[2], fecha=r[3], hora=r[4].strftime('%H:%M') if r[4] else None, lugar=r[5], notas=r[6], fuente_url=r[7], estimada=r[8], proceso_admin=r[9], proceso_cuerpo=r[10], )) # 3. Generar el .ics generador = GeneradorICS() return generador.generar(fechas, preferencias) # ======================================================================== # Ejemplo de uso # ======================================================================== if __name__ == '__main__': # Ejemplo con datos sintéticos (sin BD) fechas_sinteticas = [ FechaClave( id_fecha='f1', id_proceso='aeat__inspectores_hacienda__2026__libre', tipo='fin_plazo_solicitudes', fecha=date(2026, 6, 15), hora='23:59', notas='Presentación telemática en sede.agenciatributaria.gob.es', fuente_url='https://www.boe.es/diario_boe/txt.php?id=BOE-A-2026-XXXX', estimada=False, proceso_admin='AEAT', proceso_cuerpo='Inspectores de Hacienda', ), FechaClave( id_fecha='f2', id_proceso='aeat__inspectores_hacienda__2026__libre', tipo='primer_ejercicio', fecha=date(2026, 10, 18), hora='10:00', lugar='IFEMA, Madrid', notas='DNI + bolígrafo. Acceso 1 hora antes.', estimada=True, # todavía no oficial proceso_admin='AEAT', proceso_cuerpo='Inspectores de Hacienda', ), FechaClave( id_fecha='f3', id_proceso='ayto_madrid__tag__2026__libre', tipo='lista_provisional', fecha=date(2026, 7, 4), proceso_admin='Ayuntamiento de Madrid', proceso_cuerpo='Técnico de Administración General', notas='10 días hábiles para subsanación', ), ] generador = GeneradorICS(nombre_calendario="Mis oposiciones 2026") ics_bytes = generador.generar(fechas_sinteticas) print("=== ARCHIVO .ICS GENERADO ===") print(ics_bytes.decode('utf-8')) # Guardarlo a fichero para inspección with open('/tmp/test_calendario.ics', 'wb') as f: f.write(ics_bytes) print(f"\nGuardado en /tmp/test_calendario.ics") print("Puedes importarlo en Google Calendar, Outlook o Apple Calendar.")