import json
import logging
from datetime import datetime, timedelta
from typing import Any

from airflow import settings
from airflow.decorators import dag, task
from airflow.models.connection import Connection
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
from dmp_workflows import config
from dmp_workflows.hooks.auth import snowflake_password
from dmp_workflows.hooks.email_hook import EmailHook
from dmp_workflows.hooks.sql.deleted_fans import (
    SELECT_QUERY,
    UPDATE_QUERY,
)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


@task(retries=2, retry_delay=timedelta(seconds=30))  # type: ignore[misc]
def create_snowflake_connection() -> Any:
    session = settings.Session()
    conn = (
        session.query(Connection)
        .filter(Connection.conn_id == config.SNOWFLAKE_CONN_NAME)
        .first()
    )
    logger.info("Connection exist - %s", conn)
    if conn is None:
        logger.info("Creating connection")

        conn = Connection(
            conn_id=config.SNOWFLAKE_CONN_NAME,
            conn_type="snowflake",
            schema=config.SNOWFLAKE_SCHEMA,
            login=config.SNOWFLAKE_USER,
            password=snowflake_password,
            extra=json.dumps(
                {
                    "account": config.SNOWFLAKE_ACCOUNT,
                    "database": config.SNOWFLAKE_DATABASE,
                    "role": config.SNOWFLAKE_ROLE,
                    "warehouse": config.SNOWFLAKE_WAREHOUSE,
                    "insecure_mode": False,
                }
            ),
        )
        session = settings.Session()
        session.add(conn)
        session.commit()
        logger.info("Connection created - %s", conn)

    return conn.conn_id


@task(retries=2, retry_delay=timedelta(seconds=30))  # type: ignore[misc]
def get_deleted_fans_count(conn_id: str) -> dict[str, Any]:
    snowflake_hook = SnowflakeHook(snowflake_conn_id=conn_id)
    fans_count, last_deleted_at = snowflake_hook.get_first(SELECT_QUERY)
    logger.info("Number of deleted not reported fans` - %s", fans_count)
    return {
        "fans_count": fans_count,
        "last_deleted_at": str(last_deleted_at)
        if last_deleted_at is not None
        else None,
    }


@task(retries=2, retry_delay=timedelta(seconds=30))  # type: ignore[misc]
def send_email_with_deleted_fans_info(fans_count: int) -> bool:
    return EmailHook.send_fans_deleted_report(fans_count)


@task(retries=2, retry_delay=timedelta(seconds=30))  # type: ignore[misc]
def mark_emails_sended(conn_id: str, last_deleted_at: str) -> Any | list[Any] | None:
    snowflake_hook = SnowflakeHook(snowflake_conn_id=conn_id)
    return snowflake_hook.run(UPDATE_QUERY.format(last_deleted_at=last_deleted_at))


@dag(
    start_date=datetime(2014, 12, 2, hour=13),
    schedule_interval=timedelta(days=1),
    catchup=False,
)
def deleted_fans_dag() -> None:
    conn_id = create_snowflake_connection()
    fans_result = get_deleted_fans_count(conn_id)
    fans_count = fans_result["fans_count"]
    last_deleted_at = fans_result["last_deleted_at"]
    if not send_email_with_deleted_fans_info(fans_count):
        # send slack alert
        raise ValueError("Email was not sent")
    if last_deleted_at is not None:
        marking_result = mark_emails_sended(conn_id, last_deleted_at)
        if marking_result is None:
            raise ValueError("Data was not updated")


deleted_fans_dag()