import json
from enum import Enum
from typing import Dict

import requests
from airflow.exceptions import AirflowFailException
from airflow.models import Variable
from airflow.models.taskinstance import Context


class Color(Enum):
    RED = "red"
    BLUE = "blue"
    GREEN = "green"


# 错误处理回调函数
def send_error_message_handler(chat_id: str, context: Context):
    card = build_error_card(context)
    send_card_message(card, chat_id=chat_id)


def get_access_token():
    url = "https://open.feishu.cn/open-apis/auth/v3/app_access_token/internal/"
    config = Variable.get("auto_bot_config", {}, True)
    payload = {"app_id": config["appId"], "app_secret": config["accessToken"]}
    response = requests.post(url, json=payload)
    return response.json().get("app_access_token")


# 发送卡片消息
def send_card_message(card: object, chat_id: str = None, email: str = None):
    url = "https://open.feishu.cn/open-apis/message/v4/send/"
    token = get_access_token()
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    if chat_id:
        payload = {"chat_id": chat_id, "msg_type": "interactive", "card": card}
    elif email:
        payload = {"email": email, "msg_type": "interactive", "card": card}
    else:
        raise AirflowFailException(
            "at least one parameter from email or chat_id needs to be provided"
        )
    response = requests.post(url, headers=headers, json=payload)
    response.raise_for_status()
    response_data = response.json()
    if response_data.get("code") != 0:
        raise AirflowFailException(response_data["msg"])

    # 获取 message_id
    message_id = response_data.get("data", {}).get("message_id")
    return message_id


def update_card(message_id: str, card: object):
    url = f"https://open.feishu.cn/open-apis/im/v1/messages/{message_id}"
    token = get_access_token()
    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
    payload = {"content": json.dumps(card)}
    response = requests.patch(url, headers=headers, json=payload)
    response.raise_for_status()
    response_data = response.json()
    if response_data.get("code") != 0:
        raise AirflowFailException(response_data["msg"])


def build_status_card(
    color: Color,
    title: str,
    fields: Dict[str, str],
    components: Dict[str, str] = [],
) -> object:
    card = {
        "config": {"wide_screen_mode": True},
        "header": {
            "template": color.value,
            "title": {"tag": "plain_text", "content": title},
        },
        "elements": [
            {
                "tag": "div",
                "text": {"tag": "plain_text", "content": "Task Info"},
                "fields": [
                    {
                        "is_short": True,
                        "text": {
                            "tag": "lark_md",
                            "content": f"**{key}: **{value}",
                        },
                    }
                    for key, value in fields.items()
                ],
            },
        ],
    }
    if len(components) > 0:
        card["elements"].append(
            {
                "tag": "div",
                "text": {"tag": "plain_text", "content": "Components"},
                "fields": [
                    {
                        "is_short": False,
                        "text": {"tag": "lark_md", "content": f"** {name}: **{value}"},
                    }
                    for name, value in components.items()
                ],
            },
        )
    return card


def build_error_card(context: Context):
    task_instance = context.get("task_instance")
    dag = context.get("dag")
    dag_run = context.get("dag_run")
    task_id = task_instance.task_id
    execution_date = context["execution_date"]
    error_message = context.get("exception")

    # 获取该TaskInstance的所有XCom对象
    xcoms = task_instance.xcom_pull(task_ids=task_id)

    # 输出所有XCom对象
    xcoms_fields = []
    if xcoms:
        for key, value in xcoms.items():
            xcoms_fields.append(
                {
                    "is_short": False,
                    "text": {
                        "tag": "lark_md",
                        "content": f"**{key}: **{value}",
                    },
                }
            )

    card = {
        "config": {"wide_screen_mode": True},
        "header": {
            "template": "red",
            "title": {
                "tag": "plain_text",
                "content": f"🚨 Encounter an error {dag.dag_id}/{task_id}",
            },
        },
        "elements": [
            {
                "tag": "div",
                "text": {"tag": "plain_text", "content": "Task Info"},
                "fields": [
                    {
                        "is_short": True,
                        "text": {
                            "tag": "lark_md",
                            "content": f"**📦 DAG: **{dag.dag_id}",
                        },
                    },
                    {
                        "is_short": True,
                        "text": {
                            "tag": "lark_md",
                            "content": f"**⚙️ Task: **{task_id}",
                        },
                    },
                    {
                        "is_short": False,
                        "text": {
                            "tag": "lark_md",
                            "content": f"**🆔 Run Id: **{dag_run.run_id}",
                        },
                    },
                    {
                        "is_short": False,
                        "text": {
                            "tag": "lark_md",
                            "content": f"**🛫 Execution Date: **{execution_date}",
                        },
                    },
                ],
            },
        ],
    }
    if len(xcoms_fields) > 0:
        card["elements"].append({"tag": "hr"})
        card["elements"].append(
            {
                "tag": "div",
                "text": {"tag": "plain_text", "content": "XCOM Values"},
                "fields": xcoms_fields,
            }
        )
    if error_message:
        card["elements"].append({"tag": "hr"})
        card["elements"].append(
            {
                "tag": "div",
                "text": {"tag": "plain_text", "content": f"Error: {error_message}"},
            }
        )

    return card