# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

"""Dag sub-commands"""
import errno
import json
import logging
import signal
import subprocess
import textwrap
from typing import List

from tabulate import tabulate

from airflow import DAG, AirflowException, conf, jobs, settings
from airflow.api.client import get_current_api_client
from airflow.executors.debug_executor import DebugExecutor
from airflow.models import DagBag, DagModel, DagRun, TaskInstance
from airflow.utils import cli as cli_utils
from airflow.utils.cli import get_dag, get_dag_by_file_location, process_subdir, sigint_handler
from airflow.utils.dot_renderer import render_dag
from airflow.utils.session import create_session


def _tabulate_dag_runs(dag_runs: List[DagRun], tablefmt="fancy_grid"):
    tabulat_data = (
        {
            'ID': dag_run.id,
            'Run ID': dag_run.run_id,
            'State': dag_run.state,
            'DAG ID': dag_run.dag_id,
            'Execution date': dag_run.execution_date.isoformat(),
            'Start date': dag_run.start_date.isoformat() if dag_run.start_date else '',
        } for dag_run in dag_runs
    )
    return "\n%s" % tabulate(
        tabular_data=tabulat_data,
        tablefmt=tablefmt
    )


@cli_utils.action_logging
def dag_backfill(args, dag=None):
    """Creates backfill job or dry run for a DAG"""
    logging.basicConfig(
        level=settings.LOGGING_LEVEL,
        format=settings.SIMPLE_LOG_FORMAT)

    signal.signal(signal.SIGTERM, sigint_handler)

    dag = dag or get_dag(args.subdir, args.dag_id)

    if not args.start_date and not args.end_date:
        raise AirflowException("Provide a start_date and/or end_date")

    # If only one date is passed, using same as start and end
    args.end_date = args.end_date or args.start_date
    args.start_date = args.start_date or args.end_date

    if args.task_regex:
        dag = dag.sub_dag(
            task_regex=args.task_regex,
            include_upstream=not args.ignore_dependencies)

    run_conf = None
    if args.conf:
        run_conf = json.loads(args.conf)

    if args.dry_run:
        print("Dry run of DAG {0} on {1}".format(args.dag_id,
                                                 args.start_date))
        for task in dag.tasks:
            print("Task {0}".format(task.task_id))
            ti = TaskInstance(task, args.start_date)
            ti.dry_run()
    else:
        if args.reset_dagruns:
            DAG.clear_dags(
                [dag],
                start_date=args.start_date,
                end_date=args.end_date,
                confirm_prompt=not args.yes,
                include_subdags=True,
            )

        dag.run(
            start_date=args.start_date,
            end_date=args.end_date,
            mark_success=args.mark_success,
            local=args.local,