"""Excel/CSV → PostgreSQL with robust data type inference.

Handles:
- Integer detection with nullable Int64
- Float precision → NUMERIC vs DOUBLE PRECISION
- Date parsing (ISO, US, EU, Excel serial dates)
- Boolean detection (True/False, Yes/No, 1/0, Y/N)
- Numbers stored as text
- Leading zeros preserved as TEXT
- Mixed-type columns → TEXT
- Empty rows/columns skipped
- Multiple sheet selection
"""

from __future__ import annotations

import logging
import re
from datetime import datetime
from io import BytesIO
from pathlib import Path
from typing import Any

import numpy as np
import pandas as pd

from database.connection_manager import ConnectionManager
from database.local_pg import LocalPG
from knowledge.table_registry import TableRegistry
from knowledge.column_registry import ColumnRegistry

logger = logging.getLogger(__name__)

# Boolean value mappings
_BOOL_TRUE = {"true", "yes", "y", "1", "ja", "si", "oui", "t"}
_BOOL_FALSE = {"false", "no", "n", "0", "nein", "non", "f"}
_BOOL_VALUES = _BOOL_TRUE | _BOOL_FALSE

# Date formats to try (ordered by priority)
_DATE_FORMATS = [
    "%Y-%m-%d",           # ISO: 2024-01-15
    "%Y-%m-%dT%H:%M:%S",  # ISO with time
    "%Y-%m-%d %H:%M:%S",  # ISO with time (space)
    "%d.%m.%Y",           # EU: 15.01.2024
    "%d/%m/%Y",           # EU: 15/01/2024
    "%m/%d/%Y",           # US: 01/15/2024
    "%d-%m-%Y",           # EU: 15-01-2024
    "%d.%m.%Y %H:%M:%S",  # EU with time
    "%m/%d/%Y %H:%M:%S",  # US with time
    "%Y%m%d",             # Compact: 20240115
    "%d %b %Y",           # 15 Jan 2024
    "%d %B %Y",           # 15 January 2024
    "%b %d, %Y",          # Jan 15, 2024
]


class InferredColumn:
    """Result of type inference for a single column."""

    def __init__(
        self,
        name: str,
        pg_type: str,
        pandas_type: str | None = None,
        nullable: bool = True,
        max_length: int | None = None,
        numeric_precision: int | None = None,
        numeric_scale: int | None = None,
    ):
        self.name = name
        self.pg_type = pg_type
        self.pandas_type = pandas_type
        self.nullable = nullable
        self.max_length = max_length
        self.numeric_precision = numeric_precision
        self.numeric_scale = numeric_scale


def _has_leading_zeros(series: pd.Series) -> bool:
    """Check if any values have significant leading zeros (like codes "001234")."""
    for val in series.dropna().head(100):
        s = str(val).strip()
        if s and s[0] == "0" and len(s) > 1 and s.isdigit():
            return True
    return False


def _try_integer(series: pd.Series) -> InferredColumn | None:
    """Try to infer integer type."""
    clean = series.dropna()
    if clean.empty:
        return None

    # Check for leading zeros first — these should stay as TEXT
    if _has_leading_zeros(clean):
        return None

    try:
        converted = pd.to_numeric(clean, errors="coerce")
        if converted.isna().any():
            return None

        # Check if all values are whole numbers
        if not (converted == converted.astype("int64")).all():
            return None

        has_nulls = series.isna().any()
        min_val = int(converted.min())
        max_val = int(converted.max())

        if min_val >= -32768 and max_val <= 32767:
            pg_type = "SMALLINT"
        elif min_val >= -2147483648 and max_val <= 2147483647:
            pg_type = "INTEGER"
        else:
            pg_type = "BIGINT"

        return InferredColumn(
            name=series.name,
            pg_type=pg_type,
            pandas_type="Int64" if has_nulls else "int64",
            nullable=has_nulls,
        )
    except (ValueError, TypeError, OverflowError):
        return None


def _try_float(series: pd.Series) -> InferredColumn | None:
    """Try to infer float/numeric type."""
    clean = series.dropna()
    if clean.empty:
        return None

    # Check for leading zeros — these are codes, not numbers
    if _has_leading_zeros(clean):
        return None

    try:
        converted = pd.to_numeric(clean, errors="coerce")
        if converted.isna().sum() > len(clean) * 0.1:
            return None  # More than 10% failed — not numeric

        if converted.isna().any():
            return None

        has_nulls = series.isna().any()

        # Determine decimal places needed
        max_decimals = 0
        max_digits = 0
        for val in converted:
            s = f"{val:.10f}".rstrip("0")
            if "." in s:
                decimals = len(s.split(".")[1])
                integer_part = len(s.split(".")[0].lstrip("-"))
                max_decimals = max(max_decimals, decimals)
                max_digits = max(max_digits, integer_part + decimals)

        if max_decimals <= 2 and max_digits <= 19:
            # Monetary/fixed precision
            precision = min(max_digits + 2, 38)
            scale = max(max_decimals, 2)
            return InferredColumn(
                name=series.name,
                pg_type=f"NUMERIC({precision},{scale})",
                pandas_type="float64",
                nullable=has_nulls,
                numeric_precision=precision,
                numeric_scale=scale,
            )
        else:
            return InferredColumn(
                name=series.name,
                pg_type="DOUBLE PRECISION",
                pandas_type="float64",
                nullable=has_nulls,
            )
    except (ValueError, TypeError):
        return None


def _try_date(series: pd.Series) -> InferredColumn | None:
    """Try to infer date/timestamp type."""
    clean = series.dropna().astype(str)
    if clean.empty:
        return None

    sample = clean.head(50)

    # Check for Excel serial date numbers (numbers like 44927 = 2023-01-01)
    try:
        numeric_vals = pd.to_numeric(sample, errors="coerce")
        if numeric_vals.notna().all():
            if (numeric_vals > 20000).all() and (numeric_vals < 60000).all():
                # Likely Excel serial dates
                try:
                    converted = pd.to_datetime(
                        pd.to_numeric(series.dropna(), errors="coerce"),
                        unit="D",
                        origin="1899-12-30",
                        errors="coerce",
                    )
                    if converted.notna().sum() > len(converted) * 0.8:
                        return InferredColumn(
                            name=series.name,
                            pg_type="TIMESTAMP",
                            pandas_type="datetime64[ns]",
                            nullable=series.isna().any(),
                        )
                except Exception:
                    pass
    except Exception:
        pass

    # Try various date formats
    for fmt in _DATE_FORMATS:
        try:
            converted = pd.to_datetime(sample, format=fmt, errors="coerce")
            success_rate = converted.notna().sum() / len(sample)
            if success_rate >= 0.8:
                has_time = any(c in fmt for c in ["H", "M", "S"])
                pg_type = "TIMESTAMP" if has_time else "DATE"
                return InferredColumn(
                    name=series.name,
                    pg_type=pg_type,
                    pandas_type="datetime64[ns]",
                    nullable=series.isna().any(),
                )
        except Exception:
            continue

    # Try pandas auto-detection — but only if values look like date strings
    # Skip if all values are purely numeric (avoids false positives like latitude/longitude)
    numeric_check = pd.to_numeric(sample, errors="coerce")
    if not numeric_check.notna().all():
        try:
            converted = pd.to_datetime(sample, errors="coerce", infer_datetime_format=True)
            success_rate = converted.notna().sum() / len(sample)
            if success_rate >= 0.8:
                return InferredColumn(
                    name=series.name,
                    pg_type="TIMESTAMP",
                    pandas_type="datetime64[ns]",
                    nullable=series.isna().any(),
                )
        except Exception:
            pass

    return None


def _try_boolean(series: pd.Series) -> InferredColumn | None:
    """Try to infer boolean type."""
    clean = series.dropna().astype(str).str.strip().str.lower()
    if clean.empty:
        return None

    unique_vals = set(clean.unique())
    if unique_vals.issubset(_BOOL_VALUES) and len(unique_vals) <= 4:
        return InferredColumn(
            name=series.name,
            pg_type="BOOLEAN",
            pandas_type="boolean",
            nullable=series.isna().any(),
        )
    return None


def _infer_text(series: pd.Series) -> InferredColumn:
    """Fallback: infer as TEXT/VARCHAR."""
    clean = series.dropna().astype(str)
    if clean.empty:
        return InferredColumn(name=series.name, pg_type="TEXT", nullable=True)

    max_len = clean.str.len().max()
    if max_len <= 50:
        return InferredColumn(
            name=series.name,
            pg_type=f"VARCHAR({max(max_len * 2, 50)})",
            nullable=series.isna().any(),
            max_length=int(max_len),
        )
    elif max_len <= 255:
        return InferredColumn(
            name=series.name,
            pg_type="VARCHAR(255)",
            nullable=series.isna().any(),
            max_length=int(max_len),
        )
    else:
        return InferredColumn(
            name=series.name,
            pg_type="TEXT",
            nullable=series.isna().any(),
            max_length=int(max_len),
        )


def infer_column_type(series: pd.Series) -> InferredColumn:
    """Multi-pass type inference for a single column.

    Order: boolean → integer → float → date → text
    """
    # Skip if all null
    if series.isna().all():
        return InferredColumn(name=series.name, pg_type="TEXT", nullable=True)

    # Check for mixed types (if pandas already detected object type with varied content)
    if series.dtype == object:
        non_null = series.dropna()
        types_present = set(type(v).__name__ for v in non_null.head(100))
        # If many different Python types, just use TEXT
        if len(types_present) > 3:
            return _infer_text(series)

    # 1. Try boolean
    result = _try_boolean(series)
    if result:
        return result

    # 2. Try integer
    result = _try_integer(series)
    if result:
        return result

    # 3. Try float
    result = _try_float(series)
    if result:
        return result

    # 4. Try date
    result = _try_date(series)
    if result:
        return result

    # 5. Fallback: text
    return _infer_text(series)


def _convert_column(series: pd.Series, inferred: InferredColumn) -> pd.Series:
    """Convert a pandas Series to the inferred type for proper insertion."""
    if inferred.pg_type == "BOOLEAN":
        return series.map(
            lambda v: (
                True if str(v).strip().lower() in _BOOL_TRUE
                else False if str(v).strip().lower() in _BOOL_FALSE
                else None
            )
            if pd.notna(v)
            else None
        )

    if inferred.pg_type in ("DATE", "TIMESTAMP"):
        return pd.to_datetime(series, errors="coerce", infer_datetime_format=True)

    if inferred.pandas_type in ("Int64", "int64"):
        return pd.to_numeric(series, errors="coerce").astype("Int64")

    if inferred.pandas_type == "float64":
        return pd.to_numeric(series, errors="coerce")

    return series


def list_sheets(file_path: str | Path) -> list[str]:
    """List sheet names in an Excel file."""
    path = Path(file_path)
    if path.suffix.lower() in (".xlsx", ".xlsm"):
        import openpyxl
        wb = openpyxl.load_workbook(path, read_only=True)
        sheets = wb.sheetnames
        wb.close()
        return sheets
    elif path.suffix.lower() == ".xls":
        import xlrd
        wb = xlrd.open_workbook(str(path))
        return wb.sheet_names()
    return []


class ExcelImporter:
    """Import Excel/CSV files into PostgreSQL with robust type inference."""

    def __init__(
        self,
        cm: ConnectionManager,
        local_pg: LocalPG,
        table_registry: TableRegistry,
        column_registry: ColumnRegistry,
        ollama: Any = None,
    ) -> None:
        self._cm = cm
        self._local = local_pg
        self._table_reg = table_registry
        self._col_reg = column_registry
        self._ollama = ollama

    async def _llm_analyze_header(self, raw_df: pd.DataFrame) -> dict:
        """Ask local LLM to analyze if row 0 is a header and validate structure.

        Returns: {"has_header": bool, "suggested_names": list[str] | None, "warnings": list[str]}
        """
        if not self._ollama:
            return {"has_header": True, "suggested_names": None, "warnings": []}

        try:
            available = await self._ollama.is_available()
            if not available:
                logger.info("Ollama not available, skipping LLM header analysis")
                return {"has_header": True, "suggested_names": None, "warnings": []}
        except Exception:
            return {"has_header": True, "suggested_names": None, "warnings": []}

        # Show first 5 rows (raw, no header assumption)
        preview_rows = raw_df.head(5).fillna("").values.tolist()
        preview_text = ""
        for i, row in enumerate(preview_rows):
            preview_text += f"Row {i}: {row}\n"

        prompt = f"""Analyze this spreadsheet data. The first 5 rows are shown raw (no header assumed).

{preview_text}

Answer these questions in EXACTLY this format (no extra text):
HAS_HEADER: YES or NO
REASON: one short sentence why
WARNINGS: comma-separated list of issues, or NONE

Rules:
- HAS_HEADER=YES if Row 0 contains column names/labels (text describing what the column contains)
- HAS_HEADER=NO if Row 0 contains actual data values (numbers, dates, names of things being tracked)
- A header row typically has unique descriptive labels like "Name", "Date", "Amount", "Status"
- A data row has values like "John", "2024-01-15", "1234.56", "Active"
"""

        try:
            response = await self._ollama.generate(prompt, temperature=0.0, max_tokens=256)
            logger.info(f"LLM header analysis: {response.strip()}")

            has_header = True
            warnings = []

            for line in response.strip().split("\n"):
                line = line.strip()
                if line.startswith("HAS_HEADER:"):
                    val = line.split(":", 1)[1].strip().upper()
                    has_header = "YES" in val
                elif line.startswith("WARNINGS:"):
                    val = line.split(":", 1)[1].strip()
                    if val.upper() != "NONE" and val:
                        warnings = [w.strip() for w in val.split(",") if w.strip()]

            return {"has_header": has_header, "suggested_names": None, "warnings": warnings}
        except Exception as e:
            logger.warning(f"LLM header analysis failed (non-critical): {e}")
            return {"has_header": True, "suggested_names": None, "warnings": []}

    @staticmethod
    def _detect_csv_sep(path: Path | BytesIO) -> str:
        """Auto-detect CSV delimiter by reading the first line."""
        if isinstance(path, BytesIO):
            path.seek(0)
            first_line = path.readline().decode("utf-8", errors="replace")
            path.seek(0)
        else:
            with open(path, "r", encoding="utf-8", errors="replace") as f:
                first_line = f.readline()

        # Count candidate delimiters in the header line
        candidates = {";": first_line.count(";"), "\t": first_line.count("\t"), ",": first_line.count(",")}
        best = max(candidates, key=candidates.get)
        # Only use non-comma if it appears significantly more
        if best != "," and candidates[best] > candidates.get(",", 0):
            logger.info(f"CSV delimiter detected: {repr(best)} ({candidates[best]} occurrences)")
            return best
        return ","

    def _read_file_raw(
        self, path: Path | BytesIO, is_csv: bool, sheet_name: str | int | None = 0
    ) -> pd.DataFrame:
        """Read file with header=None to get raw rows for LLM analysis."""
        if isinstance(path, BytesIO):
            path.seek(0)
        if is_csv:
            sep = self._detect_csv_sep(path)
            return pd.read_csv(
                path if isinstance(path, BytesIO) else str(path),
                header=None, dtype=str, nrows=10, sep=sep,
            )
        return pd.read_excel(
            path if isinstance(path, BytesIO) else str(path),
            sheet_name=sheet_name, header=None, dtype=str, nrows=10,
        )

    async def import_file(
        self,
        file_path: str | Path | BytesIO,
        table_name: str,
        sheet_name: str | int | None = 0,
        original_filename: str | None = None,
    ) -> dict:
        """Import an Excel or CSV file into PostgreSQL.

        Returns import summary with table_name, row_count, column_types.
        """
        path = file_path if isinstance(file_path, BytesIO) else Path(file_path)
        filename = original_filename or (str(path) if isinstance(path, Path) else "upload")
        is_csv = filename.lower().endswith(".csv") if filename else False

        # 0. LLM pre-analysis: check if first row is header
        llm_analysis = {"has_header": True, "suggested_names": None, "warnings": []}
        try:
            raw_df = self._read_file_raw(path, is_csv, sheet_name)
            if not raw_df.empty:
                llm_analysis = await self._llm_analyze_header(raw_df)
                logger.info(f"LLM analysis: has_header={llm_analysis['has_header']}, warnings={llm_analysis['warnings']}")
        except Exception as e:
            logger.warning(f"Raw file pre-read failed (non-critical): {e}")

        # 1. Read file (with or without header based on LLM analysis)
        logger.info(f"Reading file: {filename}")
        if isinstance(path, BytesIO):
            path.seek(0)

        header_row = 0 if llm_analysis["has_header"] else None
        if is_csv:
            if isinstance(path, BytesIO):
                path.seek(0)
            sep = self._detect_csv_sep(path)
            df = pd.read_csv(
                path if isinstance(path, BytesIO) else str(path),
                dtype=str, header=header_row, sep=sep,
            )
        else:
            df = pd.read_excel(
                path if isinstance(path, BytesIO) else str(path),
                sheet_name=sheet_name, dtype=str, header=header_row,
            )

        if df.empty:
            raise ValueError("File contains no data")

        # If no header detected, generate column names
        if not llm_analysis["has_header"]:
            df.columns = [f"col_{i+1}" for i in range(len(df.columns))]
            logger.info("No header row detected by LLM — using generated column names")

        # 2. Clean up
        # Remove completely empty rows and columns
        df = df.dropna(how="all").dropna(axis=1, how="all")

        # Sanitize column names for PostgreSQL
        df.columns = [self._sanitize_column_name(c) for c in df.columns]

        # Remove duplicate column names (append _1, _2, etc.)
        seen = {}
        new_cols = []
        for col in df.columns:
            if col in seen:
                seen[col] += 1
                new_cols.append(f"{col}_{seen[col]}")
            else:
                seen[col] = 0
                new_cols.append(col)
        df.columns = new_cols

        logger.info(f"DataFrame shape after cleanup: {df.shape}")

        # 3. Infer types for each column
        inferred_columns: list[InferredColumn] = []
        for col_name in df.columns:
            inferred = infer_column_type(df[col_name])
            inferred.name = col_name
            inferred_columns.append(inferred)
            logger.info(f"  Column '{col_name}' → {inferred.pg_type}")

        # 4. Convert DataFrame columns to inferred types
        for inferred in inferred_columns:
            df[inferred.name] = _convert_column(df[inferred.name], inferred)

        # 5. Create PostgreSQL table
        pg_columns = [{"name": ic.name, "type": ic.pg_type} for ic in inferred_columns]

        table_exists = await self._cm.table_exists("v3q_data", table_name)
        if table_exists:
            await self._local.drop_table(table_name)

        await self._local.create_table_from_columns(table_name, pg_columns)
        logger.info(f"Created table v3q_data.{table_name}")

        # 6. Insert data
        row_count = await self._local.insert_dataframe(table_name, df)
        logger.info(f"Inserted {row_count} rows into v3q_data.{table_name}")

        # 7. Track in excel_imports
        import json
        column_types = {ic.name: ic.pg_type for ic in inferred_columns}
        await self._cm.execute_write("""
            INSERT INTO v3q_data.excel_imports
                (original_filename, sheet_name, local_table, row_count, column_count, column_types)
            VALUES (:filename, :sheet, :table, :rows, :cols, :types)
            ON CONFLICT (local_table) DO UPDATE SET
                row_count = EXCLUDED.row_count,
                column_count = EXCLUDED.column_count,
                column_types = EXCLUDED.column_types,
                imported_at = now()
        """, {
            "filename": filename,
            "sheet": str(sheet_name) if sheet_name is not None else None,
            "table": table_name,
            "rows": row_count,
            "cols": len(inferred_columns),
            "types": json.dumps(column_types),
        })

        # 8. Register in table_registry + column_registry
        existing = await self._table_reg.get_by_local_table(table_name)
        if existing:
            await self._table_reg.update(
                str(existing["id"]),
                row_count=row_count,
                is_synced=True,
            )
            table_id = str(existing["id"])
        else:
            reg = await self._table_reg.create(
                connection_id=None,
                source_schema=None,
                source_table=filename,
                local_table=table_name,
                display_name=table_name,
                source_system="excel",
                row_count=row_count,
                is_synced=True,
            )
            table_id = str(reg["id"])

        await self._col_reg.delete_by_table(table_id)
        for ic in inferred_columns:
            await self._col_reg.create(
                table_id=table_id,
                column_name=ic.name,
                data_type=ic.pg_type,
                nullable=ic.nullable,
                max_length=ic.max_length,
                numeric_precision=ic.numeric_precision,
                numeric_scale=ic.numeric_scale,
            )

        # 9. Auto-enrich column descriptions using local LLM
        enriched_count = 0
        if self._ollama:
            try:
                from local_llm.column_enricher import ColumnEnricher
                enricher = ColumnEnricher(self._ollama, self._cm, self._col_reg)
                domain_hints = await self._load_domain_hints()
                enriched = await enricher.enrich_table(
                    table_id=table_id,
                    local_table=table_name,
                    source_system="excel",
                    domain_hints=domain_hints,
                )
                enriched_count = len(enriched)
                logger.info(f"Auto-enriched {enriched_count} column descriptions for Excel import")
            except Exception as e:
                logger.warning(f"Column enrichment failed (non-critical): {e}")

        result = {
            "table_name": table_name,
            "row_count": row_count,
            "column_count": len(inferred_columns),
            "column_types": column_types,
            "table_id": table_id,
            "enriched_columns": enriched_count,
        }
        if not llm_analysis["has_header"]:
            result["llm_note"] = "No header row detected — generated column names (col_1, col_2, ...)"
        if llm_analysis["warnings"]:
            result["llm_warnings"] = llm_analysis["warnings"]
        return result

    async def _load_domain_hints(self) -> str | None:
        """Load domain-specific knowledge from app_settings."""
        try:
            import json as _json
            row = await self._cm.fetch_one(
                "SELECT value FROM v3q_config.app_settings WHERE key = 'domain_knowledge'",
            )
            if row:
                data = _json.loads(row["value"]) if isinstance(row["value"], str) else row["value"]
                if isinstance(data, list):
                    return "\n".join(f"- {item}" for item in data)
                elif isinstance(data, str):
                    return data
        except Exception:
            pass
        return None

    @staticmethod
    def _sanitize_column_name(name: str) -> str:
        """Sanitize a column name for PostgreSQL."""
        # Replace special characters with underscores
        sanitized = re.sub(r"[^\w]", "_", str(name).strip())
        # Remove leading/trailing underscores
        sanitized = sanitized.strip("_")
        # Ensure it doesn't start with a number
        if sanitized and sanitized[0].isdigit():
            sanitized = f"col_{sanitized}"
        # Lowercase
        sanitized = sanitized.lower()
        # Fallback
        return sanitized or "unnamed_column"
