67 lines
1.8 KiB
Python
67 lines
1.8 KiB
Python
import json
|
|
import logging
|
|
from time import sleep
|
|
from kafka import KafkaConsumer
|
|
|
|
# Configuración del registro
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(levelname)s - %(message)s'
|
|
)
|
|
|
|
|
|
def main():
|
|
|
|
# Nombre del tema de Kafka del que se consumen los mensajes
|
|
parsed_topic_name = 'parsed_recipes'
|
|
|
|
# Umbral de calorías para la notificación
|
|
calories_threshold = 200
|
|
|
|
# Crear un consumidor de Kafka
|
|
consumer = KafkaConsumer(
|
|
parsed_topic_name,
|
|
auto_offset_reset='earliest', # Inicia desde el 1er mensaje si es nuevo
|
|
bootstrap_servers=['kafka:9092'], # Servidor Kafka
|
|
api_version=(0, 10), # Versión de la API de Kafka
|
|
# Tiempo máx. de espera mensajes (milisegundos)
|
|
consumer_timeout_ms=2000
|
|
)
|
|
|
|
logging.info('[+] Iniciando el consumidor de notificaciones...')
|
|
|
|
try:
|
|
|
|
for msg in consumer:
|
|
|
|
# Decodificar el mensaje de JSON
|
|
record = json.loads(msg.value)
|
|
|
|
# Obtener el valor de calorías
|
|
calories = int(record.get('calories', 0))
|
|
title = record.get('title', 'Sin título') # Obtener el título
|
|
|
|
# Verificar si las calorías exceden el umbral
|
|
if calories > calories_threshold:
|
|
logging.warning(
|
|
f'[!] Alerta: {title} tiene {calories} calorías')
|
|
|
|
# Esperar 5 segundos antes de procesar el siguiente mensaje
|
|
sleep(5)
|
|
|
|
except Exception as ex:
|
|
|
|
logging.error(
|
|
'[!] Error en el consumidor de notificaciones', exc_info=True)
|
|
|
|
finally:
|
|
|
|
if consumer is not None:
|
|
consumer.close() # Cerrar el consumidor al finalizar
|
|
logging.info('[i] Consumidor cerrado.')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
|
|
main()
|