"""Script that dumps CloudSearch index data to kafka topic."""
import json
import random
import time

import boto3

from kafka import KafkaProducer

from src import config
from src.logger import logger


def get_index_schema(cs_domain_name):
    """Get CS domain indexed fields and their types."""
    client = boto3.client('cloudsearch', region_name=config.REGION)
    response = client.describe_index_fields(DomainName=cs_domain_name)
    logger.debug(f'Found {len(response["IndexFields"])} fields.')
    return {
        each['Options']['IndexFieldName']: each['Options']['IndexFieldType']
        for each in response['IndexFields']}


def get_cloudsearchdomain_client(details):
    """Get boto client for cloudsearchdomain."""
    return boto3.client(
        'cloudsearchdomain',
        region_name=config.REGION,
        endpoint_url=f'https://{details.get("search_endpoint")}')


def get_kafka_producer():
    """Get KafkaProducer instance."""
    return KafkaProducer(
        bootstrap_servers=config.CONNECT_BOOTSTRAP_SERVERS,
        security_protocol='SSL',
        retries=3, api_version_auto_timeout_ms=6000)


def get_kafka_key(cs_document_id, data, domain_details):
    """Get value for Kafka message's key."""
    if domain_details['kafka_key'] == 'cs_document_id':
        return cs_document_id  # this is always string
    # key field does not exist in data so log it as an error because we use the
    # ID as the _doc ID in Opensearch and this will cause issues with ingested data.
    if not data.get(domain_details['kafka_key']):
        logger.error(
            f"Field {domain_details['kafka_key']} is missing in CS data. "
            f'By default cs_document_id: {cs_document_id} will be used.',
            extra={'cs_data': json.dumps(data)})
        return cs_document_id
    return domain_details['kafka_key_datatype'](data.get(domain_details['kafka_key']))


def convert_to_datatype(field_value, data_type):
    """Convert CS search data to its original data type."""
    if data_type in ('text', 'literal') and type(field_value) is int:
        return str(field_value)
    if data_type in ('text', 'literal') and type(field_value) is list:
        return str(field_value[0]) if len(field_value) > 0 else ''
    if data_type in ('int',) and type(field_value) is str:
        return int(field_value)
    if data_type in ('int',) and type(field_value) is list:
        return int(field_value[0]) if len(field_value) > 0 else None
    # dates are returned as string.
    if data_type in ('date',) and type(field_value) is str:
        return str(field_value)
    if data_type in ('date',) and type(field_value) is list:
        return str(field_value[0]) if len(field_value) > 0 else None

    # literal-array, text-array are returned as it is.
    return field_value


def get_kafka_message(data, domain_details, schema):
    """Get json data for Kafka message."""
    return_data = {}
    for field_name, field_value in data.items():
        if field_name in domain_details.get('cs_fields_to_skip', []):
            continue

        data_type = schema.get(field_name, 'text')
        return_data[field_name] = convert_to_datatype(field_value, data_type)
    return return_data


def get_range_query(range_filter):
    """Get range search query."""
    query = 'matchall'
    field = range_filter.get('field')
    if not field:
        return query

    if range_filter.get('start') and range_filter.get('end'):
        query = f"{field}:[{range_filter.get('start')},{range_filter.get('end')}}}"
    elif not range_filter.get('start') and range_filter.get('end'):
        query = f"{field}:[,{range_filter.get('end')}}}"
    elif range_filter.get('start') and not range_filter.get('end'):
        query = f"{field}:[{range_filter.get('start')},}}"
    return query


def export(domain_details,  schema, max_rows=None, range_filter=None):
    """Export domain data based on parameters.

    Args:
        domain_details (dict): domain level configurations.
        schema (dict): Indexed fields in schema.
        max_rows (int): the name to display alongside the Hello.
        range_filter (dict): Options to filter search data.

    """
    start = 0
    size = 10000  # max is 10000 for CS search. Cannot increase it.
    total_count = 0
    actual_max_rows = max_rows + start if max_rows else None
    cursor = 'initial'

    client = get_cloudsearchdomain_client(domain_details)
    producer = get_kafka_producer()
    kafka_topic = domain_details.get('topic')
    query = get_range_query(range_filter)
    # if we request less than 10000 rows.
    if max_rows and size > max_rows:
        size = max_rows

    logger.debug(f'query = {query}, size={size}')
    while not max_rows or start < actual_max_rows:
        logger.debug(f'start = {start}')
        response = client.search(
            query=query,
            size=size,
            cursor=cursor,
            sort=domain_details.get('sort'),
            queryParser='structured',
            returnFields='_all_fields',
        )

        hits = response.get('hits', {}).get('hit', [])
        new_cursor = response.get('hits', {}).get('cursor')
        total_count += len(hits)
        logger.debug(f'hits = {len(hits)}')
        if not hits:
            break

        for hit in hits:
            field_data = get_kafka_message(hit['fields'], domain_details, schema)
            key = get_kafka_key(hit['id'], field_data, domain_details)
            message = json.dumps(field_data)
            producer.send(
                kafka_topic,
                value=message.encode('utf-8'),
                key=key.encode('utf-8') if key else None)

        start += size
        cursor = new_cursor
        producer.flush()
        time.sleep(random.randint(1, 2))

    logger.info(f'total record count = {total_count}')