from json import dumps

from kafka import KafkaProducer
from loguru import logger

from scripts.constants.app_configuration import KAFKA_HOST, KAFKA_PORT, KAFKA_TOPIC


class KafkaProducerUtil:
    def __init__(self):
        try:
            self.host = KAFKA_HOST
            self.port = KAFKA_PORT
            logger.debug(f"Connecting to Kafka with details: {self.host}, {self.port}")
            kafka_broker = [self.host + ":" + str(self.port)]
            self.producer = KafkaProducer(
                bootstrap_servers=kafka_broker,
                value_serializer=lambda v: v.encode('utf-8'),
                api_version=(0, 10, 1))
            self.producer.flush()
        except Exception as e:
            logger.error(f"Kafka connection error: {e}")

    def publish(self, topic, data):
        try:
            kafka_response = self.producer.send(topic, data)
            self.producer.flush()
            logger.debug(f" Message sent to kafka with response: {kafka_response}")
            return True
        except Exception as e:
            logger.error(e)
            return False
