from __future__ import annotations

import json
import re
from dataclasses import dataclass
from typing import Any, Dict, Iterable, List, Optional, Sequence, Union

from sqlalchemy import and_, func, or_, select
from sqlalchemy.orm import Session
from sqlalchemy.sql import Select

from db.master_attributes import MASTER_PRODUCT_CORE_MAP
from db.master_filter_virtual_attrs import is_parent_filter_code
from db.models import MasterProduct, MasterProductAttributeValue, MasterProductRelation
from db.source_imports import normalize_column_key


@dataclass(frozen=True)
class MasterAttributeFilter:
    code: str
    value: str
    match: str = "exact"

    VALID_MATCHES = frozenset({
        "exact",
        "contains",
        "not_contains",
        "starts_with",
        "not_starts_with",
        "ends_with",
        "not_ends_with",
        "gt",
        "lt",
        "empty",
        "not_equal",
    })

    def normalized_code(self) -> str:
        return normalize_column_key(self.code)

    def normalized_value(self) -> str:
        return str(self.value or "").strip()

    def normalized_values(self) -> List[str]:
        return _split_filter_values(self.value)

    def is_empty_match(self) -> bool:
        return self.match == "empty"


# canonical attribute code -> master_product column (when stored on core row)
CANONICAL_TO_CORE_COLUMN: Dict[str, str] = {
    "title": "name",
    **{canonical: column for column, canonical in MASTER_PRODUCT_CORE_MAP.items()},
    **{column: column for column in MASTER_PRODUCT_CORE_MAP},
}

_ASSEMBLY_FILTER_CODES = frozenset({"assembled_or_rta", "assembly_type"})
_ASSEMBLY_ATTR_CODE = "assembled_or_rta"
_MULTI_VALUE_SPLIT_RE = re.compile(r"[\n\r|,;]+")
_NEGATIVE_MATCHES = frozenset({"not_equal", "not_contains", "not_starts_with", "not_ends_with"})


def _split_filter_values(value: Any) -> List[str]:
    if isinstance(value, (list, tuple, set)):
        raw_parts: List[str] = []
        for item in value:
            raw_parts.extend(_split_filter_values(item))
        return raw_parts
    parts = [
        part.strip()
        for part in _MULTI_VALUE_SPLIT_RE.split(str(value or ""))
        if part.strip()
    ]
    seen = set()
    out: List[str] = []
    for part in parts:
        key = part.lower()
        if key in seen:
            continue
        seen.add(key)
        out.append(part)
    return out


def _non_empty_text(column):
    """SQL expression: column has a non-blank text value."""
    return column.is_not(None) & (func.trim(column) != "")


def _empty_text(column):
    """SQL expression: column is null or blank."""
    return or_(column.is_(None), func.trim(column) == "")


def _normalize_match(raw: str) -> str:
    match = str(raw or "exact").strip().lower().replace("'", "").replace(" ", "_").replace("-", "_")
    aliases = {
        "equals": "exact",
        "equal": "exact",
        "startswith": "starts_with",
        "start_with": "starts_with",
        "endswith": "ends_with",
        "end_with": "ends_with",
        "greater_than": "gt",
        "greater": "gt",
        "less_than": "lt",
        "less": "lt",
        "not_exist": "empty",
        "not_exists": "empty",
        "missing": "empty",
        "is_empty": "empty",
        "does_not_exist": "empty",
        "not_equals": "not_equal",
        "not_equal_to": "not_equal",
        "does_not_equal": "not_equal",
        "doesnt_equal": "not_equal",
        "ne": "not_equal",
        "!=": "not_equal",
        "does_not_contain": "not_contains",
        "doesnt_contain": "not_contains",
        "not_contain": "not_contains",
        "does_not_start_with": "not_starts_with",
        "doesnt_start_with": "not_starts_with",
        "not_start_with": "not_starts_with",
        "does_not_end_with": "not_ends_with",
        "doesnt_end_with": "not_ends_with",
        "not_end_with": "not_ends_with",
    }
    match = aliases.get(match, match)
    if match not in MasterAttributeFilter.VALID_MATCHES:
        return "exact"
    return match


def _positive_text_match_kind(match: str) -> str:
    """Map negative match modes to the positive predicate they invert."""
    if match == "not_equal":
        return "exact"
    if match == "not_contains":
        return "contains"
    if match == "not_starts_with":
        return "starts_with"
    if match == "not_ends_with":
        return "ends_with"
    return match


def _apply_text_match(column, value: str, match: str):
    """Build SQL predicate for a text column and match mode."""
    positive = _positive_text_match_kind(match)
    if positive == "exact":
        predicate = func.lower(column) == value.lower()
    else:
        lowered = func.lower(column)
        val = value.lower()
        if positive == "contains":
            predicate = lowered.like(f"%{val}%")
        elif positive == "starts_with":
            predicate = lowered.like(f"{val}%")
        elif positive == "ends_with":
            predicate = lowered.like(f"%{val}")
        elif positive == "gt":
            predicate = lowered > val
        elif positive == "lt":
            predicate = lowered < val
        else:
            raise ValueError(f"Unsupported text match: {match}")

    if match in {"not_equal", "not_contains", "not_starts_with", "not_ends_with"}:
        return or_(_empty_text(column), ~predicate)
    return predicate


def parse_master_filters(raw: Any = None) -> List[MasterAttributeFilter]:
    """Parse filters from JSON string, list of dicts, or None."""
    if raw is None:
        return []
    payload = raw
    if isinstance(raw, str):
        text = raw.strip()
        if not text:
            return []
        payload = json.loads(text)
    if not isinstance(payload, list):
        raise ValueError("master_filters must be a JSON array")

    filters: List[MasterAttributeFilter] = []
    for item in payload:
        if not isinstance(item, dict):
            continue
        code = str(item.get("code") or item.get("attribute_code") or "").strip()
        if not code:
            continue
        match = _normalize_match(str(item.get("match") or "exact"))
        if match == "empty":
            filters.append(MasterAttributeFilter(code=code, value="", match="empty"))
            continue
        raw_value = item.get("value")
        values = _split_filter_values(raw_value)
        if not values:
            continue
        filters.append(MasterAttributeFilter(code=code, value="\n".join(values), match=match))
    return filters


def master_filters_to_dicts(filters: Sequence[MasterAttributeFilter]) -> List[Dict[str, str]]:
    return [
        {"code": item.normalized_code(), "value": item.normalized_value(), "match": item.match}
        for item in filters
    ]


def _resolve_core_column(code: str) -> Optional[str]:
    normalized = normalize_column_key(code)
    if normalized in _ASSEMBLY_FILTER_CODES:
        return None
    return CANONICAL_TO_CORE_COLUMN.get(normalized)


def _is_assembly_filter_code(code: str) -> bool:
    return normalize_column_key(code) in _ASSEMBLY_FILTER_CODES


def _is_parent_filter_code(code: str) -> bool:
    return is_parent_filter_code(code)


def _parse_bool_filter_value(value: str) -> bool:
    text = str(value or "").strip().lower()
    if text in {"1", "true", "yes", "y", "on"}:
        return True
    if text in {"0", "false", "no", "n", "off"}:
        return False
    raise ValueError(f"is_parent filter value must be 0/1 or true/false, got {value!r}")


def _parent_skus_subquery():
    return select(MasterProductRelation.parent_sku).distinct()


def _apply_is_parent_filter(
    stmt: Select,
    item: MasterAttributeFilter,
    product_id_column,
) -> Select:
    want_parent = _parse_bool_filter_value(item.normalized_value())
    is_parent = MasterProduct.sku.in_(_parent_skus_subquery())
    if want_parent:
        return stmt.where(is_parent)
    return stmt.where(~is_parent)


def _assembly_bucket_from_filter(value: str, match: str) -> Optional[str]:
    """Map UI values to canonical assembly buckets used in master_product.assembly_type."""
    text = str(value or "").strip().lower()
    if not text:
        return None
    if match == "starts_with":
        if text.startswith("unassem") or text.startswith("rta"):
            return "rta"
        if text.startswith("assem"):
            return "assembled"
        return None
    if match in {"exact", "not_equal", "contains"}:
        if text == "assembled" or (text.startswith("assem") and not text.startswith("unassem")):
            return "assembled"
        if text in {"rta", "unassembled"} or text.startswith("unassem") or "unassem" in text:
            return "rta"
    return None


def _unclassified_rta_hint():
    """Products missing assembly_type but clearly RTA from SKU/name (matches import heuristics)."""
    return _empty_text(MasterProduct.assembly_type) & or_(
        func.lower(MasterProduct.name).like("%unassembled%"),
        func.lower(MasterProduct.sku).like("rta-%"),
        func.lower(MasterProduct.sku).like("rta_%"),
    )


def _matches_assembled(product_id_column):
    attr_assembled = (
        select(MasterProductAttributeValue.product_id)
        .where(MasterProductAttributeValue.attribute_code == _ASSEMBLY_ATTR_CODE)
        .where(func.lower(MasterProductAttributeValue.value) == "assembled")
    )
    return or_(
        func.lower(MasterProduct.assembly_type) == "assembled",
        product_id_column.in_(attr_assembled),
    )


def _matches_rta(product_id_column):
    attr_rta = (
        select(MasterProductAttributeValue.product_id)
        .where(MasterProductAttributeValue.attribute_code == _ASSEMBLY_ATTR_CODE)
        .where(
            or_(
                func.lower(MasterProductAttributeValue.value) == "rta",
                func.lower(MasterProductAttributeValue.value) == "unassembled",
                func.lower(MasterProductAttributeValue.value) == "unassembled/rta",
                func.lower(MasterProductAttributeValue.value).like("unassem%"),
            )
        )
    )
    return or_(
        func.lower(MasterProduct.assembly_type) == "rta",
        product_id_column.in_(attr_rta),
        _unclassified_rta_hint(),
    )


def _assembly_empty_predicate(product_id_column):
    non_empty_attr = (
        select(MasterProductAttributeValue.product_id)
        .where(MasterProductAttributeValue.attribute_code == _ASSEMBLY_ATTR_CODE)
        .where(_non_empty_text(MasterProductAttributeValue.value))
    )
    return _empty_text(MasterProduct.assembly_type) & ~product_id_column.in_(non_empty_attr)


def _apply_assembly_filter(
    stmt: Select,
    item: MasterAttributeFilter,
    product_id_column,
) -> Select:
    if item.is_empty_match():
        return stmt.where(_assembly_empty_predicate(product_id_column))

    value = item.normalized_value()
    bucket = _assembly_bucket_from_filter(value, item.match)
    if bucket == "assembled":
        positive = _matches_assembled(product_id_column)
    elif bucket == "rta":
        positive = _matches_rta(product_id_column)
    else:
        # Fallback: raw text against core column and attribute value.
        core_pred = _apply_text_match(MasterProduct.assembly_type, value, item.match)
        attr_ids = (
            select(MasterProductAttributeValue.product_id)
            .where(MasterProductAttributeValue.attribute_code == _ASSEMBLY_ATTR_CODE)
            .where(_apply_text_match(MasterProductAttributeValue.value, value, item.match))
        )
        positive = or_(core_pred, product_id_column.in_(attr_ids))

    if item.match == "not_equal":
        # Avoid SQL three-valued logic from NOT(NULL); express negation explicitly.
        if bucket == "rta":
            return stmt.where(
                or_(
                    _matches_assembled(product_id_column),
                    _assembly_empty_predicate(product_id_column),
                )
            )
        if bucket == "assembled":
            return stmt.where(
                or_(
                    _matches_rta(product_id_column),
                    _assembly_empty_predicate(product_id_column),
                )
            )
        return stmt.where(~positive)
    return stmt.where(positive)


def _assembly_positive_predicate(value: str, match: str, product_id_column):
    bucket = _assembly_bucket_from_filter(value, match)
    if bucket == "assembled":
        return _matches_assembled(product_id_column)
    if bucket == "rta":
        return _matches_rta(product_id_column)
    core_pred = _apply_text_match(MasterProduct.assembly_type, value, _positive_text_match_kind(match))
    attr_ids = (
        select(MasterProductAttributeValue.product_id)
        .where(MasterProductAttributeValue.attribute_code == _ASSEMBLY_ATTR_CODE)
        .where(_apply_text_match(MasterProductAttributeValue.value, value, _positive_text_match_kind(match)))
    )
    return or_(core_pred, product_id_column.in_(attr_ids))


def _apply_multi_value_assembly_filter(
    stmt: Select,
    item: MasterAttributeFilter,
    values: Sequence[str],
    product_id_column,
) -> Select:
    positive = or_(*[_assembly_positive_predicate(value, item.match, product_id_column) for value in values])
    if item.match in _NEGATIVE_MATCHES:
        return stmt.where(or_(_assembly_empty_predicate(product_id_column), ~positive))
    return stmt.where(positive)


def _combine_text_predicates(column, values: Sequence[str], match: str):
    predicates = [_apply_text_match(column, value, match) for value in values]
    if match in _NEGATIVE_MATCHES:
        return and_(*predicates)
    return or_(*predicates)


def _apply_multi_value_core_filter(stmt: Select, column, values: Sequence[str], match: str) -> Select:
    return stmt.where(_combine_text_predicates(column, values, match))


def _apply_multi_value_attribute_filter(
    stmt: Select,
    *,
    code: str,
    values: Sequence[str],
    match: str,
    product_id_column,
) -> Select:
    if match in _NEGATIVE_MATCHES:
        positive_match = _positive_text_match_kind(match)
        matching = select(MasterProductAttributeValue.product_id).where(
            MasterProductAttributeValue.attribute_code == code
        )
        matching = matching.where(
            or_(*[_apply_text_match(MasterProductAttributeValue.value, value, positive_match) for value in values])
        )
        return stmt.where(~product_id_column.in_(matching))

    attr_stmt = select(MasterProductAttributeValue.product_id).where(
        MasterProductAttributeValue.attribute_code == code
    )
    attr_stmt = attr_stmt.where(
        or_(*[_apply_text_match(MasterProductAttributeValue.value, value, match) for value in values])
    )
    return stmt.where(product_id_column.in_(attr_stmt))


def apply_master_attribute_filters(
    stmt: Select,
    *,
    filters: Sequence[MasterAttributeFilter],
    product_id_column=None,
) -> Select:
    """Intersect stmt with one or more master attribute filters (AND semantics)."""
    if not filters:
        return stmt
    if product_id_column is None:
        product_id_column = MasterProduct.id

    for item in filters:
        code = item.normalized_code()
        if not code:
            continue

        if _is_assembly_filter_code(code):
            values = item.normalized_values()
            if len(values) > 1:
                stmt = _apply_multi_value_assembly_filter(stmt, item, values, product_id_column)
            else:
                stmt = _apply_assembly_filter(stmt, item, product_id_column)
            continue

        if _is_parent_filter_code(code):
            stmt = _apply_is_parent_filter(stmt, item, product_id_column)
            continue

        if item.is_empty_match():
            core_column = _resolve_core_column(code)
            if core_column and hasattr(MasterProduct, core_column):
                column = getattr(MasterProduct, core_column)
                stmt = stmt.where(_empty_text(column))
                continue

            non_empty_attr = (
                select(MasterProductAttributeValue.product_id)
                .where(MasterProductAttributeValue.attribute_code == code)
                .where(_non_empty_text(MasterProductAttributeValue.value))
            )
            stmt = stmt.where(~product_id_column.in_(non_empty_attr))
            continue

        value = item.normalized_value()
        values = item.normalized_values()
        if not values:
            continue

        core_column = _resolve_core_column(code)
        if core_column and hasattr(MasterProduct, core_column):
            column = getattr(MasterProduct, core_column)
            if len(values) > 1:
                stmt = _apply_multi_value_core_filter(stmt, column, values, item.match)
            else:
                stmt = stmt.where(_apply_text_match(column, values[0], item.match))
            continue

        if len(values) > 1:
            stmt = _apply_multi_value_attribute_filter(
                stmt,
                code=code,
                values=values,
                match=item.match,
                product_id_column=product_id_column,
            )
            continue

        if item.match == "not_equal":
            matching = select(MasterProductAttributeValue.product_id).where(
                MasterProductAttributeValue.attribute_code == code
            ).where(
                _apply_text_match(MasterProductAttributeValue.value, values[0], "exact")
            )
            stmt = stmt.where(~product_id_column.in_(matching))
            continue

        if item.match in {"not_contains", "not_starts_with", "not_ends_with"}:
            matching = select(MasterProductAttributeValue.product_id).where(
                MasterProductAttributeValue.attribute_code == code
            ).where(
                _apply_text_match(
                    MasterProductAttributeValue.value,
                    values[0],
                    {"not_contains": "contains", "not_starts_with": "starts_with", "not_ends_with": "ends_with"}[
                        item.match
                    ],
                )
            )
            stmt = stmt.where(~product_id_column.in_(matching))
            continue

        attr_stmt = select(MasterProductAttributeValue.product_id).where(
            MasterProductAttributeValue.attribute_code == code
        )
        attr_stmt = attr_stmt.where(
            _apply_text_match(MasterProductAttributeValue.value, values[0], item.match)
        )
        stmt = stmt.where(product_id_column.in_(attr_stmt))

    return stmt


def list_master_attribute_values(
    session: Session,
    attribute_code: str,
    *,
    search: Optional[str] = None,
    limit: int = 100,
) -> List[str]:
    """Distinct values for a master attribute (core column or attribute_value table).

    When a locked manual attribute vocabulary covers this code, return only those
    allowed values (pollution guard for filter / manual-attributes UI).
    """
    code = normalize_column_key(attribute_code)
    if not code:
        return []

    limit = max(1, min(int(limit or 100), 500))

    from db.manual_attributes import locked_values_for_code

    locked = locked_values_for_code(session, code)
    if locked is not None:
        ordered = list(locked)
        if search:
            needle = search.strip().lower()
            ordered = [value for value in ordered if needle in value.lower()]
        return ordered[:limit]

    if _is_parent_filter_code(code):
        values = ["0", "1", "false", "true", "no", "yes"]
        if search:
            needle = search.strip().lower()
            values = [value for value in values if needle in value.lower()]
        return values[:limit]
    if code in _ASSEMBLY_FILTER_CODES:
        core_values = [
            str(value).strip()
            for (value,) in session.execute(
                select(func.distinct(MasterProduct.assembly_type))
                .select_from(MasterProduct)
                .where(MasterProduct.is_active.is_(True))
                .where(_non_empty_text(MasterProduct.assembly_type))
                .order_by(MasterProduct.assembly_type)
                .limit(limit)
            ).all()
            if str(value).strip()
        ]
        attr_values = [
            str(value).strip()
            for (value,) in session.execute(
                select(func.distinct(MasterProductAttributeValue.value))
                .join(MasterProduct, MasterProduct.id == MasterProductAttributeValue.product_id)
                .where(MasterProduct.is_active.is_(True))
                .where(MasterProductAttributeValue.attribute_code == _ASSEMBLY_ATTR_CODE)
                .where(_non_empty_text(MasterProductAttributeValue.value))
                .order_by(MasterProductAttributeValue.value)
                .limit(limit)
            ).all()
            if str(value).strip()
        ]
        labels = set(core_values) | set(attr_values)
        # Present canonical UI labels alongside stored values.
        if "RTA" in labels or any(v.lower().startswith("unassem") for v in labels):
            labels.add("Unassembled")
        if "Assembled" in labels:
            labels.add("Assembled")
        ordered = sorted(labels, key=lambda v: v.lower())
        if search:
            needle = search.strip().lower()
            ordered = [v for v in ordered if needle in v.lower()]
        return ordered[:limit]

    core_column = _resolve_core_column(code)
    if core_column and hasattr(MasterProduct, core_column):
        column = getattr(MasterProduct, core_column)
        stmt = (
            select(func.distinct(column))
            .select_from(MasterProduct)
            .where(MasterProduct.is_active.is_(True))
            .where(column.is_not(None))
            .where(column != "")
        )
        if search:
            stmt = stmt.where(func.lower(column).like(f"%{search.strip().lower()}%"))
        stmt = stmt.order_by(column).limit(limit)
        return [str(value).strip() for (value,) in session.execute(stmt).all() if str(value).strip()]

    stmt = (
        select(func.distinct(MasterProductAttributeValue.value))
        .join(MasterProduct, MasterProduct.id == MasterProductAttributeValue.product_id)
        .where(MasterProduct.is_active.is_(True))
        .where(MasterProductAttributeValue.attribute_code == code)
        .where(MasterProductAttributeValue.value.is_not(None))
        .where(MasterProductAttributeValue.value != "")
    )
    if search:
        stmt = stmt.where(func.lower(MasterProductAttributeValue.value).like(f"%{search.strip().lower()}%"))
    stmt = stmt.order_by(MasterProductAttributeValue.value).limit(limit)
    return [str(value).strip() for (value,) in session.execute(stmt).all() if str(value).strip()]


def filter_kwargs_from_master_filters(
    filters: Union[Sequence[MasterAttributeFilter], Any, None],
) -> Dict[str, Any]:
    """Keyword bundle for functions that accept master_filters=."""
    parsed = filters if isinstance(filters, list) and filters and isinstance(filters[0], MasterAttributeFilter) else parse_master_filters(filters)
    return {"master_filters": parsed}