"""Spotify Data Ingestion Workflow."""
from collections import defaultdict
import copy
from datetime import date as date_module
from datetime import datetime
from multiprocessing.pool import Pool
import os
from tempfile import NamedTemporaryFile
from typing import Dict
import uuid

import boto3
from boto3.exceptions import S3UploadFailedError
from boto3.s3.transfer import TransferConfig
from botocore.exceptions import ClientError
from garcon import task
from garcon_contrib.aws.garcon_s3 import remove_files_from_path
from garcon_contrib.dynamo_feed_status import garcon_feed_status
from garcon_contrib.dynamo_feed_status.garcon_feed_status import \
    STATUS_DOWNLOADED
from requests import HTTPError
from requests import RequestException
from requests.exceptions import RetryError

from feed_ingestion import logger
from feed_ingestion.conf.config import BOTO3_CONFIG, merge_configs
from feed_ingestion.flows.helpers import get_sf_config
from feed_ingestion.flows.helpers import handle_http_download_error
from feed_ingestion.flows.spotify import config, smart_downloader
from feed_ingestion.flows.spotify.snowflake_executor import Spotify
from feed_ingestion.flows.spotify.spotify_api import SpotifyAPI
from feed_ingestion.tasks import assert_valid_str_bool, check_status, \
    overall_status_tasks
from feed_ingestion.tasks.load_raw_table_tasks_sf import TASK_ID
from feed_ingestion.tasks.s3_tasks import copy_s3_key
from feed_ingestion.util import task_status
from feed_ingestion.util.aws.s3 import get_list_of_files_and_directories
from feed_ingestion.util.context_util import get_context_values
from feed_ingestion.util.log_status import log_feed_ingestion_completed_status
from feed_ingestion.util.sentry_util import send_error_or_warning

_RETENTION_POLICY_ERROR_MESSAGE = (
    'The requested reload date {context_date} is out of retention policy: '
    'it is {current_days} days from current one, but it must be not more '
    'than {retention_days} days'
)
STOP_RESPONSE = {'stop': True}


@task.decorate(timeout=1000)
def check_date(activity, date, licensor, use_s3):
    """Check if it's possible to ingest a date.

    Because of Spotify user_id issue it's impossible to backfill anything
    before 21/02/2019 without losing new user ids.
    If date is earlier than 21/02/2019 then stop the flow.

    Args:
        activity (ActivityWorker): The Garcon activity worker.
        date (str): Reporting date (YYYY-MM-DD).
        licensor (str): The lisensor.
    """
    date = date or date_module.today().strftime('%Y-%m-%d')
    date_obj = datetime.strptime(date, '%Y-%m-%d').date()
    assert licensor in config.spotify_api_licensors, \
        f'unsupported licensor "{licensor}"'

    days_from_current = (date_module.today() - date_obj).days
    if days_from_current > config.spotify_api_retention_days and not use_s3:
        error_message = _RETENTION_POLICY_ERROR_MESSAGE.format(
            context_date=date,
            current_days=days_from_current,
            retention_days=config.spotify_api_retention_days,
        )
        activity.logger.error(error_message)
        raise ValueError(error_message)
    return {}


@task.decorate(timeout=1000)
def check_feed_status(activity, date, reload, licensor):
    """Check and reset feed status if it is needed.

    Args:
        activity (ActivityWorker): The Garcon activity worker.
        date (str): Reporting date (YYYY-MM-DD).
        reload (str or None): If 'True' delete all feed statuses in DynamoDB.
        licensor (str): The lisensor.

    Returns:
        dict: {} or STOP_RESPONSE from check_ingested_status decorator
    """
    overall_feed_name = '_'.join([config.feed_name, licensor])

    if reload == 'True':
        activity.logger.info('Delete status for feed: {} {} '.format(
            overall_feed_name, date))
        garcon_feed_status.delete_status(overall_feed_name, date)
    else:
        overall_status = garcon_feed_status.get_overall_status(
            overall_feed_name, date)

        if overall_status == garcon_feed_status.STATUS_INGESTED:
            return STOP_RESPONSE
    return {'feed_name': overall_feed_name}


@task.decorate(timeout=1000)
def bootstrap(activity, date, reload, reports, licensor, use_s3,
              use_partitioned='True'):
    """Bootstrap workflow by injecting initial context from config.

    Args:
        activity (ActivityWorker): The Garcon activity worker.
        date (str): Reporting date (YYYY-MM-DD).
        reload (str or None): If 'True' delete all feed statuses in DynamoDB.
        reports (str or None): List of the reports to ingest (optional).
        licensor (str): The lisensor.
        use_s3 (str or None): If 'True' use s3 instead of Spotify API.
        use_partitioned (str or None): If 'True' use Partitioned Spotify API.
    Returns:
        dict: Initial context for the workflow.
    """
    use_partitioned = use_partitioned or 'True'
    assert_valid_str_bool(use_partitioned)
    date = date or date_module.today().strftime('%Y-%m-%d')
    parsed_date = datetime.strptime(date, '%Y-%m-%d')
    reports_list = get_context_values(reports, list(config.reports.keys()))
    common_reports = list(config.common_reports)
    dimension_tables = copy.deepcopy(config.dimension_tables)

    reports_status_names = defaultdict(dict)
    archive_paths = defaultdict(dict)
    temp_staging_raw_names = defaultdict(dict)
    drop_paths = defaultdict(dict)

    activity.logger.info('Bootstrap: {} {} {}'.format(
        config.feed_name, licensor, date))

    for report_name in set(reports_list + common_reports):
        report_feed_name = _get_feed_name(licensor, report_name)
        if reload == 'True' and report_name in reports_list:
            garcon_feed_status.delete_status(report_feed_name, date)

        if not check_report(report_name, report_feed_name, date):
            continue

        reports_status_names[report_name] = report_feed_name

        archive_paths[report_name] = config.s3['archive_path'].format(
            report_name=report_name, date=parsed_date, licensor=licensor)

        drop_paths[report_name] = config.s3['drop_path'].format(
            report_name=report_name, date=parsed_date, licensor=licensor)

        temp_staging_raw_names[report_name] = (
            config.temp_staging_raw_table.format(
                date=date.replace('-', ''),
                report=report_name,
                licensor=licensor))

    facts_feed_name = '_'.join(
        [config.feed_name, licensor, config.fact_analytics_report])
    aggregated_feed_name = '_'.join(
        [config.feed_name, licensor, config.aggregated_report])
    return {
        'date': date,
        'facts_feed_name': facts_feed_name,
        'aggregated_feed_name': aggregated_feed_name,
        'date_as_in_uuid': date,
        'archive_paths': archive_paths,
        'use_partitioned': use_partitioned,
        'reports_status_names': reports_status_names,
        'temp_staging_raw_names': temp_staging_raw_names,
        'licensor': licensor,
        'dimension_tables': dimension_tables,
        'use_s3': use_s3,
        'drop_paths': drop_paths,
        'jenkins_config': config.jenkins_config
    }


@task.decorate(timeout=config.SMART_DOWNLOADER_TIMEOUT_SECONDS + 60 * 3)
@check_status(task_id='grab_drop_files')
def grab_drop_files_partitioned(
        activity, feed_name, date, report_name, archive_path, licensor):
    """Upload partitioned report files to s3 location using Smart Downloader.

    Args:
        activity (ActivityWorker): The activity worker.
        feed_name (str): Feed name of workflow execution for status updates.
        date (str): Reporting date (YYYY-MM-DD).
        report_name (str): Name of the report to ingest.
        archive_path (str): Archive path on S3 for licensor and report.
        licensor (str): Name of licensor for downloading files.
    """
    s3_path = 's3://{bucket}/{archive_path}'.format(
        bucket=config.data_bucket, archive_path=archive_path)
    remove_files_from_path(activity, s3_path, False)

    licensor_config = config.spotify_api_credentials[licensor]
    spotify_api = SpotifyAPI(
        client_id=licensor_config['client_id'],
        client_secret=licensor_config['client_secret'],
        licensor_name=licensor_config['licensor'],
        version=licensor_config['version']
    )

    try:
        activity.logger.info(
            'date:{} - Download Start for {} report {}'.format(date, licensor,
                                                               report_name))
        urls_to_s3 = _prepare_urls_to_s3(
            date, report_name, s3_path, spotify_api)
        result = _download_urls_from_api_to_s3(spotify_api, urls_to_s3)
        status_value = garcon_feed_status.STATUS_DOWNLOADED
        activity.logger.info(
            'date:{} - Download Complete for {} report {}'.format(date,
                                                                  licensor,
                                                                  report_name))
    except LookupError as e:
        result = {
            'stop': True,
            'message': str(e),
        }
        status_value = garcon_feed_status.STATUS_NOT_INGESTED

    overall_status_tasks.set_overall_status(
        activity=activity,
        date=date,
        feed_name=feed_name,
        status=status_value
    )

    return result


def _prepare_urls_to_s3(
        date: str,
        report_name: str,
        s3_path: str,
        spotify_api: SpotifyAPI):
    """
    Prepare mapping from Spotify Partitioned API Endpoint URL to S3 location.

    Args:
        date: str
        report_name:
        s3_path:
        spotify_api:

    Returns: dict where key is SpotifyAPI url and value is S3 path
    Raises: LookupError if no countries data available

    """
    if not s3_path.endswith('/'):
        raise ValueError(f's3_path should end with slash: {s3_path}')
    url_template = spotify_api._resource_url
    if report_name == 'aggregated_streams':
        resource_name = 'aggregatedstreams'
    else:
        resource_name = report_name
    year, month, day = date.split('-')
    url = url_template.format(
        licensor_name=spotify_api.licensor_name,
        resource_name=resource_name,
        version=spotify_api.version,
        year=year,
        month=month,
        day=day,
    )
    if config.reports[report_name].get('use_countries'):
        countries = spotify_api.get_available_countries_for_url(url)
        if not countries:
            raise LookupError(f'No countries available for {url}')
        url_to_s3 = {}
        for country in countries:
            country_url = f'{url}/{country}/partitions'
            country_s3_path = f'{s3_path}{country}/'
            url_to_s3[country_url] = country_s3_path
    else:
        partitioned_url = f'{url}/partitions'
        url_to_s3 = {
            partitioned_url: s3_path
        }
    return url_to_s3


def _create_download_tasks_for_partitions(partitions, s3_location):
    """Create DownloadTasks for SmartDownloader for given partitions.

    Args:
        partitions: list of SpotifyAPI partitions
        s3_location: s3 path key should ends with slash "/"

    Returns: None
    """
    assert s3_location.endswith('/'), s3_location
    tasks = []
    for partition in partitions:
        source_url = partition['uri']
        filename = partition['description']
        destination_url = f'{s3_location}{filename}'
        task = smart_downloader.DownloadTask(
            source_url=source_url,
            destination_url=destination_url,
        )
        tasks.append(task)
    return tasks


def _download_urls_from_api_to_s3(
        spotify_api: SpotifyAPI,
        url_to_s3: Dict[str, str]):
    """Download given SpotifyAPI urls to S3 location using SmartDownloader.

    Args:
        spotify_api:
        url_to_s3: dict key is SpotifyAPI url and value is S3 path
    Returns:
        dict with key 'status' equals either 'COMPLETE' or 'ERROR'
    Raises:
        LookupError if download expectedly failed (i.e. no data yet)
        RuntimeError when data processing error occurred
    """
    job_id = f'swf-{config.feed_name}-{uuid.uuid4()}'
    logger.info(f'Downloader jobId={job_id}')

    def _gen_tasks():
        for url, s3_path in url_to_s3.items():
            try:
                partitions = spotify_api.get_partitions_for_url(url)
            except (RetryError, HTTPError) as e:
                raise LookupError(
                    f'Data not available for {url}: {str(e)}') from e
            download_tasks = _create_download_tasks_for_partitions(
                partitions=partitions,
                s3_location=s3_path
            )
            for download_task in download_tasks:
                yield download_task

    download_requests = smart_downloader.download_request_batched_generator(
        tasks=_gen_tasks(),
        batch_size=config.SMART_DOWNLOADER_BATCH_SIZE
    )
    table_name = config.SMART_DOWNLOADER_TASKS_DYNAMODB_TABLE
    n_download_requests = smart_downloader.send_download_requests(
        download_requests=download_requests,
        job_id=job_id,
        table_name=table_name,
        ttl_timeout_seconds=config.SMART_DOWNLOADER_TASKS_TTL_SECONDS,
    )
    logger.info(f'{n_download_requests} download requests have been sent.')

    try:
        states = smart_downloader.await_downloads_completion(
            job_id=job_id,
            expected_number_of_items=n_download_requests,
            table_name=table_name,
            timeout_seconds=config.SMART_DOWNLOADER_TIMEOUT_SECONDS,
        )
        n_failed_tasks = states.get('ERROR')
        if n_failed_tasks:
            raise ValueError(f'Final states: {states}')
    except (TimeoutError, ValueError) as e:
        raise RuntimeError(
            f'Download failed: {str(e)}. '
            f'Lookup DynamoDB "{table_name}" '
            f'for jobId="{job_id}"')

    return_value = {
        'jobId': job_id,
        'n_tasks': n_download_requests,
    }
    return return_value


@task.decorate(timeout=36000)
@check_status()
def grab_drop_files(
        activity, feed_name, date, report_name, archive_path, licensor):
    """Upload files to archive location.

    Args:
        activity (ActivityWorker): The activity worker.
        feed_name (str): Feed name of workflow execution for status updates.
        date (str): Reporting date (YYYY-MM-DD).
        report_name (str): Name of the report to ingest.
        archive_path (str): Archive path on S3 for licensor and report.
        licensor (str): Name of licensor for downloading files.
    """
    full_path = 's3://{bucket}/{archive_path}'.format(
        bucket=config.data_bucket, archive_path=archive_path)
    remove_files_from_path(activity, full_path, False)

    try:
        res = _grab_licensor_files(
            activity, feed_name, licensor, date,
            archive_path, report_name)

        if STOP_RESPONSE in res:
            return STOP_RESPONSE
    except RequestException as e:
        return _handle_connection_error(activity, feed_name, report_name,
                                        date, e)

    overall_status_tasks.set_overall_status(
        activity, date, feed_name, garcon_feed_status.STATUS_DOWNLOADED)


def _grab_licensor_files(activity, feed_name, licensor, date, archive_path,
                         report_name):
    """Download files for a specific licensor.

    Args:
        feed_name (str): Feed name of workflow execution for status updates.
        licensor (str): Name of licensor for downloading files.
        date (str): Reporting date (YYYY-MM-DD).
        archive_path (str): Archive path on S3.
        report_name (str): Name of the report to ingest.
    Returns:
        list: List of download results.
    """
    if not config.reports[report_name].get('use_countries', False):
        return [_grab_resource_wrapper(
            activity, feed_name, licensor, archive_path, date, report_name)]
    else:
        args = [
            (activity, feed_name, licensor,
             archive_path, date, report_name, country)
            for country in config.countries]
        with Pool(processes=config.pool_size) as pool:
            res = pool.starmap_async(_grab_resource_wrapper, args)
            pool.close()
            pool.join()
            res = res.get()
        return res


def _grab_resource_wrapper(
        activity, feed_name, licensor,
        archive_path, date, report_name, country=None):
    """Grab a single resource.

    Args:
        feed_name (str): Feed name of workflow execution for status updates.
        licensor (str): Name of licensor for downloading files.
        archive_path (str): Archive path on S3.
        date (str): Reporting date (YYYY-MM-DD).
        report_name (str): Name of the report to ingest.
        country (str): Country name.
    Returns:
        None | STOP_RESPONSE: Returns None if download was successful
            or STOP_RESPONSE otherwise.
    """
    spotify_api = SpotifyAPI(
        config.spotify_api_credentials[licensor]['client_id'],
        config.spotify_api_credentials[licensor]['client_secret'],
        config.spotify_api_credentials[licensor]['licensor'],
        config.spotify_api_credentials[licensor]['version'])
    try:
        _grab_resource(
            activity, feed_name, spotify_api,
            archive_path, date, report_name, country)
    except HTTPError as e:
        if (country and country == 'JP' and e.response.status_code == 404
                and licensor.lower() in ('smejp', 'smejpintl')):
            return _handle_download_error(activity, feed_name, report_name,
                                          date, e)

        if (country and country != 'JP' and e.response.status_code == 404
                and licensor.lower() in ('smejp', 'smejpintl')):
            return

        if licensor.lower() == 'smecharity' and e.response.status_code == 404:
            return
        if (country and e.response.status_code == 404
                and country not in config.expected_countries):
            return
        else:
            return _handle_download_error(activity, feed_name, report_name,
                                          date, e)


@task.decorate(timeout=36000)
@check_status(task_id='grab_drop_files')
def grab_drop_files_from_s3(
        activity, feed_name, date, report_name, archive_path, drop_path):
    """Upload files to archive location.

    Args:
        activity (ActivityWorker): The activity worker.
        feed_name (str): Feed name of workflow execution for status updates.
        date (str): Reporting date (YYYY-MM-DD).
        report_name (str): Name of the report to ingest.
        archive_path (str): Archive path on S3 for licensor and report.
        drop_path (str): Archive path on S3 for licensor and report.
    """

    def copy_file(filename):
        """Copy S3 object to new path."""
        from_path = '{s3_path}{file_name}'.format(
            s3_path=s3_drop_path, file_name=filename)
        to_path = '{s3_path}{file_name}'.format(
            s3_path=full_path, file_name=filename)
        activity.logger.info('Starting to copy file {}'.format(from_path))

        try:
            copy_s3_key(from_path, to_path)
        except ClientError:
            return STOP_RESPONSE

    full_path = 's3://{bucket}/{archive_path}'.format(
        bucket=config.data_bucket, archive_path=archive_path)
    remove_files_from_path(activity, full_path, return_deleted_files=False)

    s3_drop_path = 's3://{bucket}/{drop_path}'.format(
        bucket=config.drop_bucket, drop_path=drop_path)

    if not config.reports[report_name].get('use_countries', False):
        filename = '{report_name}_{date}.gz'.format(
            report_name=report_name, date=date.replace('-', ''))

        if copy_file(filename) == STOP_RESPONSE:
            return STOP_RESPONSE

    else:
        files = get_list_of_files_and_directories(s3_drop_path)
        if not files:
            return STOP_RESPONSE
        for file_path in files:
            filename = os.path.basename(file_path)

            if copy_file(filename) == STOP_RESPONSE:
                return STOP_RESPONSE

    overall_status_tasks.set_overall_status(
        activity, date, feed_name, garcon_feed_status.STATUS_DOWNLOADED)


@task.decorate(timeout=600)
def check_available_reports(
        activity, date, reports_status_names, licensor=None):
    """Return new reports which are available.

    Args:
        date (str): Reporting date (YYYY-MM-DD).
        reports_status_names (dict): List of reports,
            which should be processed.
        licensor (str): Name of the licensor.
    Returns:
        dict: of reports and their statuses, which files were uploaded on S3
            statuses of possibility loading staging_raw and facts tables.
    """

    def check_report_for_ingest(report_name, report_names):
        """Return True if report need to be ingested.

        It will return true if:
         - it is common report (users or tracks),
         - it is fact_analytics (streams) report,
         - other report, but not report from config.download_only_reports list.

         True for report from download_only_reports should not be returned,
         because it isn't needed to ingest it into snowflake, only store on s3.
        """
        return (
                (report_name in config.common_reports
                 or config.fact_analytics_report in report_names
                 or garcon_feed_status.get_overall_status(
                            report_names[report_name], date) !=
                 garcon_feed_status.STATUS_INGESTED)
                and report_name not in config.download_only_reports)

    common_reports = set(config.common_reports)

    downloaded_reports = {}
    for report_name, feed_name in reports_status_names.items():
        if task_status.is_completed_task(
                feed_name, date, 'grab_drop_files'):
            downloaded_reports[report_name] = feed_name

    available_reports = defaultdict(dict)
    # dict of activities that flow can launch now
    necessary_activities = dict()
    # stop if common reports are not available
    # or if available only common reports
    if (not common_reports.issubset(downloaded_reports.keys())
            or common_reports == downloaded_reports.keys()):
        return STOP_RESPONSE

    for report_name in downloaded_reports.keys():
        if check_report_for_ingest(report_name, downloaded_reports):
            available_reports[report_name] = downloaded_reports[report_name]

    if common_reports & available_reports.keys():
        necessary_activities['load_common_table'] = True
    # if there are reports besides common reports and download_only_reports
    if common_reports != (
            downloaded_reports.keys() - config.download_only_reports):
        necessary_activities['load_staging_raw'] = True
    if config.fact_analytics_report in downloaded_reports.keys():
        necessary_activities['load_fact_analytics'] = True

    if not available_reports:
        return STOP_RESPONSE
    activity.logger.info(
        'date:{} - Download Complete for {}'.format(date, licensor))

    log_feed_ingestion_completed_status(activity,
                                        f'{config.feed_name}_{licensor}', date,
                                        STATUS_DOWNLOADED)

    return {
        'available_reports': available_reports,
        'necessary_activities': necessary_activities}


@task.decorate(timeout=600)
def create_transitional_common_tables(
        activity, feed_name, date, report_name, transitional_temp_table,
        sfdb_params):
    """Create transitional tables for common report.

    Args:
        activity (ActivityWorker): The Garcon activity worker.
        feed_name (str): Feed name of workflow execution for status updates.
        date (str): Reporting date (YYYY-MM-DD).
        report_name (str): Name of the report to ingest.
        transitional_temp_table (str): The transitional table name for report.
        sfdb_params (dict): Dict with params to optionally override default
            ones (Snowflake db and schema name).
    """
    if task_status.is_completed_task(feed_name, date, 'load_common_tables'):
        activity.logger.info(
            'load_common_tables for {date} for {report_name} '
            'already complete'.format(date=date, report_name=report_name))
        return

    sf_config = merge_configs(get_sf_config(config.secrets_path), sfdb_params)
    with Spotify(sf_config) as executor:
        executor.create_transitional_temp_staging_raw_table(
            report_name, transitional_temp_table)
        activity.logger.info('{table} was created'.format(
            table=transitional_temp_table))


@task.decorate(timeout=7200)
@check_status()
def load_common_tables(
        activity, date, feed_name, report_name, temp_staging_raw_table,
        transitional_temp_table, staging_raw_table, licensor, sfdb_params):
    """Load staging raw tables for common reports.

    Args:
        activity (ActivityWorker): The Garcon activity worker.
        date (str): Reporting date (YYYY-MM-DD).
        feed_name (str): Feed name of workflow execution for status updates.
        report_name (str): Name of the report to ingest.
        temp_staging_raw_table (dict): The temporary staging raw table name.
        transitional_temp_table (str): The transitional table name for report.
        staging_raw_table (str): The staging raw table name.
        licensor (str): Name of the licensor to ingest.
        sfdb_params (dict): Dict with params to optionally override default
            ones (Snowflake db and schema name).
    """
    sf_config = merge_configs(get_sf_config(config.secrets_path), sfdb_params)
    with Spotify(sf_config) as executor:
        executor.load_transitional_common_tables(
            date, temp_staging_raw_table, transitional_temp_table,
            staging_raw_table, report_name, licensor)
        activity.logger.info('{table} was loaded'.format(
            table=transitional_temp_table))
        executor.load_common_staging_raw_table(
            date, transitional_temp_table, staging_raw_table,
            report_name, licensor)
        activity.logger.info('{table} was loaded'.format(
            table=transitional_temp_table))
        executor.drop_table(transitional_temp_table)
        activity.logger.info('{} was dropped'.format(
            transitional_temp_table))


@task.decorate(timeout=14000)
@check_status()
def load_aggregated_skips_and_saves(
        activity, date, feed_name, sfdb_params, licensor):
    """Load aggregated_skips_and_saves.

    Args:
        activity (ActivityWorker): The activity worker.
        date (str): Date of the data being process (YYYY-MM-DD).
        feed_name (str): Name of the feed to get executor class.
        sfdb_params (dict): Dict with params to optionally override default
            ones (Snowflake db and schema name).
        licensor (str): The licensor.
    """
    # check that data for aggregated_streams was already loaded
    required_task_id = 'set_overall_status_POPULATED_RAW_TABLE'
    if not task_status.is_completed_task(feed_name, date, required_task_id):
        return STOP_RESPONSE

    task_id = '{}_load_aggregated_skips_and_saves'.format(licensor)
    sf_config = merge_configs(get_sf_config(config.secrets_path), sfdb_params)
    with Spotify(sf_config) as executor:
        executor.delete_from_aggregated_skips_and_saves(
            date, licensor)
        activity.logger.info(
            'Data was deleted from load_aggregated_skips_and_saves table '
            'for {}'.format(licensor))
        executor.load_aggregated_skips_and_saves(date, licensor)
        activity.logger.info(
            'load_aggregated_skips_and_saves table was loaded '
            'for {}'.format(licensor))

    activity.logger.info(
        'Task {task_id} of {feed_name} for {date} completed'.format(
            task_id=task_id, feed_name=feed_name, date=date))


@task.decorate(timeout=600)
def drop_temp_table(activity, temp_table_name, sfdb_params):
    """Drop temp staging raw table.

    Args:
        activity (ActivityWorker): The Garcon activity worker.
        temp_table_name (str): Table name to drop.
        sfdb_params (dict): Dict with params to optionally override default
            ones (Snowflake db and schema name).
    """
    sf_config = merge_configs(get_sf_config(config.secrets_path), sfdb_params)
    with Spotify(sf_config) as executor:
        executor.drop_table(temp_table_name)
    activity.logger.info('{} was dropped'.format(temp_table_name))


@task.decorate(timeout=600)
def set_status_ingested(activity, date, reports_status_names):
    """Set report status INGESTED for reports if they have been taks completed.

    Download_only_report is considered completed if the status is DOWNLOADED.
    Not fact_analytics (streams) reports are considered completed if the status
    is POPULATED_RAW_TABLE.
    For fact_analytics (streams) report status is updated to INGESTED after
    loading into fact_analytics is done.

    Args:
        activity (ActivityWorker): The Garcon activity worker.
        date (str): Reporting date (YYYY-MM-DD).
        reports_status_names (dict): List of reports,
            which should be processed.
    """

    def check_report_has_been_processed(report_name, overall_status):
        """Check if this report has been processed.

        It will return true if:
        - it is not fact_analytics report (streams),
            and it has been loaded to staging raw table,
        - it is report from config.download_only_reports,
            and it has been downloaded.
        """
        return (
                overall_status == garcon_feed_status.STATUS_POPULATED_RAW_TABLE
                and report_name != config.fact_analytics_report
                or
                overall_status == garcon_feed_status.STATUS_DOWNLOADED
                and report_name in config.download_only_reports
        )

    for report_name, feed_name in reports_status_names.items():
        overall_status = garcon_feed_status.get_overall_status(
            feed_name, date)
        if check_report_has_been_processed(report_name, overall_status):
            overall_status_tasks.set_overall_status(
                activity, date, feed_name, garcon_feed_status.STATUS_INGESTED)
            activity.logger.info(
                'Setting status for feed: {feed_name} with date: {date} '
                'to: {status}, via a task'.format(
                    feed_name=feed_name, date=date,
                    status=garcon_feed_status.STATUS_INGESTED))


@task.decorate(timeout=600)
def set_overall_status_ingested(activity, date, licensor):
    """Set overall feed status INGESTED if all reports are completed.

    Args:
        activity (ActivityWorker): The Garcon activity worker.
        date (str): Reporting date (YYYY-MM-DD).
        licensor (str): The licensor.
    """
    reports = dict(config.reports)
    overall_feed_name = '_'.join([config.feed_name, licensor])
    # remove users report because it is not ingested any more
    del reports['users']
    for report_name in reports:
        overall_status = garcon_feed_status.get_overall_status(
            _get_feed_name(licensor, report_name), date)
        if overall_status != garcon_feed_status.STATUS_INGESTED:
            return STOP_RESPONSE

    overall_status_tasks.set_overall_status(
        activity=activity,
        date=date,
        feed_name=overall_feed_name,
        status=garcon_feed_status.STATUS_INGESTED,
    )

    activity.logger.info(
        'Setting status for feed: {feed_name} with date: {date} '
        'to: {status}, via a task'.format(
            feed_name=overall_feed_name, date=date,
            status=garcon_feed_status.STATUS_INGESTED))

    # it is needed to add staging_raw_table_tasks to completed tasks
    # for amContent to prevent loading it into snowflake in case of soft reload
    for report_name in config.common_reports:
        task_status.mark_completed_task(
            _get_feed_name(licensor, report_name), date, TASK_ID)


@task.decorate(timeout=60)
def sns_publish_message(activity, feed_name, date, topic, message, subject):
    """Send SNS messages to specific topic w/subject & message.

    Assumes AWS_ACCESS_KEY_ID / AWS_SECRET_ACCESS_KEY are set as env vars

    Args:
        activity (ActivityWorker): The swf activity worker.
        feed_name (str): Feed name of workflow execution for status updates.
        date (str): Reporting date (YYYY-MM-DD).
        topic (str): Topic ARN (ex.arn:aws:sns:us-east-1:103233932089:dev_test)
        message (str): The message you want to send to the topic. Messages
        must be UTF-8 encoded strings and be at most 4KB in size.
        subject (str): Optional parameter to be used as the "Subject" line of
        the email notifications.
    """
    task_id = 'sns_publish_message'

    if task_status.is_completed_task(feed_name, date, task_id):
        activity.logger.info(
            'Task {task_id} of {feed_name} for {date} already complete, and '
            '"reload" flag was not passed, skipping...'.format(
                task_id=task_id, feed_name=feed_name, date=date))
        return

    if subject:
        client = boto3.client('sns', config=BOTO3_CONFIG)
        client.publish(TopicArn=topic, Message=message, Subject=subject)
        activity.logger.info('SNS report about update_dim_tables sent')

    task_status.mark_completed_task(feed_name, date, task_id)
    activity.logger.info(
        'Task {task_id} of {feed_name} for {date} completed'.format(
            task_id=task_id, feed_name=feed_name, date=date))


def _grab_resource(
        activity, feed_name, spotify_api,
        archive_path, date, file_type, country=None):
    """Download data from Spotify API and upload to S3.

    Args:
        feed_name (str): Feed name of workflow execution for status updates.
        spotify_api (SpotifyAPI): Spotify API instance.
        archive_path (str): Archive path on S3.
        date (str): Date.
        file_type (str): Spotify API resource type.
        country (str): Country for streams resource.
    """
    parsed_date = datetime.strptime(date, '%Y-%m-%d')

    with NamedTemporaryFile('wb') as file:
        _dispatch(file, spotify_api, file_type, date, country)
        try:
            filename = config.file_pattern.format(
                date=parsed_date,
                country_code='_{}'.format(country) if country else '',
                licensor=spotify_api.licensor_name,
                report_name=file_type)
            _upload_resource(archive_path, filename, file)
        except S3UploadFailedError as e:
            _handle_upload_error(activity, feed_name, archive_path, date, e)


def _dispatch(fd, spotify_api, file_type, date, country=None):
    """Dispatch Spotify API call.

    Args:
        fd (File): File descriptor of result file.
        spotify_api (SpotifyAPI): Spotify API instance.
        file_type (str): Spotify API resource type.
        date (str): Date.
        country (str): Country for streams resource.
    """
    if file_type == 'tracks':
        spotify_api.get_tracks_to_file(fd, date)
    elif file_type == 'users':
        spotify_api.get_users_to_file(fd, date)
    elif file_type == 'streams':
        spotify_api.get_streams_to_file(fd, date, country)
    elif file_type == 'sub_30_sec_streams':
        spotify_api.get_sub_30_sec_streams_to_file(fd, date, country)
    elif file_type == 'aggregated_streams':
        spotify_api.get_aggregated_streams_to_file(fd, date)


def _handle_upload_error(activity, feed_name, archive_path, date, exc):
    """Handle S3 upload Errors.

    Args:
        feed_name (str): Feed name of workflow execution for status updates.
        archive_path (str): Archive path on S3.
        date (str): Reporting date (YYYY-MM-DD).
        exc (S3UploadFailedError | HTTPError): Thrown exception.
    """
    overall_status_tasks.set_overall_status(
        activity, date, feed_name, garcon_feed_status.STATUS_NOT_AVAILABLE)
    logger.error(
        'Cannot upload file to {path}. {exception_body}'.format(
            path=archive_path, exception_body=exc))
    raise exc


def _upload_resource(archive_path, filename, fd):
    """Upload file to S3.

    Args:
       archive_path (str): Archive path on S3.
       filename (str): Filename.
       fd (File): File descriptor of file to upload.
    """
    s3 = boto3.client('s3')
    key_name = '{bucket_path}{file_name}'.format(
        bucket_path=archive_path,
        file_name=filename)
    s3.upload_file(
        fd.name, config.data_bucket, key_name, Config=TransferConfig())
    logger.info(
        '{filename} uploaded to {path}'.format(
            filename=filename, path=archive_path))


def _handle_download_error(activity, feed_name, file_type, date, exc):
    """Handle Spotify API download error.

    Args:
        feed_name (str): Feed name of workflow execution for status updates.
        date (str): Reporting date (YYYY-MM-DD).
        file_type (str): Spotify API resource type.
        exc (HTTPError): Thrown exception.
    """
    overall_status_tasks.set_overall_status(
        activity, date, feed_name, garcon_feed_status.STATUS_NOT_AVAILABLE)
    logger.error(
        'Cannot download {file_type} from Spotify API. '
        '{exception_body}'.format(
            file_type=file_type,
            exception_body=str(exc)))
    return handle_http_download_error(exc, raise_error=False)


def check_report(report_name, report_feed_name, date):
    """Check status if report is already ingested.

     Args:
        report_name (str): Name of the report to ingest.
        report_feed_name (str): Feed name for status updates.
        date (str): Reporting date (YYYY-MM-DD).
    Returns:
        (bool): True if report need to be ingested, otherwise False.
    """
    return (report_name in config.common_reports
            or garcon_feed_status.get_overall_status(
                report_feed_name, date) != garcon_feed_status.STATUS_INGESTED)


def _handle_connection_error(activity, feed_name, file_type, date, exc):
    """Handle Spotify API connection error.

    Args:
        feed_name (str): Feed name of workflow execution for status updates.
        date (str): Reporting date (YYYY-MM-DD).
        file_type (str): Spotify API resource type.
        exc (RequestException): Thrown exception.
    """
    overall_status_tasks.set_overall_status(
        activity, date, feed_name, garcon_feed_status.STATUS_NOT_AVAILABLE)
    logger.error(
        'Got {exception_code} {exeption_reason} error from Spotify API '
        'while requesting {file_type} report.'.format(
            file_type=file_type,
            exception_code=exc.response.status_code if exc.response else exc,
            exeption_reason=exc.response.reason if exc.response else ''))

    if os.environ.get('SENTRY_DSN'):
        send_error_or_warning(exc)

    return STOP_RESPONSE


def _get_feed_name(licensor, report_name):
    """Generate feed_name for specified licensor and report.

    Args:
        licensor (str): Nme of licensor.
        report_name (str): Name of report.

    Returns:
        str: feed_name.
    """
    return '_'.join([config.feed_name, licensor, report_name])