"""Chartmetric Charts Ingestion Workflow."""

from garcon.param import StaticParam
from garcon_contrib.dynamo_feed_status import garcon_feed_status as feed_status

from feed_ingestion.flows import base
from feed_ingestion.flows.chartmetric_charts import config
from feed_ingestion.flows.chartmetric_charts import generators
from feed_ingestion.flows.chartmetric_charts import tasks


class Flow(base.FlowBase, base.FlowConfigMixin):
    """Class representing the workflow."""

    def __init__(self):
        """Initialize flow object."""
        super(Flow, self).__init__(feed_name=config.feed_name, version='0.1')
        self.timeout = 18600

    def decider(self, schedule, context):
        """Activity decider.

        Args:
            schedule (callable): The scheduler method.
        """
        bootstrap = schedule('bootstrap', self.bootstrap)
        # define which flow to process:
        skip_update_relations = context.get('skip_update_relations') == 'True'

        clear_log_table = schedule(
            'clear_log_table',
            self.clear_log_table,
            requires=[bootstrap])

        create_staging_fact = schedule(
            'create_staging_fact',
            self.create_staging_fact,
            requires=[clear_log_table])

        previous_task = create_staging_fact
        if len(list(
                set(bootstrap.result.get('bootstrap.platform_names')) & set(
                    config.platforms_to_check_if_chart_is_changed))) != 0:
            find_unchanged_charts = schedule(
                'find_unchanged_charts',
                self.find_unchanged_charts,
                requires=[create_staging_fact])
            previous_task = find_unchanged_charts

        load_staging_fact = schedule(
            'load_staging_fact',
            self.load_staging_fact,
            requires=[previous_task])

        update_isrc_to_label_participant_mapping = schedule(
            'update_isrc_to_label_participant_mapping',
            self.update_isrc_to_label_participant_mapping,
            requires=[load_staging_fact])

        update_upc_to_label_participant_mapping = schedule(
            'update_upc_to_label_participant_mapping',
            self.update_upc_to_label_participant_mapping,
            requires=[update_isrc_to_label_participant_mapping])

        previous_task = update_upc_to_label_participant_mapping

        if not skip_update_relations:

            update_globalsoundrecording_relations_kafka = schedule(
                'update_globalsoundrecording_relations_kafka',
                self.update_globalsoundrecording_relations_kafka,
                requires=[previous_task])
            if update_globalsoundrecording_relations_kafka.result.get(
                    'check_for_errors_in_dlq.stop'):
                return

            update_soundrecording_noisrc_nodes_kafka = schedule(
                'update_soundrecording_noisrc_nodes_kafka',
                self.update_soundrecording_noisrc_nodes_kafka,
                requires=[update_globalsoundrecording_relations_kafka])
            if update_soundrecording_noisrc_nodes_kafka.result.get(
                    'check_for_errors_in_dlq.stop'):
                return

            update_trackandparticipant_relations_kafka = schedule(
                'update_trackandparticipant_relations_kafka',
                self.update_trackandparticipant_relations_kafka,
                requires=[update_soundrecording_noisrc_nodes_kafka])
            if update_trackandparticipant_relations_kafka.result.get(
                    'check_for_errors_in_dlq.stop'):
                return

            update_productandlabel_relations_kafka = schedule(
                'update_productandlabel_relations_kafka',
                self.update_productandlabel_relations_kafka,
                requires=[update_trackandparticipant_relations_kafka],
                input={
                    'productandlabel_platforms_to_skip': [
                        'tiktok', 'recochoku', 'linemusic',
                        'soundcloud', 'youtube']
                })
            if update_productandlabel_relations_kafka.result.get(
                    'check_for_errors_in_dlq.stop'):
                return

            previous_task = update_productandlabel_relations_kafka

        update_staging_fact_sound_recording = schedule(
            'update_staging_fact_sound_recording',
            self.update_staging_fact_sound_recording,
            requires=[previous_task])

        update_staging_fact_public_product = schedule(
            'update_staging_fact_public_product',
            self.update_staging_fact_public_product,
            requires=[update_staging_fact_sound_recording])

        update_dim_tables = schedule(
            'update_dim_tables',
            self.update_dim_tables,
            requires=[update_staging_fact_public_product])

        load_fact_data = schedule(
            'load_fact_data',
            self.load_fact_data,
            requires=[update_dim_tables])

        update_position_change = schedule(
            'update_position_change',
            self.update_position_change,
            requires=[load_fact_data])

        refresh_aggregated_table = schedule(
            'refresh_aggregated_table',
            self.refresh_aggregated_table,
            requires=[update_position_change])

        update_dim_chart_latest_chart_date = schedule(
            'update_dim_chart_latest_chart_date',
            self.update_dim_chart_latest_chart_date,
            requires=[refresh_aggregated_table])
        previous = [update_dim_chart_latest_chart_date]

        # Only for Spotify Charts. Call it before prime caching.
        if context.get('platform_names'):
            platform_names = context.get('platform_names').split(',')
            if len(platform_names) == 1 and 'spotify' in platform_names:
                refresh_fact_charts_filtered_recording = schedule(
                    'refresh_fact_charts_filtered_recording',
                    self.refresh_fact_charts_filtered_recording,
                    requires=[update_dim_chart_latest_chart_date])

                refresh_charts_for_employee_aggregate = schedule(
                    'refresh_charts_for_employee_aggregate',
                    self.refresh_charts_for_employee_aggregate,
                    requires=[update_dim_chart_latest_chart_date])

                previous = [refresh_fact_charts_filtered_recording,
                            refresh_charts_for_employee_aggregate]

        ows_charts_cache_prime = schedule(
            'ows_charts_cache_prime',
            self.ows_charts_cache_prime,
            requires=previous)

        insert_into_log_table = schedule(
            'insert_into_log_table',
            self.insert_into_log_table,
            requires=[ows_charts_cache_prime])

        schedule(
            'set_overall_status_ingested',
            self.set_overall_status,
            requires=[insert_into_log_table],
            input={'status': feed_status.STATUS_INGESTED})

    @property
    def bootstrap(self):
        """Bootstrap initial configuration."""
        return self.create(
            name='bootstrap',
            tasks=base.SyncRunner(
                tasks.bootstrap.fill(
                    namespace='bootstrap',
                    date='context_date',
                    date_limit='context_date_limit',
                    platform_names='platform_names',
                    reload='reload',
                    days_back='days_back')))

    @property
    def clear_log_table(self):
        """Clear log table in case of force refresh."""
        return self.create(
            name='clear_log_table',
            generators=[generators.platform_names_days_back_generator],
            tasks=base.AsyncRunner(
                tasks.clear_log_table.fill(
                    namespace='clear_log_table',
                    feed_name='bootstrap.feed_name',
                    date='platform_day_back',
                    date_limit='bootstrap.date_limit',
                    platform_name='platform_name',
                    reload='bootstrap.reload'),
                max_workers=4))

    @property
    def insert_into_log_table(self):
        """Insert unique keys into the log table in Snowflake."""
        return self.create(
            name='insert_into_log_table',
            tasks=base.AsyncRunner(
                tasks.insert_into_log_table.fill(
                    namespace='insert_into_log_table',
                    feed_name='bootstrap.feed_name',
                    date='bootstrap.date',
                    date_limit='bootstrap.date_limit',
                    ingestion_started_at='bootstrap.ingestion_started_at'),
                max_workers=4))

    @property
    def create_staging_fact(self):
        """Create staging fact table."""
        return self.create(
            name='create_staging_fact',
            tasks=base.AsyncRunner(
                tasks.create_staging_fact.fill(
                    namespace='create_staging_fact')))

    @property
    def find_unchanged_charts(self):
        """Activity to find and log unchanged charts."""
        return self.create(
            name='find_unchanged_charts',
            generators=[generators.platform_names_days_change_check_generator],
            tasks=base.SyncRunner(
                tasks.find_unchanged_charts.fill(
                    namespace='find_unchanged_charts',
                    date='platform_day_back',
                    date_limit='bootstrap.date_limit',
                    platform_name='platform_name')))

    @property
    def load_staging_fact(self):
        """Activity to load fact_charts."""
        return self.create(
            name='load_staging_fact',
            generators=[generators.platform_names_days_back_generator],
            tasks=base.SyncRunner(
                tasks.load_staging_fact.fill(
                    namespace='load_staging_fact',
                    date='platform_day_back',
                    date_limit='bootstrap.date_limit',
                    platform_name='platform_name')))

    @property
    def update_globalsoundrecording_relations_kafka(self):
        """Activity to update GlobalSoundRecording relations via kafka."""
        return self.create(
            name='update_globalsoundrecording_relations_kafka',
            tasks=base.SyncRunner(
                tasks.push_metadata_to_kafka.fill(
                    namespace='push_globalsoundrecording',
                    run_id='execution.run_id',
                    message_key=StaticParam('ISRC'),
                    sf_query_name=StaticParam(
                        'get_sf_globalsoundrecording_relations'),
                    kafka_topic_suffix=StaticParam(
                        'globalsoundrecording.relations')),
                tasks.check_completion_status_in_kafka.fill(
                    namespace='check_completion_status_in_kafka',
                    run_id='execution.run_id',
                    latest_offsets='push_globalsoundrecording.latest_offsets',
                    kafka_topic_suffix=StaticParam(
                        'globalsoundrecording.relations')),
                tasks.check_for_errors_in_dlq.fill(
                    namespace='check_for_errors_in_dlq',
                    workflow_id='execution.workflow_id',
                    run_id='execution.run_id',
                    sf_query_id='push_globalsoundrecording.sf_query_id')))

    @property
    def update_soundrecording_noisrc_nodes_kafka(self):
        """Activity to update SoundRecording nodes via kafka."""
        return self.create(
            name='update_soundrecording_noisrc_nodes_kafka',
            tasks=base.SyncRunner(
                tasks.push_metadata_to_kafka.fill(
                    namespace='push_soundrecording_noisrc',
                    run_id='execution.run_id',
                    message_key=StaticParam('CHARTMETRIC_TRACK_ID'),
                    sf_query_name=StaticParam(
                        'get_sf_soundrecording_noisrc_nodes'),
                    kafka_topic_suffix=StaticParam(
                        'soundrecording.noisrc.nodes')),
                tasks.check_completion_status_in_kafka.fill(
                    namespace='check_completion_status_in_kafka',
                    run_id='execution.run_id',
                    latest_offsets='push_soundrecording_noisrc.latest_offsets',
                    kafka_topic_suffix=StaticParam(
                        'soundrecording.noisrc.nodes')),
                tasks.check_for_errors_in_dlq.fill(
                    namespace='check_for_errors_in_dlq',
                    workflow_id='execution.workflow_id',
                    run_id='execution.run_id',
                    sf_query_id='push_soundrecording_noisrc.sf_query_id')))

    @property
    def update_trackandparticipant_relations_kafka(self):
        """Activity to update Track and Participant ralations via kafka."""
        return self.create(
            name='update_trackandparticipant_relations_kafka',
            generators=[generators.platform_names_generator],
            tasks=base.SyncRunner(
                tasks.push_metadata_to_kafka.fill(
                    namespace='push_trackandparticipant',
                    run_id='execution.run_id',
                    message_key=StaticParam('ISRC'),
                    platform_name='platform_name',
                    sf_query_name=StaticParam(
                        'get_sf_trackandparticipant_relations'),
                    kafka_topic_suffix=StaticParam(
                        '{}.trackandparticipant.relations')),
                tasks.check_completion_status_in_kafka.fill(
                    namespace='check_completion_status_in_kafka',
                    run_id='execution.run_id',
                    platform_name='platform_name',
                    latest_offsets='push_trackandparticipant.latest_offsets',
                    kafka_topic_suffix=StaticParam(
                        '{}.trackandparticipant.relations')),
                tasks.check_for_errors_in_dlq.fill(
                    namespace='check_for_errors_in_dlq',
                    workflow_id='execution.workflow_id',
                    run_id='execution.run_id',
                    sf_query_id='push_trackandparticipant.sf_query_id')))

    @property
    def update_productandlabel_relations_kafka(self):
        """Activity to update Product and Label ralations via kafka."""
        return self.create(
            name='update_productandlabel_relations_kafka',
            generators=[generators.platform_names_generator],
            tasks=base.SyncRunner(
                tasks.push_metadata_to_kafka.fill(
                    namespace='push_productandlabel',
                    run_id='execution.run_id',
                    message_key=StaticParam('CHARTMETRIC_ALBUM_ID'),
                    platform_name='platform_name',
                    platforms_to_skip=(
                        'productandlabel_platforms_to_skip'),
                    sf_query_name=StaticParam(
                        'get_sf_productandlabel_relations'),
                    kafka_topic_suffix=StaticParam(
                        '{}.productandlabel.relations')),
                tasks.check_completion_status_in_kafka.fill(
                    namespace='check_completion_status_in_kafka',
                    skip_kafka_step='push_productandlabel.skip_kafka_step',
                    run_id='execution.run_id',
                    platform_name='platform_name',
                    latest_offsets='push_productandlabel.latest_offsets',
                    kafka_topic_suffix=StaticParam(
                        '{}.productandlabel.relations')),
                tasks.check_for_errors_in_dlq.fill(
                    namespace='check_for_errors_in_dlq',
                    skip_kafka_step='push_productandlabel.skip_kafka_step',
                    workflow_id='execution.workflow_id',
                    run_id='execution.run_id',
                    sf_query_id='push_productandlabel.sf_query_id')))

    @property
    def update_staging_fact_sound_recording(self):
        """Update sound_recording_id in staging_fact."""
        return self.create(
            name='update_staging_fact_sound_recording',
            tasks=base.SyncRunner(
                tasks.update_staging_fact_sound_recording.fill(
                    namespace='update_staging_fact_sound_recording')))

    @property
    def update_staging_fact_public_product(self):
        """Update public_product_id in staging_fact."""
        return self.create(
            name='update_staging_fact_public_product',
            tasks=base.SyncRunner(
                tasks.update_staging_fact_public_product.fill(
                    namespace='update_staging_fact_public_product')))

    @property
    def update_dim_tables(self):
        """Update dimension tables (dim_chart)."""
        return self.create(
            name='update_dim_tables',
            tasks=base.SyncRunner(
                tasks.update_dim_tables.fill(
                    namespace='update_dim_tables')))

    @property
    def update_isrc_to_label_participant_mapping(self):
        """Update ISRC_TO_LABEL_PARTICIPANT table."""
        return self.create(
            name='update_isrc_to_label_participant_mapping',
            tasks=base.SyncRunner(
                tasks.update_isrc_to_label_participant_mapping.fill(
                    namespace='update_isrc_to_label_participant_mapping')))

    @property
    def update_upc_to_label_participant_mapping(self):
        """Update UPC_TO_LABEL_PARTICIPANT table."""
        return self.create(
            name='update_upc_to_label_participant_mapping',
            tasks=base.SyncRunner(
                tasks.update_upc_to_label_participant_mapping.fill(
                    namespace='update_upc_to_label_participant_mapping')))

    @property
    def load_fact_data(self):
        """Activity to load to fact_charts from staging fact table.

        Deletes any existing data for this platform and date in fact_charts.
        """
        return self.create(
            name='load_fact_data',
            generators=[generators.platform_names_days_back_generator],
            tasks=base.SyncRunner(
                tasks.load_fact_data.fill(
                    namespace='load_fact_data',
                    date='platform_day_back',
                    date_limit='bootstrap.date_limit',
                    platform_name='platform_name')))

    @property
    def update_position_change(self):
        """Activity to load update position_change column."""
        return self.create(
            name='update_position_change',
            tasks=base.SyncRunner(
                tasks.update_position_change.fill(
                    namespace='update_position_change',
                    date='bootstrap.date',
                    platform_days_back='bootstrap.platform_days_back'
                )))

    @property
    def update_dim_chart_latest_chart_date(self):
        """Activity to load to fact_charts from staging fact table.

        Deletes any existing data for this platform and date in fact_charts.
        """
        return self.create(
            name='update_dim_chart_latest_chart_date',
            tasks=base.SyncRunner(
                tasks.update_dim_chart_latest_chart_date.fill()))

    @property
    def refresh_aggregated_table(self):
        """Update aggregated tables (dim_aggregated_chart)."""
        return self.create(
            name='refresh_aggregated_table',
            tasks=base.SyncRunner(
                tasks.refresh_aggregated_table.fill(
                    namespace='refresh_aggregated_table')))

    @property
    def refresh_charts_for_employee_aggregate(self):
        """Update charts_for_employee_aggregate tables."""
        return self.create(
            name='refresh_charts_for_employee_aggregate',
            tasks=base.SyncRunner(
                tasks.refresh_charts_for_employee_aggregate.fill(
                    namespace='refresh_charts_for_employee_aggregate')))

    @property
    def refresh_fact_charts_filtered_recording(self):
        """Update fact_charts_filtered_recording tables."""
        return self.create(
            name='refresh_fact_charts_filtered_recording',
            tasks=base.SyncRunner(
                tasks.refresh_fact_charts_filtered_recording.fill(
                    namespace='refresh_fact_charts_filtered_recording')))

    @property
    def ows_charts_cache_prime(self):
        """Activity to trigger selective cache invalidation and priming."""
        return self.create(
            name='ows_charts_cache_prime',
            tasks=base.SyncRunner(
                tasks.ows_charts_cache_prime.fill(
                    platform_days_back='bootstrap.platform_days_back',
                    date_limit='bootstrap.date_limit',
                ),
            ),
        )