"""
BandsInTown Data Ingestion Workflow tasks.

Tasks to ingest data from BandsInTown FTP Drop feed into
fact tables in Delphi Snowflake.
"""
from datetime import datetime
import io
import re

import boto3
from garcon import task
from garcon_contrib.aws.utils import garcon_s3
from garcon_contrib.dynamo_feed_status import garcon_feed_status
import smart_open
from snowflake_connector.etl_connector import SnowflakeSQLExecutor, SQLLoader

from feed_ingestion.conf.config import merge_configs
from feed_ingestion.flows import BandsInTownSF
from feed_ingestion.flows.bandsintown import config
from feed_ingestion.flows.bandsintown.gpg_util import gpg
from feed_ingestion.flows.helpers import get_sf_config
from feed_ingestion.util import task_status
from feed_ingestion.util.aws import s3 as s3utils

sql_loader = SQLLoader(__file__)


@task.decorate(timeout=600)
def bootstrap(activity, date, report_name, reload=None):
    """Bootstrap workflow by getting the correct configurations.

    Args:
        report_name:
        activity (ActivityWorker): The activity worker.
        date (str): Reporting date (YYYY-MM-DD).
        reload (str or None): If 'True' then clear all feed statuses.

    Returns:
        dict: Initial context of the workflow.
    """
    activity.logger.info(
        'Bootstrapping {feed_name}...'.format(feed_name=config.feed_name))

    assert report_name in config.reports

    feed_name = '_'.join([config.feed_name, report_name])

    if reload == 'True':
        activity.logger.info('Delete status for feed: {} {} '.format(
            feed_name, date))
        garcon_feed_status.delete_status(feed_name, date)
    else:
        overall_status = garcon_feed_status.get_overall_status(
            feed_name, date)
        if overall_status == garcon_feed_status.STATUS_INGESTED:
            message = 'already ingested for {}'.format(date)
            activity.logger.info(message)
            return {
                'feed_name': feed_name,
                'message': message,
                'stop': True
            }

    date_dt = datetime.strptime(date, '%Y-%m-%d').date()
    s3_drop_path = config.s3['drop'].format(date=date_dt)
    s3_staging_path = config.s3['staging'].format(date=date_dt)

    file_name = (
        config.reports[report_name]['file_pattern'].format(date=date_dt)
    )

    temp_table_name = config.temp_table_name.format(
        report_name=report_name,
        date=date_dt)

    staging_raw_table = config.reports[report_name]['staging_raw_table']

    kwargs = {
        'date': date,
        'report_name': report_name,
        'error_limit': config.snowflake_error_limit,
        'file_pattern': '.*{}.*'.format(re.escape(
            file_name.replace('.gpg', ''))),
    }

    drop_file_pattern = rf'.*\/{file_name}'

    return dict(
        date=date,
        file_name=file_name,
        feed_name=feed_name,
        report_name=report_name,
        s3_drop_path=s3_drop_path,
        s3_staging_path=s3_staging_path,
        staging_raw_table=staging_raw_table,
        temp_table_name=temp_table_name,
        sf_kwargs=kwargs,
        drop_file_pattern=drop_file_pattern
    )


def get_source_file_content(activity, feed_name, date, s3_drop_path,
                            file_name):
    """Get source file content.

    Args:
        activity: The activity worker.
        file_name: S3 file name
        s3_drop_path: S3 path to file
        date: report date
        feed_name: feed name

    Yields:
        string: file content.
    """
    source_key = '{dir}{file}'.format(
        dir=s3_drop_path,
        file=file_name)

    file_content = None

    try:
        file_content = smart_open.smart_open(source_key, 'rb')
    except FileNotFoundError as err:
        garcon_feed_status.set_overall_status(
            feed_name, date, garcon_feed_status.STATUS_NOT_AVAILABLE)
        garcon_feed_status.set_missing_files(
            feed_name, date, [file_name])
        activity.logger.error(str(err))

    return file_content


def decrypt(content):
    """Decrypt file content.

    Args:
        content (str): encrypted file content.

    Yields:
        string: decrypted file content.
    """
    return gpg.decrypt_file(content, passphrase=config.GPG_PASSPHRASE)


@task.decorate(timeout=3000)
def grab_drop_file(
        activity, feed_name, date, file_name, s3_drop_path, s3_staging_path):
    """Obtain drop file from sftp, decrypt and stage for ingestion.

    Args:
        activity: The activity worker.
        feed_name: Feed name
        date: report date
        file_name: S3 file name
        s3_drop_path: S3 path to file
        s3_staging_path: S3 path to file

    Yields:
        dict: staging_file_name
    """
    content = get_source_file_content(
        activity, feed_name, date, s3_drop_path, file_name
    )

    if not content:
        return {'stop': True}

    activity.logger.info(f'Content was downloaded. File: {file_name}')

    decrypted_content = decrypt(content)

    activity.logger.info(f'Content was decrypted. File: {file_name}')

    staging_file_name = file_name.replace('.gpg', '.gz')

    staging_key_name = '{dir}{file}'.format(
        dir=s3_staging_path,
        file=staging_file_name)

    staging_bucket_name, staging_file_path = garcon_s3.extract_bucket_path(
        staging_key_name)

    s3utils.delete_s3_obj(staging_bucket_name, staging_file_path)

    s3 = boto3.client('s3')

    s3.upload_fileobj(
        io.BytesIO(decrypted_content.data),
        staging_bucket_name,
        staging_file_path,
        ExtraArgs={'ExpectedBucketOwner': config.expected_bucket_owner}
    )

    return dict(
        staging_file_name=staging_file_name,
    )


@task.decorate(timeout=1000)
def load_staging_raw_table(
        activity, date, temp_staging_raw_table,
        filename, sfdb_params, staging_raw_table, report_name,
        licensor, secrets_path=None):
    """Load data to staging_raw_table .

    Runs for each vendor separately.

    Args:
        activity (ActivityWorker): The garcon activity worker.
        date (str): YYYY-MM-DD date of rows to delete from staging_raw table.
        temp_staging_raw_table (str): Temporary staging raw table.
        filename (str): Name of source filename.
        sfdb_params (dict): Dictionary stores Snowflake params.
        staging_raw_table (str): The staging raw table name.
        report_name (str): Name of report being ingested.
        licensor (str): Name of the licensor to ingest.
        secrets_path (str): Secrets manager path of the flow.
    """
    activity.logger.info('Deleting {} from {}'.format(
        date, staging_raw_table))

    sf_config = merge_configs(get_sf_config(secrets_path), sfdb_params)
    BandsInTownSF(sf_config).execute_query(
        sql_loader, 'delete_from_staging_raw', {
            'db': sf_config['db'],
            'schema': sf_config['schema'],
            'staging_raw_table': staging_raw_table,
            'date': date,
        }
    )
    activity.logger.info('Loading {} into {}'.format(
        date, staging_raw_table))
    params = {
        'licensor': licensor,
        'filename': filename,
        'download_date': date
    }
    BandsInTownSF(sf_config).load_staging_raw_table(
        date,
        staging_raw_table,
        temp_staging_raw_table=temp_staging_raw_table,
        report_name=report_name,
        **params)


@task.decorate(timeout=7200)
def drop_temp_table(activity, temp_table_name, sfdb_params, secrets_path=None):
    """Drop temporary table.

    Args:
        activity (ActivityWorker): The garcon activity worker.
        temp_table_name (str): name of table to drop.
        sfdb_params (dict): Dictionary stores Snowflake params.
        secrets_path (str): Secrets manager path of the flow.
    """
    activity.logger.info('Dropping temp table {}'.format(temp_table_name))
    sf_config = merge_configs(get_sf_config(secrets_path), sfdb_params)
    with SnowflakeSQLExecutor(sf_config) as executor:
        executor.execute_query(
            sql_loader, 'drop_temp_table', {
                'db': sf_config['db'],
                'schema': sf_config['schema'],
                'table_name': temp_table_name,
            }
        )


@task.decorate(timeout=600)
def mark_ingested_file(activity, feed_name, date, source_file):
    """Put list of injected file in DynamoDb.

    Args:
        activity (ActivityWorker): The activity worker.
        feed_name (str): Feed name of workflow execution for status updates.
        date (str): Reporting date (YYYY-MM-DD).
        source_file (str): file name.
    """
    activity.logger.info(
        f'Mark ingested file for {feed_name} {date}: \n {source_file}')
    task_status.set_values(feed_name, date, 'ingested_files', [source_file])


@task.decorate(timeout=600)
def delete_staging_file(activity, feed_name, date, s3_staging_path, file_name):
    """Delete staging file.

    Args:
        activity (ActivityWorker): The activity worker.
        feed_name (str): Feed name of workflow execution for status updates.
        date (str): Reporting date (YYYY-MM-DD).
        s3_staging_path (str): S3 path to staging file.
        file_name (str): File name of staging file.
    """
    staging_file_name = file_name.replace('.gpg', '.gz')
    staging_key_name = '{dir}{file}'.format(
        dir=s3_staging_path,
        file=staging_file_name)

    activity.logger.info(
        f'Deleting staging file for {feed_name} {date}: \n {staging_file_name}'
    )
    staging_bucket_name, staging_file_path = garcon_s3.extract_bucket_path(
        staging_key_name)

    s3utils.delete_s3_obj(staging_bucket_name, staging_file_path)


@task.decorate(timeout=3600)
def delete_old_snapshots(
        activity, staging_raw_table, sfdb_params, secrets_path=None):
    """Delete old snapshots from staging raw table, keeping only last 5 days.

    Args:
        activity (ActivityWorker): The garcon activity worker.
        staging_raw_table (str): The staging raw table name.
        sfdb_params (dict): Dictionary stores Snowflake params.
        secrets_path (str): Secrets manager path of the flow.
    """
    activity.logger.info(
        f'Deleting old snapshots from {staging_raw_table}')

    sf_config = merge_configs(get_sf_config(secrets_path), sfdb_params)
    BandsInTownSF(sf_config).delete_old_snapshots(staging_raw_table)

    activity.logger.info(
        f'Successfully deleted old snapshots from {staging_raw_table} table')