import logging
import os
import urllib.parse

import airflow.models
from airflow.providers.slack.hooks.slack_webhook import SlackWebhookHook

SLACK_WEBHOOK_TOKEN = "https://hooks.slack.com/services/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"


def on_dag_failure_slack_callback(**context):
    from airflow.utils.log.secrets_masker import mask_secret

    if os.environ.get("IS_LOCAL", "false") == "true":
        logging.info("Running Airflow locally, thus skipping the Slack notification...")
        return None

    # https://airflow.apache.org/docs/apache-airflow/stable/security/secrets/index.html
    mask_secret(SLACK_WEBHOOK_TOKEN)

    stage = os.environ["STAGE"]
    airflow_url = f"https://airflow-{stage}.fansifter.cloud"
    emoji = f":red_circle:{' :bangbang:' if stage == 'live' else ''}"

    ti: airflow.models.TaskInstance = context.get("task_instance")
    dag_run: airflow.models.DagRun = ti.get_dagrun()

    dag = ti.dag_id
    exec_date = str(context.get("execution_date"))
    run_id = context.get("run_id")
    error = context.get("exception")
    failed_task_instances = dag_run.get_task_instances(state="failed")
    failed_task_ids = [task.task_id for task in failed_task_instances]
    failed_task = failed_task_ids[0]

    message_notification_text = f"{emoji} Airflow {stage.upper()}: Run failed!"
    message_blocks = [
        {
            "type": "header",
            "text": {"type": "plain_text", "text": f"{emoji} Airflow {stage.upper()}: Run failed!", "emoji": True},
        },
        {
            "type": "section",
            "fields": [
                {
                    "type": "mrkdwn",
                    "text": f"*Task:*\n{failed_task}"
                    if len(failed_task_ids) == 1
                    else f"*Tasks:*\n{', '.join(failed_task_ids)}",
                },
                {
                    "type": "mrkdwn",
                    "text": f"*DAG:*\n<{airflow_url}/graph?dag_id={dag}&execution_date={urllib.parse.quote(exec_date)}|{dag}>",
                },
            ],
        },
        {
            "type": "section",
            "fields": [
                {"type": "mrkdwn", "text": f"*DAG Run:*\n{run_id}"},
                {
                    "type": "mrkdwn",
                    "text": f"*Log:*\n<{airflow_url}/log?dag_id={dag}&task_id={failed_task}&execution_date={urllib.parse.quote(exec_date)}|View>",
                },
            ],
        },
    ]
    if error is not None:
        message_blocks.append(
            {"type": "section", "fields": [{"type": "mrkdwn", "text": f"*Exception:*\n{error}"},],}
        )

    slack_hook = SlackWebhookHook(
        webhook_token=SLACK_WEBHOOK_TOKEN, message=message_notification_text, blocks=message_blocks
    )

    try:
        slack_hook.execute()
    except Exception as e:
        logging.error(e, exc_info=e)