from __future__ import annotations

from typing import Any, Dict, List, Optional, 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 MasterProduct, ProductChannelAssignment
from db.source_imports import normalize_column_key


ACTIVE_STATUS = "active"
REMOVED_STATUS = "removed"


def import_channel_assignment_dataframe(
    session: Session,
    df: pd.DataFrame,
    *,
    reason: str = "csv_import",
) -> Dict[str, Any]:
    """Import sku→channel assignments from CSV: columns sku, channel_code[, status].

    Also accepts one wide row per SKU with a pipe/comma separated `channels` column.
    """
    requested: List[Dict[str, Any]] = []
    skipped = 0
    for raw in df.to_dict(orient="records"):
        record = {normalize_column_key(k): v for k, v in raw.items()}
        sku = _clean(record.get("sku") or record.get("item"))
        if not sku:
            skipped += 1
            continue
        channels = _channels_from_record(record)
        if not channels:
            skipped += 1
            continue
        status = _status(record.get("status") or record.get("assignment_status"))
        for channel in channels:
            requested.append({"sku": sku, "channel_code": channel, "assignment_status": status})
    return _apply_assignments(session, requested, reason=reason, skipped=skipped, input_rows=len(df))


def assign_skus(
    session: Session,
    *,
    skus: List[str],
    channel_code: str,
    reason: str = "manual",
) -> Dict[str, Any]:
    return _apply_assignments(
        session,
        [{"sku": str(s).strip(), "channel_code": channel_code, "assignment_status": ACTIVE_STATUS} for s in skus if str(s).strip()],
        reason=reason,
        skipped=0,
        input_rows=len(skus),
    )


def remove_skus(
    session: Session,
    *,
    skus: List[str],
    channel_code: str,
    reason: str = "manual_removal",
) -> Dict[str, Any]:
    return _apply_assignments(
        session,
        [{"sku": str(s).strip(), "channel_code": channel_code, "assignment_status": REMOVED_STATUS} for s in skus if str(s).strip()],
        reason=reason,
        skipped=0,
        input_rows=len(skus),
    )


def list_assignments(
    session: Session,
    *,
    channel_code: Optional[str] = None,
    status: Optional[str] = None,
    limit: Optional[int] = None,
) -> List[Dict[str, Any]]:
    stmt = select(ProductChannelAssignment).order_by(
        ProductChannelAssignment.channel_code, ProductChannelAssignment.sku
    )
    if channel_code:
        stmt = stmt.where(ProductChannelAssignment.channel_code == normalize_column_key(channel_code))
    if status:
        stmt = stmt.where(ProductChannelAssignment.assignment_status == status)
    if limit:
        stmt = stmt.limit(limit)
    return [
        {
            "sku": row.sku,
            "channel_code": row.channel_code,
            "assignment_status": row.assignment_status,
            "reason": row.reason or "",
            "updated_at": row.updated_at.isoformat() if row.updated_at else "",
        }
        for row in session.scalars(stmt).all()
    ]


def active_skus_for_channel(session: Session, channel_code: str) -> Set[str]:
    channel = normalize_column_key(channel_code)
    return {
        sku
        for (sku,) in session.execute(
            select(ProductChannelAssignment.sku)
            .where(ProductChannelAssignment.channel_code == channel)
            .where(ProductChannelAssignment.assignment_status == ACTIVE_STATUS)
        ).all()
    }


def _apply_assignments(
    session: Session,
    requested: List[Dict[str, Any]],
    *,
    reason: str,
    skipped: int,
    input_rows: int,
) -> Dict[str, Any]:
    requested = _dedupe(requested)
    skus = sorted({r["sku"] for r in requested})
    product_ids = {
        sku: product_id
        for sku, product_id in session.execute(
            select(MasterProduct.sku, MasterProduct.id).where(MasterProduct.sku.in_(skus))
        ).all()
    }
    unknown_skus = sorted(set(skus) - set(product_ids))

    rows = []
    for item in requested:
        product_id = product_ids.get(item["sku"])
        if product_id is None:
            continue
        rows.append(
            {
                "product_id": product_id,
                "sku": item["sku"],
                "channel_code": item["channel_code"],
                "assignment_status": item["assignment_status"],
                "reason": reason,
                "payload": None,
            }
        )
    if rows:
        stmt = insert(ProductChannelAssignment).values(rows)
        stmt = stmt.on_conflict_do_update(
            constraint="uq_product_channel_assignment_product_channel",
            set_={
                "sku": stmt.excluded.sku,
                "assignment_status": stmt.excluded.assignment_status,
                "reason": stmt.excluded.reason,
                "updated_at": sa.func.now(),
            },
        )
        session.execute(stmt)
    return {
        "input_rows": input_rows,
        "upserted": len(rows),
        "skipped": skipped,
        "unknown_skus": unknown_skus[:100],
        "unknown_sku_count": len(unknown_skus),
    }


def _channels_from_record(record: Dict[str, Any]) -> List[str]:
    single = _clean(record.get("channel_code") or record.get("channel"))
    multi = _clean(record.get("channels") or record.get("product_websites"))
    raw_values: List[str] = []
    if single:
        raw_values.append(single)
    if multi:
        raw_values.extend(multi.replace("|", ",").split(","))
    channels = []
    for value in raw_values:
        channel = normalize_column_key(value)
        if channel and channel not in channels:
            channels.append(channel)
    return channels


def _dedupe(rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
    by_key: Dict[tuple, Dict[str, Any]] = {}
    for row in rows:
        by_key[(row["sku"], row["channel_code"])] = row
    return list(by_key.values())


def _status(value: Any) -> str:
    text = str(value or "").strip().lower()
    if text in {"removed", "remove", "inactive", "off", "0", "false", "no"}:
        return REMOVED_STATUS
    return ACTIVE_STATUS


def _clean(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