from __future__ import annotations

import json
from typing import Any, Dict, Iterable, List, Optional

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 ChannelAttributeAlias
from db.source_imports import normalize_column_key


def import_channel_attribute_alias_dataframe(session: Session, df: pd.DataFrame) -> Dict[str, int]:
    rows: List[Dict[str, Any]] = []
    skipped = 0
    for raw in df.to_dict(orient="records"):
        normalized = {normalize_column_key(k): v for k, v in raw.items()}
        channel_code = normalize_column_key(_first(normalized, "channel_code", "channel", "target_scope") or "")
        canonical_code = normalize_column_key(
            _first(normalized, "canonical_code", "canonical_attribute_code", "attribute_code") or ""
        )
        channel_attribute_code = normalize_column_key(
            _first(
                normalized,
                "channel_attribute_code",
                "existing_channel_code",
                "existing_attribute_code",
                "magento_attribute_code",
                "shopify_attribute_code",
                "plytix_attribute_code",
            )
            or ""
        )
        if not channel_code or not canonical_code or not channel_attribute_code:
            skipped += 1
            continue
        scope = normalize_column_key(_first(normalized, "mapping_scope", "scope") or "dynamic") or "dynamic"
        if scope not in {"static", "dynamic"}:
            scope = "dynamic"
        rows.append(
            {
                "channel_code": channel_code,
                "canonical_code": canonical_code,
                "channel_attribute_code": channel_attribute_code,
                "mapping_scope": scope,
                "action": normalize_column_key(_first(normalized, "action") or "use_existing") or "use_existing",
                "data_type": normalize_column_key(_first(normalized, "data_type") or "") or None,
                "transform_rule": _parse_json_dict(_first(normalized, "transform_rule")),
                "is_active": _to_bool(_first(normalized, "is_active"), default=True),
                "notes": _clean_text(_first(normalized, "notes", "reason")),
            }
        )

    if rows:
        stmt = insert(ChannelAttributeAlias).values(rows)
        stmt = stmt.on_conflict_do_update(
            constraint="uq_channel_attribute_alias_channel_canonical_attr",
            set_={
                "mapping_scope": stmt.excluded.mapping_scope,
                "action": stmt.excluded.action,
                "data_type": stmt.excluded.data_type,
                "transform_rule": stmt.excluded.transform_rule,
                "is_active": stmt.excluded.is_active,
                "notes": stmt.excluded.notes,
                "updated_at": sa.func.now(),
            },
        )
        session.execute(stmt)
    return {"input_rows": len(df), "upserted": len(rows), "skipped": skipped}


def export_channel_attribute_aliases(session: Session) -> List[Dict[str, Any]]:
    rows = []
    for alias in session.scalars(
        select(ChannelAttributeAlias).order_by(
            ChannelAttributeAlias.channel_code,
            ChannelAttributeAlias.canonical_code,
            ChannelAttributeAlias.channel_attribute_code,
        )
    ).all():
        rows.append(
            {
                "channel_code": alias.channel_code,
                "canonical_code": alias.canonical_code,
                "channel_attribute_code": alias.channel_attribute_code,
                "mapping_scope": alias.mapping_scope,
                "action": alias.action,
                "data_type": alias.data_type or "",
                "transform_rule": json.dumps(alias.transform_rule or {}, sort_keys=True),
                "is_active": alias.is_active,
                "notes": alias.notes or "",
            }
        )
    return rows


def channel_aliases_by_canonical(
    session: Session,
    channel_code: str,
    *,
    canonical_codes: Optional[Iterable[str]] = None,
) -> Dict[str, List[ChannelAttributeAlias]]:
    channel = normalize_column_key(channel_code)
    rows: Dict[str, List[ChannelAttributeAlias]] = {}
    if not channel:
        return rows
    stmt = (
        select(ChannelAttributeAlias)
        .where(ChannelAttributeAlias.channel_code == channel)
        .where(ChannelAttributeAlias.is_active.is_(True))
    )
    if canonical_codes is not None:
        wanted = {normalize_column_key(code) for code in canonical_codes if normalize_column_key(code)}
        if not wanted:
            return rows
        stmt = stmt.where(ChannelAttributeAlias.canonical_code.in_(sorted(wanted)))
    for alias in session.scalars(stmt).all():
        rows.setdefault(normalize_column_key(alias.canonical_code), []).append(alias)
    return rows


def channel_aliases_by_attribute(session: Session, channel_code: str) -> Dict[str, ChannelAttributeAlias]:
    channel = normalize_column_key(channel_code)
    rows: Dict[str, ChannelAttributeAlias] = {}
    if not channel:
        return rows
    for alias in session.scalars(
        select(ChannelAttributeAlias)
        .where(ChannelAttributeAlias.channel_code == channel)
        .where(ChannelAttributeAlias.is_active.is_(True))
    ).all():
        rows.setdefault(normalize_column_key(alias.channel_attribute_code), alias)
    return rows


def _first(row: Dict[str, Any], *keys: str) -> Any:
    for key in keys:
        value = row.get(normalize_column_key(key))
        if _clean_text(value) is not None:
            return 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, *, default: bool) -> bool:
    text = _clean_text(value)
    if text is None:
        return default
    return text.lower() in {"1", "true", "yes", "y", "on", "active"}


def _parse_json_dict(value: Any) -> Optional[Dict[str, Any]]:
    text = _clean_text(value)
    if not text:
        return None
    try:
        parsed = json.loads(text)
    except json.JSONDecodeError:
        return {"rule": text}
    return parsed if isinstance(parsed, dict) else {"value": parsed}