"""
Spotify Marquee Data Ingestion Workflow tasks.

Tasks to ingest data from Spotify Marquee Drop feed into
fact tables in Snowflake.
"""
from datetime import datetime
import re

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

from feed_ingestion.conf.config import merge_configs
from feed_ingestion.flows import SpotifyMarqueeSF
from feed_ingestion.flows.helpers import get_sf_config
from feed_ingestion.flows.spotify_marquee import config
from feed_ingestion.tasks import s3_tasks

sql_loader = SQLLoader(__file__)


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

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

    Returns:
        dict: Initial context of the workflow.
    """
    if not licensor:
        licensor = 'theorchard'

    activity.logger.info(
        'Bootstrapping {feed_name}...'.format(feed_name=config.feed_name))

    assert report_name in config.reports

    feed_name = '_'.join([config.feed_name, licensor, 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_archive_path = config.s3['archive'].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,
        'file_pattern': '.*{}.*'.format(re.escape(
            file_name.replace('.txt.gz', ''))),
        'report_name': report_name,
        'licensor': licensor,
        'error_limit': config.snowflake_error_limit,
    }

    source_files_dict = {
        'files': [{'file_name': file_name, 'found': True}]
    }

    return dict(
        date=date,
        file_name=file_name,
        feed_name=feed_name,
        report_name=report_name,
        licensor=licensor,
        s3_drop_path=s3_drop_path,
        s3_archive_path=s3_archive_path,
        staging_raw_table=staging_raw_table,
        temp_table_name=temp_table_name,
        sf_kwargs=kwargs,
        source_files_dict=source_files_dict
    )


@task.decorate(timeout=3000)
def grab_drop_file(
        activity, feed_name, date, file_name, s3_drop_path, s3_archive_path):
    """Copy drop file from sme S3 bucket to theorchard s3."""
    source_key_name = '{dir}{file}'.format(
        dir=s3_drop_path,
        file=file_name)

    archive_key_name = '{dir}{file}'.format(
        dir=s3_archive_path,
        file=file_name)

    source_bucket_name, source_bucket_path = garcon_s3.extract_bucket_path(
        source_key_name)
    archive_bucket_name, archive_bucket_path = garcon_s3.extract_bucket_path(
        archive_key_name)

    result = s3_tasks.copy_file(
        activity,
        source_bucket_name,
        source_key_name=source_bucket_path,
        destination_bucket_name=archive_bucket_name,
        destination_key_name=archive_bucket_path,
        replace=True
    )
    file_name_exists = result.get(file_name)

    if not file_name_exists:
        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])
        return {'stop': True}


@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)
    SpotifyMarqueeSF(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
    }
    SpotifyMarqueeSF(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,
            }
        )