import os
import json
from kafka import KafkaConsumer

KAFKA_BROKER_URL = os.environ.get('KAFKA_BROKER_URL')
METRIC_TOPIC = os.environ.get("METRIC_TOPIC")

metric_agg = {
    "processed_per_second": [],
    "records_processed": [],
    "blacklisted_ips": [],
    "end_ts": []
}


def aggregate_stats(agg_dict: dict) -> dict:
    """
    Aggregates list of metrics into sums and averages
    to give overall performance of microservice.
    Returns dict of form:
    {
      "Records_Processed": int,
      "Avg_Records_Per_Second": float,
      "Blacklisted": int
    }

    :param agg_dict: aggregate dict of lists
    :return: dict of form above
    """
    # If data is empty, don't do anything
    if agg_dict["processed_per_second"] == []:
        return {}

    return {
        "Records_Processed": sum(agg_dict["records_processed"]),
        "Avg_Records_Per_Second":
            sum(agg_dict["processed_per_second"]) /
            len(agg_dict["processed_per_second"]),
        "Blacklisted": sum(agg_dict["blacklisted_ips"])
    }


if __name__ == '__main__':
    consumer = KafkaConsumer(
        METRIC_TOPIC,
        bootstrap_servers=KAFKA_BROKER_URL,
        value_deserializer=lambda value: json.loads(value),
        consumer_timeout_ms=15000
    )

    # Consumer metrics from kafka topic
    for message in consumer:
        for key, val in message.value.items():
            metric_agg[key].append(val)

    # When we're done, aggregate results
    # And print off metrics
    result = aggregate_stats(metric_agg)
    print("SLIDING METRICS")
    for key, val in result.items():
        print(key, ":", val)
    print()
