"""Docstring."""

import json
import os

import boto3

from confluent_kafka import Producer


def main():
    """Entrypoint."""
    try:
        cluster_name = os.environ['CLUSTER_NAME']
        topic_name = os.environ['TOPIC_NAME']
        os.environ['AWS_PROFILE']
    except KeyError as e:
        print(f'Set {e} in .env')
        exit(1)

    # read in events data and validate
    events = []
    try:
        input_dir = 'inputs'
        for events_file in os.listdir(input_dir):
            if events_file.endswith('.json'):
                with open(os.path.join(input_dir, events_file), 'rb') as f:
                    events += json.load(f)
    except FileNotFoundError:
        print(f'Unable to open "{events_file}"')
        exit(1)
    except json.decoder.JSONDecodeError:
        print(f'Unable to read "{events_file}" as valid json')
        exit(1)
    if not events:
        print('No events loaded from input directory')
        exit(1)
    if not type(events) == list:
        print('Events must be formatted as list')
        exit(1)

    # boto client
    msk_client = boto3.client('kafka')

    # find cluster by name
    list_response = msk_client.list_clusters_v2(
        ClusterNameFilter=cluster_name
    )
    if list_response['ResponseMetadata']['HTTPStatusCode'] != 200:
        print('Unexpected response from "list_cluster_v2"')
        print(list_response)
        exit(1)
    cluster = None
    try:
        cluster = [
            x for x in list_response['ClusterInfoList']
            if x['ClusterName'] == cluster_name
        ][0]
    except IndexError:
        print(f'No cluster found with name {cluster_name}')
        exit(1)

    # find broker servers by ARN
    broker_response = msk_client.get_bootstrap_brokers(
        ClusterArn=cluster['ClusterArn']
    )
    if broker_response['ResponseMetadata']['HTTPStatusCode'] != 200:
        print('Unexpected response from "get_bootstrap_brokers"')
        print(broker_response)
        exit(1)
    broker_servers = broker_response['BootstrapBrokerStringTls']

    # setup producer
    p = Producer(
        {
            'bootstrap.servers': broker_servers,
            'message.timeout.ms': 3000,
            'security.protocol': 'SSL'
        }
    )

    # add events to topic
    print(f'Delivering {len(events)} to {topic_name} on {cluster_name}...')

    def report(err, msg):
        if err is not None:
            print(('Message delivery failed: {}'.format(err)))
        else:
            print('Message delivered to {}-{}'.format(msg.partition(), msg.offset()))  # noqa:E501
        print(msg.value())
    for event in events:
        data = json.dumps(event)
        p.produce(topic_name, data.encode('utf-8'), callback=report)
    p.flush()


if __name__ == '__main__':
    main()