#

import logging
from typing import Any, Optional

import email_validator
from snowflake.snowpark import DataFrame, Session
from snowflake.snowpark.functions import cast, col, contains, lit, split, udf
from snowflake.snowpark.types import StringType


def model(dbt: Any, session: Session) -> DataFrame:
    @udf(
        name="validate_email",
        return_type=StringType(),
        input_types=[StringType()],
        replace=True,
    )
    def validate_email(email: str) -> Optional[str]:
        try:
            email_validator.validate_email(email, check_deliverability=False)
        except email_validator.EmailNotValidError:
            logging.error(f"Unable to validate email: {email}")
            return None
        else:
            return email

    target_name = dbt.config.get("target_name")

    if target_name in ["prod", "qa", "test"]:
        # Load tables from actual source tables
        disposable_domains = dbt.ref("DISPOSABLE_EMAIL_DOMAINS_DBT").select(
            "EMAIL_DOMAIN"
        )
        invalid_domains = (
            session.table("EMAIL_DOMAIN_VALIDATION")
            .select("EMAIL_DOMAIN")
            .filter(col("IS_VALID") == False)  # noqa: E712
        )
        email_blocklist = session.table("EMAIL_BLOCKLIST").select("EMAIL")

        results = (
            dbt.ref("STG_EXTERNAL_EVENT_DBT")
            .select("FAN_ID", "FAN_EMAIL")
            .filter(contains(col("FAN_EMAIL"), lit("@")))
            .distinct()
            .cache_result()
            .with_column(
                "FAN_EMAIL_VALIDATED",
                validate_email(col("FAN_EMAIL")),
            )
            .filter(
                col("FAN_EMAIL_VALIDATED").is_not_null(),
            )
            .join(
                email_blocklist,
                col("FAN_EMAIL_VALIDATED") == email_blocklist["EMAIL"],
                "leftanti",
            )
            .with_column(
                "FAN_EMAIL_DOMAIN",
                cast(split(col("FAN_EMAIL_VALIDATED"), lit("@"))[1], StringType()),
            )
            .join(
                disposable_domains,
                col("FAN_EMAIL_DOMAIN") == disposable_domains["EMAIL_DOMAIN"],
                "leftanti",
            )
            .join(
                invalid_domains,
                col("FAN_EMAIL_DOMAIN") == invalid_domains["EMAIL_DOMAIN"],
                "leftanti",
            )
            .select(
                col("FAN_ID"),
                col("FAN_EMAIL_VALIDATED").alias("FAN_EMAIL"),
                col("FAN_EMAIL_DOMAIN"),
            )
        )
    else:
        # speed up local testing
        results = (
            dbt.ref("STG_EXTERNAL_EVENT_DBT")
            .select("FAN_ID", "FAN_EMAIL")
            .filter(contains(col("FAN_EMAIL"), lit("@")))
            .distinct()
            .with_column(
                "FAN_EMAIL_DOMAIN",
                cast(split(col("FAN_EMAIL"), lit("@"))[1], StringType()),
            )
        )
    return results