import re
from typing import Any, Optional

import langcodes
from snowflake.snowpark import DataFrame, Session
from snowflake.snowpark.functions import array_agg, col, udf
from snowflake.snowpark.types import StringType


def model(dbt: Any, session: Session) -> DataFrame:
    dbt.config(packages=["langcodes"])

    @udf(
        name="normalize_language",
        return_type=StringType(),
        input_types=[StringType()],
        replace=True,
        packages=["langcodes"],
    )
    def normalize_language(raw: str) -> Optional[str]:
        import re
        import langcodes
        if not raw:
            return None
        raw = raw.strip()
        if not raw or re.search(r"\d", raw):
            return None

        def _is_iso639_1(code: str) -> bool:
            return bool(code and len(code) == 2 and code.isalpha())

        # Try as a BCP 47 tag first — handles ISO codes like 'fr', 'FR', 'eng', 'zho', 'ES'
        try:
            tag = langcodes.Language.get(raw).language
            if tag and _is_iso639_1(tag):
                return tag.lower()
        except Exception:
            pass
        # Try as a natural-language name — handles 'French', 'deutsch', 'Français'
        try:
            tag = langcodes.find(raw).to_tag().split("-")[0]
            if _is_iso639_1(tag):
                return tag.lower()
        except Exception:
            pass
        return None

    fan_c = dbt.source("crm", "fan_c")

    valid_codes = {
        row[0].lower()
        for row in dbt.ref("iso_639_1").select("code").collect()
        if row[0] is not None
    }

    sql = f"""
    SELECT
        SHA2(LOWER(TRIM(EMAIL_C)), 256) AS FAN_ID,
        PREFERRED_LANGUAGE_C AS LANGUAGE_ORIGINAL,
        F.VALUE AS LANGUAGE_RAW
    FROM {fan_c.table_name},
    LATERAL SPLIT_TO_TABLE(PREFERRED_LANGUAGE_C, ',') AS F
    WHERE
        PREFERRED_LANGUAGE_C IS NOT NULL
        AND NOT IS_DELETED
        AND NOT _FIVETRAN_DELETED
        AND NOT REGEXP_LIKE(PREFERRED_LANGUAGE_C, '.*[0-9].*')
    """

    return (
        session.sql(sql)
        .cache_result()
        .withColumn("LANGUAGE_NORMALIZED", normalize_language(col("LANGUAGE_RAW")))
        .filter(col("LANGUAGE_NORMALIZED").is_not_null())
        .filter(col("LANGUAGE_NORMALIZED").isin(list(valid_codes)))
        .groupBy(col("FAN_ID"), col("LANGUAGE_ORIGINAL"))
        .agg(array_agg(col("LANGUAGE_NORMALIZED")).alias("LANGUAGE"))
        .select(
            col("FAN_ID"),
            col("LANGUAGE"),
            col("LANGUAGE_ORIGINAL"),
        )
    )