from json import dumps
from kafka import KafkaProducer
from loguru import logger
from scripts.constants.app_configuration import Kafka


def insert_values_dalmia(ts, my_dict):
    kairos_writer = KairosWriter()
    kairos_writer.write_data(
        {
            ts: my_dict
        },
        Kafka.kafka_topic
    )
    logger.info("Data pushed successfully!")


class KafkaProducerUtil:
    def __init__(self):
        try:
            self.host = Kafka.kafka_host
            self.port = Kafka.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


class KairosWriter(KafkaProducerUtil):

    def write_data(self, data_json, topic):
        site_id = "site_107"
        logger.debug(f"Data being pushed to kafka topic: {topic}")
        msg_counter = 0
        for k, v in data_json.items():
            timestamp, data = self.data_validator(k, v)
            timestamp = timestamp * 1000
            write_json = {
                "data": data,
                "site_id": site_id,
                "gw_id": "gw_{}".format(site_id.lstrip("site_")),  # The lstrip(s) removes leading whitespace (on the left)             
                "pd_id": "pd_{}".format(site_id.lstrip("site_")),  # The rstrip(s) removes the trailing whitespace (on the right)
                "timestamp": timestamp,
                "msg_id": msg_counter,
                "partition": "",
                "retain_flag": False
            }
            logger.debug(f"Timestamp: {timestamp}, Values: {data}")
            self.publish(topic, dumps(write_json))
            msg_counter += 1

        return msg_counter

    def audit_data(self, data_json, topic):
        logger.debug(f"Audit Data being pushed to kafka topic: {topic}")
        msg_counter = len(data_json)
        for each in data_json:
            self.publish(topic, dumps(each))
        return msg_counter

    @staticmethod
    def data_validator(timestamp, data):
        logger.debug("Validating the data to remove Nan values")
        __temp__ = {}
        for k, v in data.items():
            if not k.startswith("site"):
                continue
            if isinstance(v, (int, float)) and str(v) not in ('nan', 'inf'):  # This function will return True if the "v" is one of the types in the tuple.
                __temp__[k] = v
        return int(timestamp), __temp__
