from __future__ import annotations

from datetime import datetime
from typing import Any, Dict, List, Optional, Sequence, Set

import pandas as pd
import sqlalchemy as sa
from sqlalchemy import select
from sqlalchemy.dialects.postgresql import insert
from sqlalchemy.orm import Session

from db.models import AttributeDef, PlytixAttributeCatalog
from db.source_imports import infer_destination_hint, normalize_column_key


CODE_ALIASES = [
    "attribute_code",
    "code",
    "key",
    "field_key",
    "field_code",
    "name",  # Plytix export "Name" column
    "attribute_name",
    "label",
]
LABEL_ALIASES = ["label", "name", "attribute_name", "attribute_label", "display_name"]
TYPE_ALIASES = ["type", "data_type", "field_type", "attribute_type"]
GROUP_ALIASES = ["group", "group_name", "groups", "attribute_group", "section", "family"]
REQUIRED_ALIASES = ["required", "is_required", "mandatory"]
ACTIVE_ALIASES = ["active", "is_active", "enabled", "status"]


def import_plytix_attribute_catalog_dataframe(
    session: Session,
    df: pd.DataFrame,
    *,
    source_label: str = "plytix",
    exclude_destination_hints: Optional[Sequence[str]] = ("shopify",),
) -> Dict[str, int]:
    rows: List[Dict[str, Any]] = []
    skipped = 0
    skipped_channel = 0
    duplicates_merged = 0
    imported_at = datetime.utcnow()
    columns = [str(c) for c in df.columns]
    excluded_hints = {
        str(hint).strip().lower()
        for hint in (exclude_destination_hints or [])
        if str(hint).strip()
    }

    def _skip_column(column_name: str) -> bool:
        nonlocal skipped_channel
        if not excluded_hints:
            return False
        hint = infer_destination_hint(normalize_column_key(column_name))
        if hint and hint.lower() in excluded_hints:
            skipped_channel += 1
            return True
        if "shopify" in excluded_hints and "shopify" in str(column_name).lower():
            skipped_channel += 1
            return True
        return False

    if len(df) == 0:
        for column in columns:
            if _skip_column(column):
                continue
            code = normalize_column_key(column)
            if not code:
                skipped += 1
                continue
            duplicates_merged += _append_catalog_row(
                rows,
                {
                    "source_label": source_label,
                    "attribute_code": code,
                    "label": column,
                    "data_type": None,
                    "group_name": None,
                    "destination_hint": infer_destination_hint(code),
                    "is_required": None,
                    "is_active": True,
                    "raw_payload": {"column_name": column},
                    "imported_at": imported_at,
                    "updated_at": imported_at,
                },
            )

    for raw in df.to_dict(orient="records"):
        payload = {str(k): _clean_text(v) for k, v in raw.items()}
        keyed = {normalize_column_key(k): v for k, v in payload.items()}
        code_source = _first(keyed, CODE_ALIASES)
        if not code_source and len(columns) == 1:
            code_source = columns[0]
        code = normalize_column_key(code_source or "")
        if not code:
            skipped += 1
            continue
        if excluded_hints and infer_destination_hint(code) in excluded_hints:
            skipped_channel += 1
            continue
        label = _clean_text(_first(keyed, LABEL_ALIASES)) or str(code_source or code)
        duplicates_merged += _append_catalog_row(
            rows,
            {
                "source_label": source_label,
                "attribute_code": code,
                "label": label,
                "data_type": _clean_text(_first(keyed, TYPE_ALIASES)),
                "group_name": _clean_text(_first(keyed, GROUP_ALIASES)),
                "destination_hint": infer_destination_hint(code),
                "is_required": _to_bool(_first(keyed, REQUIRED_ALIASES)),
                "is_active": _active_value(_first(keyed, ACTIVE_ALIASES)),
                "raw_payload": payload,
                "imported_at": imported_at,
                "updated_at": imported_at,
            },
        )

    if rows:
        stmt = insert(PlytixAttributeCatalog).values(rows)
        stmt = stmt.on_conflict_do_update(
            constraint="uq_plytix_attribute_catalog_source_code",
            set_={
                "label": stmt.excluded.label,
                "data_type": stmt.excluded.data_type,
                "group_name": stmt.excluded.group_name,
                "destination_hint": stmt.excluded.destination_hint,
                "is_required": stmt.excluded.is_required,
                "is_active": stmt.excluded.is_active,
                "raw_payload": stmt.excluded.raw_payload,
                "imported_at": stmt.excluded.imported_at,
                "updated_at": sa.func.now(),
            },
        )
        session.execute(stmt)

    return {
        "input_rows": len(df),
        "input_columns": len(columns),
        "upserted": len(rows),
        "skipped": skipped,
        "skipped_channel_attributes": skipped_channel,
        "duplicates_merged": duplicates_merged,
    }


def sync_catalog_to_attribute_def(session: Session, *, source_label: str = "plytix") -> int:
    """Mirror Plytix attribute catalog rows into attribute_def for mapping dropdowns."""
    rows = session.scalars(
        select(PlytixAttributeCatalog)
        .where(PlytixAttributeCatalog.source_label == source_label)
        .where(PlytixAttributeCatalog.is_active.is_(True))
        .order_by(PlytixAttributeCatalog.attribute_code)
    ).all()
    count = 0
    for row in rows:
        code = normalize_column_key(row.attribute_code)
        if not code:
            continue
        existing = session.get(AttributeDef, code)
        if existing is None:
            session.add(
                AttributeDef(
                    attribute_code=code,
                    frontend_label=row.label or code,
                    type=row.data_type or "text",
                    is_required=bool(row.is_required) if row.is_required is not None else False,
                )
            )
        else:
            if row.label:
                existing.frontend_label = row.label
            if row.data_type:
                existing.type = row.data_type
            if row.is_required is not None:
                existing.is_required = bool(row.is_required)
        count += 1
    return count


def export_plytix_attribute_catalog(session: Session) -> List[Dict[str, Any]]:
    rows = []
    for row in session.scalars(
        select(PlytixAttributeCatalog).order_by(PlytixAttributeCatalog.source_label, PlytixAttributeCatalog.attribute_code)
    ).all():
        rows.append(
            {
                "source_label": row.source_label,
                "attribute_code": row.attribute_code,
                "label": row.label or "",
                "data_type": row.data_type or "",
                "group_name": row.group_name or "",
                "destination_hint": row.destination_hint or "",
                "is_required": row.is_required if row.is_required is not None else "",
                "is_active": row.is_active,
                "imported_at": row.imported_at,
            }
        )
    return rows


def _append_catalog_row(rows: List[Dict[str, Any]], row: Dict[str, Any]) -> int:
    """Append or merge a row so one upsert batch never repeats the conflict key."""
    for existing in rows:
        if (
            existing["source_label"] == row["source_label"]
            and existing["attribute_code"] == row["attribute_code"]
        ):
            _merge_catalog_row(existing, row)
            return 1
    rows.append(row)
    return 0


def _merge_catalog_row(existing: Dict[str, Any], incoming: Dict[str, Any]) -> None:
    for key in ("data_type", "group_name", "destination_hint", "is_required"):
        if existing.get(key) in (None, "") and incoming.get(key) not in (None, ""):
            existing[key] = incoming[key]

    existing["is_active"] = bool(existing.get("is_active", True) or incoming.get("is_active", True))
    existing["label"] = _merge_text_values(existing.get("label"), incoming.get("label"))
    existing["raw_payload"] = _merge_raw_payloads(existing.get("raw_payload"), incoming.get("raw_payload"))


def _merge_text_values(current: Any, incoming: Any) -> Optional[str]:
    values = []
    for value in (current, incoming):
        text = _clean_text(value)
        if text and text not in values:
            values.append(text)
    return " | ".join(values) if values else None


def _merge_raw_payloads(current: Any, incoming: Any) -> Dict[str, Any]:
    merged: Dict[str, Any] = {}
    duplicate_labels: List[str] = []
    for payload in (current, incoming):
        if not isinstance(payload, dict):
            continue
        for key, value in payload.items():
            if key == "duplicate_column_names" and isinstance(value, list):
                for label in value:
                    if label not in duplicate_labels:
                        duplicate_labels.append(label)
                continue
            if key == "column_name":
                text = _clean_text(value)
                if text and text not in duplicate_labels:
                    duplicate_labels.append(text)
            elif key not in merged or merged.get(key) in (None, ""):
                merged[key] = value
    if duplicate_labels:
        merged["column_name"] = duplicate_labels[0]
        merged["duplicate_column_names"] = duplicate_labels
    return merged


def _first(row: Dict[str, Any], aliases: List[str]) -> Optional[str]:
    for alias in aliases:
        value = row.get(normalize_column_key(alias))
        if _clean_text(value) is not None:
            return _clean_text(value)
    return None


def _clean_text(value: Any) -> Optional[str]:
    if value is None:
        return None
    text = str(value).strip()
    if not text or text.lower() in {"nan", "none", "null"}:
        return None
    return text


def _to_bool(value: Any) -> Optional[bool]:
    text = _clean_text(value)
    if text is None:
        return None
    lowered = text.lower()
    if lowered in {"1", "true", "yes", "y", "on", "required", "mandatory"}:
        return True
    if lowered in {"0", "false", "no", "n", "off", "optional"}:
        return False
    return None


def _active_value(value: Any) -> bool:
    text = _clean_text(value)
    if text is None:
        return True
    lowered = text.lower()
    if lowered in {"0", "false", "no", "n", "off", "disabled", "inactive", "archived"}:
        return False
    return True