import logging
from typing import Any, Optional

import pandas as pd

from semantic_models.exceptions import NotFoundError, ValidationError
from semantic_models.services.dss_utils import fetch_distinct_values, get_dataset_location_info, is_categorical_type
from semantic_models.utils import add_limit_to_query, quote_identifier

logger = logging.getLogger(__name__)

DEFAULT_SAMPLE_SIZE = 10000

MAX_SAMPLE_VALUES = 5

MAX_SAMPLE_VALUE_LENGTH = 30

LOW_CARDINALITY_THRESHOLD = 50
HIGH_CARDINALITY_RATIO = 0.9


def get_dataset_schema(project_key: str, dataset_name: str) -> dict[str, Any]:
    try:
        import dataiku

        location = get_dataset_location_info(project_key, dataset_name)

        client = dataiku.api_client()
        project = client.get_project(project_key)
        dataset = project.get_dataset(dataset_name)

        schema = dataset.get_schema()
        columns_info = schema.get("columns", [])

        columns = []
        for col in columns_info:
            columns.append(
                {
                    "name": col["name"],
                    "type": col.get("type", "string"),
                    "description": col.get("comment") or None,
                }
            )

        return {
            "columns": columns,
            "connection": location.connection,
            "full_table": location.full_table,
            "dialect": location.dialect,
        }

    except (ValidationError, NotFoundError):
        raise
    except Exception as e:
        logger.exception(f"Failed to get schema for {project_key}.{dataset_name}")
        raise NotFoundError(f"Dataset not found or unavailable: {e}")


def profile_all_columns(
    connection: str,
    full_table: str,
    columns: list[dict[str, Any]],
    dialect: str = "postgres",
    sample_size: int = DEFAULT_SAMPLE_SIZE,
) -> dict[str, dict[str, Any]]:
    import dataiku

    profile_cache: dict[str, dict[str, Any]] = {}

    # Initialize profiles with schema info
    for col in columns:
        col_name = col["name"]
        profile_cache[col_name] = {
            "type": col.get("type", "string"),
            "description": col.get("description"),
            "distinct_count": 0,
            "null_count": 0,
            "sample_values": [],
            "is_categorical": False,
        }

    if not columns:
        return profile_cache

    # Build column list for SELECT (quote each column name)
    quoted_columns = [quote_identifier(col["name"], dialect) for col in columns]
    columns_sql = ", ".join(quoted_columns)

    # Build sample query with dialect-aware LIMIT
    sql = f"SELECT {columns_sql} FROM {full_table}"
    sql = add_limit_to_query(sql, sample_size, dialect)

    # Log SQL query (truncate if very long)
    sql_preview = sql[:500] + "..." if len(sql) > 500 else sql
    logger.info(f"[Profiling SQL] {sql_preview}")

    try:
        executor = dataiku.core.sql.SQLExecutor2(connection=connection)
        df = executor.query_to_df(sql)

        total_rows = len(df)
        # Log response summary
        logger.info(f"[Profiling Result] {total_rows} rows, {len(df.columns)} columns")
        if total_rows == 0:
            logger.warning(f"No rows returned from {full_table}")
            return profile_cache

        # Compute stats for each column
        for col in columns:
            col_name = col["name"]
            col_type = col.get("type", "string")

            if col_name not in df.columns:
                # Column might have been quoted differently
                continue

            series = df[col_name]
            profile = profile_cache[col_name]

            # Basic stats
            null_count = int(series.isna().sum())
            non_null = series.dropna()
            distinct_count = int(non_null.nunique())

            profile["null_count"] = null_count
            profile["distinct_count"] = distinct_count

            # Determine if categorical
            is_low_cardinality = distinct_count <= LOW_CARDINALITY_THRESHOLD
            cardinality_ratio = distinct_count / max(1, total_rows - null_count)
            is_high_cardinality = cardinality_ratio > HIGH_CARDINALITY_RATIO
            is_categorical = is_categorical_type(col_type) and is_low_cardinality and not is_high_cardinality
            profile["is_categorical"] = is_categorical

            # Get sample values for categorical columns
            if is_categorical:
                try:
                    # Get top distinct values
                    value_counts = non_null.value_counts().head(MAX_SAMPLE_VALUES)
                    sample_values = []
                    for val in value_counts.index:
                        if val is not None:
                            str_val = str(val)
                            if len(str_val) > MAX_SAMPLE_VALUE_LENGTH:
                                str_val = str_val[:MAX_SAMPLE_VALUE_LENGTH] + "..."
                            sample_values.append(str_val)
                    profile["sample_values"] = sample_values
                except Exception as e:
                    logger.debug(f"Failed to get sample values for {col_name}: {e}")

        # Log profile summary
        cat_cols = [c["name"] for c in columns if profile_cache.get(c["name"], {}).get("is_categorical")]
        logger.info(f"Profiled {len(columns)} columns from {full_table} ({total_rows} sample rows)")
        if cat_cols:
            logger.info(
                f"[Profiling] Categorical columns ({len(cat_cols)}): {cat_cols[:5]}{'...' if len(cat_cols) > 5 else ''}"
            )
        return profile_cache

    except Exception as e:
        logger.warning(f"Failed to profile columns from {full_table}: {e}")
        # Return profiles with just schema info
        return profile_cache


def get_column_profile(
    profile_cache: dict[str, dict[str, Any]],
    column_name: str,
) -> Optional[dict[str, Any]]:
    return profile_cache.get(column_name)


def get_top_values(
    connection: str,
    full_table: str,
    column_name: str,
    dialect: str = "postgres",
    limit: int = 5,
) -> list[str]:
    try:
        return fetch_distinct_values(
            connection=connection,
            full_table=full_table,
            column_name=column_name,
            dialect=dialect,
            limit=limit,
            max_value_length=MAX_SAMPLE_VALUE_LENGTH,
        )
    except ValueError as e:
        logger.warning(f"Failed to get top values for {column_name}: {e}")
        return []
    except Exception as e:
        logger.warning(f"Failed to get top values for {column_name}: {e}")
        return []
