"""Store Specific Itunes Priority Workflow."""

from garcon import task
import numpy as np
import pandas as pd
from snowflake_connector.etl_connector import SnowflakeSQLExecutor
from snowflake_connector.etl_connector import SQLLoader

from feed_ingestion.flows.helpers import get_sf_config
from feed_ingestion.flows.itunes_priority import config
from feed_ingestion.util import datalytics_util
from feed_ingestion.util import read_sheet as gdoc
from feed_ingestion.util.aws import s3 as s3utils


STOP_RESPONSE = {'stop': True}

sql_loader = SQLLoader(__file__)


@task.decorate(timeout=1000)
def bootstrap(activity):
    """Bootstrap workflow.

    Args:
        activity (ActivityWorker): The Garcon activity worker.
    Returns:
        dict: Initial context for the workflow.
    """
    return {
        'priority_sheet_id': config.PRIORITY_SHEET_ID,
        'priority_table': config.PRIORITY_TABLE,
        's3_path': config.S3_PATH,
        'priority_email_list': config.PRIORITY_EMAIL_LIST
    }


@task.decorate(timeout=10000)
def process_files(activity, priority_sheet_id, priority_table, s3_path, aws):
    """Process files.

    Args:
        activity (ActivityWorker): The Garcon activity worker.
        priority_sheet_id (str): Id of sheet.
        priority_table (str): Name of table of appleids.
        s3_path (str): s3 path for priority_ticket_appleid.csv.
        aws (dict): AWS credentials.
    Returns:
        dict: STOP_RESPONSE or confirmation.
    """
    try:
        df_new = gdoc.sheet_2_df(
            priority_sheet_id, 'Priority List', 'A1:A', np.int64
        ).drop_duplicates(
            subset='Apple_ID'
        ).sort_values(
            by='Apple_ID', ascending=True
        ).reset_index(drop=True)
    except Exception:
        activity.logger.error('gdoc id %s not found' % priority_sheet_id)
        return STOP_RESPONSE

    try:
        df_old = s3utils.read_csv(s3_path, dtype={'Apple_ID': np.int64})
        if (isinstance(df_old, str) and
                'The specified key does not exist' in df_old):
            # File does not exist. Use empty DataFrame
            df_old = pd.DataFrame()
        else:
            df_old = (df_old.drop_duplicates(subset='Apple_ID')
                      .sort_values(by='Apple_ID', ascending=True)
                      .reset_index(drop=True))
    except Exception:
        activity.logger.error('%s not found' % priority_table)
        return STOP_RESPONSE

    if not df_new.equals(df_old):
        activity.logger.info('New data in the spreadsheet. Processing.')
        # send new priorities to s3
        s3utils.to_csv(
            df=df_new,
            s3_path=s3_path,
            index=False
        )
        priority_appleids = df_new['Apple_ID'].unique()

        sf_config = get_sf_config(config.secrets_path)
        SnowflakeSQLExecutor(sf_config).execute_query(
            sql_loader, 'truncate_table', {
                'table_name': priority_table
            }
        )
        rows_added = SnowflakeSQLExecutor(sf_config).execute_query(
            sql_loader, 'load_data', {
                'table_name': priority_table,
                'values': ','.join(
                    ["({ID},'yes')".format(ID=c_id) for c_id in
                        priority_appleids])
            }
        )
        return {'rows_added': rows_added}
    activity.logger.info('No new data in the spreadsheet.')
    return STOP_RESPONSE


@task.decorate(timeout=10000)
def send_emails(activity, priority_table, priority_email_list, rows_added):
    """Process files found on S3.

    Args:
        activity (ActivityWorker): The Garcon activity worker.
        priority_table (str): Name of table of appleids.
        priority_email_list (list): List of emails to send notification to.
        rows_added (int): Number of rows processed.
    """
    BODY = ('{N} apple ids added to {T}'.format(
        N=rows_added, T=priority_table))

    ses_response = datalytics_util.send_emails(
            to=priority_email_list,
            sub='Priority Apple Ids Updated.',
            body=BODY
        )
    activity.logger.debug('SES response: {}.'.format(ses_response))