#!/usr/bin/env python3
"""Backload executor exec script.

It pick backload tasks to process and executes them.
"""

import json
import logging
import os
import subprocess
import sys

from feed_ingestion import flows
from feed_ingestion.bin.snowflake_executor \
    import BackloadTasksExecutor
from feed_ingestion.conf.config import SF_CONFIG
from feed_ingestion.util.aws import swf as swf_util

logging.basicConfig(level=logging.INFO)

LIMIT_CONCURRENT_EXECUTIONS_BY_FLOW = {
    'amazon_music': 3,
    'spotify': 2,
    'itunes': 4,
    'youtube_bulk_reports': 10,
    'ritmogestion': 7,
    'tiktok': 2,
    'apple_music_streams': 1,
    'DEFAULT': 3,
}


def _flow_limit(flow):
    """Return max number of concurrent executions for the flow."""
    return LIMIT_CONCURRENT_EXECUTIONS_BY_FLOW.get(
        flow, LIMIT_CONCURRENT_EXECUTIONS_BY_FLOW['DEFAULT'])


def start_tasks(only_flows=()):
    """Execute tasks which can be executed."""
    with BackloadTasksExecutor(SF_CONFIG) as sf_executor:
        flows_tasks = sf_executor.select_flows_to_process()
        for flow, num_tasks in flows_tasks:
            logging.info(f'{flow} has {num_tasks} tasks to process')
            if only_flows and flow not in only_flows:
                logging.warning(f'Skipping flow {flow} '
                                f'(runs only: {only_flows})')
                continue
            flow_entity = flows.get_flow(flow)

            running_tasks = swf_util.count_running_workflows_by_type(
                flow_entity.domain, flow_entity.name)
            limit = _flow_limit(flow)
            logging.info(f'{flow} has {running_tasks} running tasks '
                         f'(limit {limit})')
            if running_tasks >= limit:
                logging.info(f'Skip flow {flow}')

            num_tasks_to_run = min(limit - running_tasks, num_tasks)
            num_tasks_to_run = max(num_tasks_to_run, 0)
            tasks = sf_executor.select_backload_tasks_by_flow(
                flow=flow,
                limit=num_tasks_to_run)
            for licensor, date, context in tasks:
                context_json = json.loads(context)
                context_json['backfill'] = 'True'
                patched_context = json.dumps(context_json)

                logging.info(f'Running flow={flow}, context={patched_context}')
                call = ['garcon', 'exec', flow, '--context', patched_context]
                return_code = subprocess.call(call)

                logging.info(f'Exec returns {return_code}')
                if return_code == 0:
                    sf_executor.update_backload_tasks_as_processed(
                        licensor=licensor,
                        flow=flow,
                        context=context,
                        date=date,
                    )


def main():
    """Execute main method."""
    flows_to_process = os.environ.get('ONLY_FLOWS', sys.argv[1:])
    start_tasks(only_flows=flows_to_process)


if __name__ == '__main__':
    main()