from __future__ import annotations

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

import pandas as pd
from sqlalchemy import select
from sqlalchemy.orm import Session

from db.models import (
    ChannelTaxonomyMapping,
    MagentoCategoryRegistry,
    PlytixProductSourceSnapshot,
    ShopifyCollectionRegistry,
)
from db.source_imports import normalize_column_key


def master_path_key(
    *,
    category_l1: Optional[str] = None,
    category_l2: Optional[str] = None,
    category_l3: Optional[str] = None,
    collection: Optional[str] = None,
) -> str:
    parts = [
        normalize_column_key(category_l1 or ""),
        normalize_column_key(category_l2 or ""),
        normalize_column_key(category_l3 or ""),
        normalize_column_key(collection or ""),
    ]
    return "|".join(parts)


def master_path_from_fields(fields: Dict[str, Any]) -> Tuple[str, str, str, str]:
    l1 = _first_text(fields, "category_l1", "category_l_1", "category1")
    l2 = _first_text(fields, "category_l2", "category_l_2", "category2")
    l3 = _first_text(fields, "category_l3", "category_l_3", "category3")
    coll = _first_text(fields, "collection", "collections")
    return l1 or "", l2 or "", l3 or "", coll or ""


def export_taxonomy_csv_rows(
    session: Session,
    channel: str,
    *,
    connection_id: Optional[int] = None,
) -> List[Dict[str, Any]]:
    channel = (channel or "").strip().lower()
    if channel == "magento":
        return _export_magento_categories(session, connection_id)
    if channel == "shopify":
        return _export_shopify_collections(session, connection_id)
    if channel == "plytix":
        return _export_plytix_category_paths(session)
    return []


def _export_magento_categories(session: Session, connection_id: Optional[int]) -> List[Dict[str, Any]]:
    stmt = select(MagentoCategoryRegistry).order_by(MagentoCategoryRegistry.path_names)
    if connection_id is not None:
        stmt = stmt.where(MagentoCategoryRegistry.connection_id == connection_id)
    rows: List[Dict[str, Any]] = []
    for row in session.scalars(stmt).all():
        path_names = row.path_names or row.name or ""
        parts = [p.strip() for p in path_names.split("/") if p.strip()]
        rows.append(
            {
                "taxonomy_kind": "category",
                "remote_id": row.category_id,
                "handle_or_path": path_names,
                "title": row.name,
                "level": row.level,
                "is_active": row.is_active,
                "category_l1": parts[0] if len(parts) > 0 else "",
                "category_l2": parts[1] if len(parts) > 1 else "",
                "category_l3": parts[2] if len(parts) > 2 else "",
                "collection": "",
            }
        )
    return rows


def _export_shopify_collections(session: Session, connection_id: Optional[int]) -> List[Dict[str, Any]]:
    from db.models import ShopifyTaxonomyRegistry

    stmt = select(ShopifyCollectionRegistry).order_by(ShopifyCollectionRegistry.title)
    if connection_id is not None:
        stmt = stmt.where(ShopifyCollectionRegistry.connection_id == connection_id)
    rows: List[Dict[str, Any]] = []
    for row in session.scalars(stmt).all():
        rows.append(
            {
                "taxonomy_kind": row.collection_type or "collection",
                "remote_id": row.collection_id,
                "handle_or_path": row.handle or "",
                "title": row.title,
                "level": "",
                "is_active": True,
                "category_l1": "",
                "category_l2": "",
                "category_l3": "",
                "collection": row.title,
                "product_count": row.product_count,
            }
        )
    for row in session.scalars(select(ShopifyTaxonomyRegistry).order_by(ShopifyTaxonomyRegistry.full_name)).all():
        rows.append(
            {
                "taxonomy_kind": "product_category",
                "remote_id": row.taxonomy_id,
                "handle_or_path": row.full_name,
                "title": row.name,
                "level": row.level,
                "is_active": True,
                "category_l1": "",
                "category_l2": "",
                "category_l3": "",
                "collection": "",
                "is_leaf": row.is_leaf,
            }
        )
    return rows


def _export_plytix_category_paths(session: Session) -> List[Dict[str, Any]]:
    seen: set[str] = set()
    rows: List[Dict[str, Any]] = []
    stmt = select(PlytixProductSourceSnapshot.payload).where(PlytixProductSourceSnapshot.valid_to.is_(None))
    for (payload,) in session.execute(stmt).all():
        if not isinstance(payload, dict):
            continue
        l1 = _first_text(payload, "category_l1", "category_l_1", "category1", "categories")
        l2 = _first_text(payload, "category_l2", "category_l_2", "category2")
        l3 = _first_text(payload, "category_l3", "category_l_3", "category3")
        coll = _first_text(payload, "collection", "collections")
        key = master_path_key(category_l1=l1, category_l2=l2, category_l3=l3, collection=coll)
        if not key.replace("|", "").strip() or key in seen:
            continue
        seen.add(key)
        rows.append(
            {
                "taxonomy_kind": "derived_path",
                "remote_id": "",
                "handle_or_path": key.replace("|", " / "),
                "title": coll or l3 or l2 or l1 or key,
                "level": "",
                "is_active": True,
                "category_l1": l1 or "",
                "category_l2": l2 or "",
                "category_l3": l3 or "",
                "collection": coll or "",
            }
        )
    rows.sort(key=lambda r: r["handle_or_path"])
    return rows


def import_taxonomy_mapping_csv(session: Session, df: pd.DataFrame) -> Dict[str, int]:
    inserted = 0
    updated = 0
    skipped = 0
    for _, row in df.iterrows():
        channel_code = str(row.get("channel_code") or "").strip().lower()
        if channel_code not in {"magento", "shopify", "plytix"}:
            skipped += 1
            continue
        l1 = _cell(row, "master_category_l1", "category_l1")
        l2 = _cell(row, "master_category_l2", "category_l2")
        l3 = _cell(row, "master_category_l3", "category_l3")
        coll = _cell(row, "master_collection", "collection")
        path_key = _cell(row, "master_path_key") or master_path_key(
            category_l1=l1, category_l2=l2, category_l3=l3, collection=coll
        )
        if not path_key.replace("|", "").strip():
            skipped += 1
            continue
        connection_id = _optional_int(row.get("connection_id"))
        existing = session.scalar(
            select(ChannelTaxonomyMapping)
            .where(ChannelTaxonomyMapping.channel_code == channel_code)
            .where(ChannelTaxonomyMapping.master_path_key == path_key)
            .where(
                ChannelTaxonomyMapping.connection_id.is_(None)
                if connection_id is None
                else ChannelTaxonomyMapping.connection_id == connection_id
            )
        )
        payload = {
            "master_category_l1": l1,
            "master_category_l2": l2,
            "master_category_l3": l3,
            "master_collection": coll,
            "master_path_key": path_key,
            "channel_taxonomy_kind": _cell(row, "channel_taxonomy_kind") or "collection",
            "channel_remote_id": _cell(row, "channel_remote_id", "remote_id"),
            "channel_handle_or_path": _cell(row, "channel_handle_or_path", "handle_or_path"),
            "transform_rule": _json_cell(row.get("transform_rule") or row.get("rule")),
            "is_active": str(row.get("is_active", "true")).strip().lower() not in {"0", "false", "no"},
            "notes": _cell(row, "notes"),
        }
        if existing is None:
            session.add(ChannelTaxonomyMapping(channel_code=channel_code, connection_id=connection_id, **payload))
            inserted += 1
        else:
            for key, value in payload.items():
                setattr(existing, key, value)
            updated += 1
    session.flush()
    return {"inserted": inserted, "updated": updated, "skipped": skipped}


def upsert_taxonomy_mapping(
    session: Session,
    *,
    channel_code: str,
    master_path_key: str,
    channel_remote_id: str,
    channel_taxonomy_kind: str = "collection",
    connection_id: Optional[int] = None,
    master_category_l1: Optional[str] = None,
    master_category_l2: Optional[str] = None,
    master_category_l3: Optional[str] = None,
    master_collection: Optional[str] = None,
    channel_handle_or_path: Optional[str] = None,
    transform_rule: Optional[Dict[str, Any]] = None,
    is_active: bool = True,
    notes: Optional[str] = None,
    mapping_id: Optional[int] = None,
) -> Dict[str, Any]:
    channel = str(channel_code or "").strip().lower()
    path_key = str(master_path_key or "").strip()
    if channel not in {"magento", "shopify", "plytix"}:
        raise ValueError("channel_code must be magento, shopify, or plytix")
    if not path_key:
        raise ValueError("master_path_key is required")

    row = None
    if mapping_id is not None:
        row = session.get(ChannelTaxonomyMapping, int(mapping_id))
    if row is None:
        stmt = (
            select(ChannelTaxonomyMapping)
            .where(ChannelTaxonomyMapping.channel_code == channel)
            .where(ChannelTaxonomyMapping.master_path_key == path_key)
        )
        if connection_id is None:
            stmt = stmt.where(ChannelTaxonomyMapping.connection_id.is_(None))
        else:
            stmt = stmt.where(ChannelTaxonomyMapping.connection_id == connection_id)
        row = session.scalar(stmt.limit(1))

    payload = {
        "channel_code": channel,
        "connection_id": connection_id,
        "master_category_l1": master_category_l1,
        "master_category_l2": master_category_l2,
        "master_category_l3": master_category_l3,
        "master_collection": master_collection,
        "master_path_key": path_key,
        "channel_taxonomy_kind": channel_taxonomy_kind,
        "channel_remote_id": str(channel_remote_id),
        "channel_handle_or_path": channel_handle_or_path,
        "transform_rule": transform_rule,
        "is_active": bool(is_active),
        "notes": notes,
    }
    if row is None:
        row = ChannelTaxonomyMapping(**payload)
        session.add(row)
    else:
        for key, value in payload.items():
            setattr(row, key, value)
    session.flush()
    return {
        "id": row.id,
        "channel_code": row.channel_code,
        "connection_id": row.connection_id,
        "master_path_key": row.master_path_key,
        "channel_remote_id": row.channel_remote_id,
        "channel_taxonomy_kind": row.channel_taxonomy_kind,
        "is_active": row.is_active,
    }


def deactivate_taxonomy_mapping(session: Session, mapping_id: int) -> bool:
    row = session.get(ChannelTaxonomyMapping, int(mapping_id))
    if row is None:
        return False
    row.is_active = False
    session.flush()
    return True


def list_taxonomy_mappings(
    session: Session,
    *,
    channel_code: str,
    connection_id: Optional[int] = None,
) -> List[Dict[str, Any]]:
    stmt = (
        select(ChannelTaxonomyMapping)
        .where(ChannelTaxonomyMapping.channel_code == channel_code)
        .order_by(ChannelTaxonomyMapping.master_path_key)
    )
    if connection_id is not None:
        stmt = stmt.where(
            (ChannelTaxonomyMapping.connection_id == connection_id)
            | (ChannelTaxonomyMapping.connection_id.is_(None))
        )
    return [
        {
            "id": row.id,
            "channel_code": row.channel_code,
            "connection_id": row.connection_id,
            "master_category_l1": row.master_category_l1,
            "master_category_l2": row.master_category_l2,
            "master_category_l3": row.master_category_l3,
            "master_collection": row.master_collection,
            "master_path_key": row.master_path_key,
            "channel_taxonomy_kind": row.channel_taxonomy_kind,
            "channel_remote_id": row.channel_remote_id,
            "channel_handle_or_path": row.channel_handle_or_path,
            "transform_rule": row.transform_rule or {},
            "is_active": row.is_active,
            "notes": row.notes,
        }
        for row in session.scalars(stmt).all()
    ]


def resolve_taxonomy_mappings_for_fields(
    session: Session,
    *,
    channel_code: str,
    fields: Dict[str, Any],
    connection_id: Optional[int] = None,
) -> List[ChannelTaxonomyMapping]:
    l1, l2, l3, coll = master_path_from_fields(fields)
    keys = {
        master_path_key(category_l1=l1, category_l2=l2, category_l3=l3, collection=coll),
        master_path_key(category_l1=l1, category_l2=l2, category_l3=l3),
        master_path_key(category_l1=l1, category_l2=l2),
        master_path_key(category_l1=l1),
        master_path_key(collection=coll),
    }
    keys = {k for k in keys if k.replace("|", "").strip()}
    if not keys:
        return []
    stmt = (
        select(ChannelTaxonomyMapping)
        .where(ChannelTaxonomyMapping.channel_code == channel_code)
        .where(ChannelTaxonomyMapping.is_active.is_(True))
        .where(ChannelTaxonomyMapping.master_path_key.in_(sorted(keys)))
    )
    if connection_id is not None:
        stmt = stmt.where(
            (ChannelTaxonomyMapping.connection_id == connection_id)
            | (ChannelTaxonomyMapping.connection_id.is_(None))
        )
    rows = list(session.scalars(stmt).all())
    rows.sort(key=lambda r: (0 if r.connection_id else 1, r.master_path_key))
    seen: set[str] = set()
    out: List[ChannelTaxonomyMapping] = []
    for row in rows:
        if row.master_path_key in seen:
            continue
        seen.add(row.master_path_key)
        out.append(row)
    return out


def _first_text(payload: Dict[str, Any], *keys: str) -> Optional[str]:
    norm = {normalize_column_key(str(k)): v for k, v in payload.items()}
    for key in keys:
        nk = normalize_column_key(key)
        value = payload.get(key)
        if value is None:
            value = norm.get(nk)
        text = str(value or "").strip()
        if text and text.lower() not in {"nan", "none", "null"}:
            return text
    return None


def _cell(row: pd.Series, *names: str) -> Optional[str]:
    for name in names:
        if name in row.index:
            text = str(row.get(name) or "").strip()
            if text and text.lower() not in {"nan", "none", "null"}:
                return text
    return None


def _optional_int(value: Any) -> Optional[int]:
    text = str(value or "").strip()
    if not text:
        return None
    try:
        return int(text)
    except (TypeError, ValueError):
        return None


def _json_cell(value: Any) -> Optional[Dict[str, Any]]:
    text = str(value or "").strip()
    if not text or text.lower() in {"nan", "none", "null"}:
        return None
    try:
        parsed = json.loads(text)
    except json.JSONDecodeError:
        return {"raw": text}
    return parsed if isinstance(parsed, dict) else {"value": parsed}