#  Licensed to Elasticsearch B.V. under one or more contributor
#  license agreements. See the NOTICE file distributed with
#  this work for additional information regarding copyright
#  ownership. Elasticsearch B.V. licenses this file to you under
#  the Apache License, Version 2.0 (the "License"); you may
#  not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
# 	http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing,
#  software distributed under the License is distributed on an
#  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
#  KIND, either express or implied.  See the License for the
#  specific language governing permissions and limitations
#  under the License.


import collections.abc
from copy import copy
from typing import (
    TYPE_CHECKING,
    Any,
    Callable,
    ClassVar,
    Dict,
    Generic,
    Iterable,
    Iterator,
    List,
    Mapping,
    Optional,
    Tuple,
    Type,
    Union,
    cast,
)

from elastic_transport.client_utils import DEFAULT
from typing_extensions import Self, TypeAlias, TypeVar

from .exceptions import UnknownDslObject, ValidationException

if TYPE_CHECKING:
    from elastic_transport import ObjectApiResponse

    from elasticsearch import AsyncElasticsearch, Elasticsearch

    from .document_base import DocumentOptions
    from .field import Field
    from .index_base import IndexBase
    from .response import Hit  # noqa: F401
    from .types import Hit as HitBaseType

UsingType: TypeAlias = Union[str, "Elasticsearch"]
AsyncUsingType: TypeAlias = Union[str, "AsyncElasticsearch"]
AnyUsingType: TypeAlias = Union[str, "Elasticsearch", "AsyncElasticsearch"]

_ValT = TypeVar("_ValT")  # used by AttrDict
_R = TypeVar("_R", default="Hit")  # used by Search and Response classes

SKIP_VALUES = ("", None)
EXPAND__TO_DOT = True

DOC_META_FIELDS = frozenset(
    (
        "id",
        "routing",
    )
)

META_FIELDS = frozenset(
    (
        # Elasticsearch metadata fields, except 'type'
        "index",
        "using",
        "score",
        "version",
        "seq_no",
        "primary_term",
    )
).union(DOC_META_FIELDS)


def _wrap(val: Any, obj_wrapper: Optional[Callable[[Any], Any]] = None) -> Any:
    if isinstance(val, dict):
        return AttrDict(val) if obj_wrapper is None else obj_wrapper(val)
    if isinstance(val, list):
        return AttrList(val)
    return val


def _recursive_to_dict(value: Any) -> Any:
    if hasattr(value, "to_dict"):
        return value.to_dict()
    elif isinstance(value, dict) or isinstance(value, AttrDict):
        return {k: _recursive_to_dict(v) for k, v in value.items()}
    elif isinstance(value, list) or isinstance(value, AttrList):
        return [recursive_to_dict(elem) for elem in value]
    else:
        return value


class AttrList(Generic[_ValT]):
    def __init__(
        self, l: List[_ValT], obj_wrapper: Optional[Callable[[_ValT], Any]] = None
    ):
        # make iterables into lists
        if not isinstance(l, list):
            l = list(l)
        self._l_ = l
        self._obj_wrapper = obj_wrapper

    def __repr__(self) -> str:
        return repr(self._l_)

    def __eq__(self, other: Any) -> bool:
        if isinstance(other, AttrList):
            return other._l_ == self._l_
        # make sure we still equal to a dict with the same data
        return bool(other == self._l_)

    def __ne__(self, other: Any) -> bool:
        return not self == other

    def __getitem__(self, k: Union[int, slice]) -> Any:
        l = self._l_[k]
        if isinstance(k, slice):
            return AttrList[_ValT](l, obj_wrapper=self._obj_wrapper)  # type: ignore[arg-type]
        return _wrap(l, self._obj_wrapper)

    def __setitem__(self, k: int, value: _ValT) -> None:
        self._l_[k] = value

    def __iter__(self) -> Iterator[Any]:
        return map(lambda i: _wrap(i, self._obj_wrapper), self._l_)

    def __len__(self) -> int:
        return len(self._l_)

    def __nonzero__(self) -> bool:
        return bool(self._l_)

    __bool__ = __nonzero__

    def __getattr__(self, name: str) -> Any:
        return getattr(self._l_, name)

    def __getstate__(self) -> Tuple[List[_ValT], Optional[Callable[[_ValT], Any]]]:
        return self._l_, self._obj_wrapper

    def __setstate__(
        self, state: Tuple[List[_ValT], Optional[Callable[[_ValT], Any]]]
    ) -> None:
        self._l_, self._obj_wrapper = state

    def to_list(self) -> List[_ValT]:
        return self._l_


class AttrDict(Generic[_ValT]):
    """
    Helper class to provide attribute like access (read and write) to
    dictionaries. Used to provide a convenient way to access both results and
    nested dsl dicts.
    """

    _d_: Dict[str, _ValT]
    RESERVED: Dict[str, str] = {"from_": "from"}

    def __init__(self, d: Dict[str, _ValT]):
        # assign the inner dict manually to prevent __setattr__ from firing
        super().__setattr__("_d_", d)

    def __contains__(self, key: object) -> bool:
        return key in self._d_

    def __nonzero__(self) -> bool:
        return bool(self._d_)

    __bool__ = __nonzero__

    def __dir__(self) -> List[str]:
        # introspection for auto-complete in IPython etc
        return list(self._d_.keys())

    def __eq__(self, other: Any) -> bool:
        if isinstance(other, AttrDict):
            return other._d_ == self._d_
        # make sure we still equal to a dict with the same data
        return bool(other == self._d_)

    def __ne__(self, other: Any) -> bool:
        return not self == other

    def __repr__(self) -> str:
        r = repr(self._d_)
        if len(r) > 60:
            r = r[:60] + "...}"
        return r

    def __getstate__(self) -> Tuple[Dict[str, _ValT]]:
        return (self._d_,)

    def __setstate__(self, state: Tuple[Dict[str, _ValT]]) -> None:
        super().__setattr__("_d_", state[0])

    def __getattr__(self, attr_name: str) -> Any:
        try:
            return self.__getitem__(attr_name)
        except KeyError:
            raise AttributeError(
                f"{self.__class__.__name__!r} object has no attribute {attr_name!r}"
            )

    def __delattr__(self, attr_name: str) -> None:
        try:
            del self._d_[self.RESERVED.get(attr_name, attr_name)]
        except KeyError:
            raise AttributeError(
                f"{self.__class__.__name__!r} object has no attribute {attr_name!r}"
            )

    def __getitem__(self, key: str) -> Any:
        return _wrap(self._d_[self.RESERVED.get(key, key)])

    def __setitem__(self, key: str, value: _ValT) -> None:
        self._d_[self.RESERVED.get(key, key)] = value

    def __delitem__(self, key: str) -> None:
        del self._d_[self.RESERVED.get(key, key)]

    def __setattr__(self, name: str, value: _ValT) -> None:
        # the __orig__class__ attribute has to be treated as an exception, as
        # is it added to an object when it is instantiated with type arguments
        if (
            name in self._d_ or not hasattr(self.__class__, name)
        ) and name != "__orig_class__":
            self._d_[self.RESERVED.get(name, name)] = value
        else:
            # there is an attribute on the class (could be property, ..) - don't add it as field
            super().__setattr__(name, value)

    def __iter__(self) -> Iterator[str]:
        return iter(self._d_)

    def to_dict(self, recursive: bool = False) -> Dict[str, _ValT]:
        return cast(
            Dict[str, _ValT], _recursive_to_dict(self._d_) if recursive else self._d_
        )

    def keys(self) -> Iterable[str]:
        return self._d_.keys()

    def items(self) -> Iterable[Tuple[str, _ValT]]:
        return self._d_.items()


class DslMeta(type):
    """
    Base Metaclass for DslBase subclasses that builds a registry of all classes
    for given DslBase subclass (== all the query types for the Query subclass
    of DslBase).

    It then uses the information from that registry (as well as `name` and
    `shortcut` attributes from the base class) to construct any subclass based
    on it's name.

    For typical use see `QueryMeta` and `Query` in `elasticsearch.dsl.query`.
    """

    name: str
    _classes: Dict[str, type]
    _type_name: str
    _types: ClassVar[Dict[str, Type["DslBase"]]] = {}

    def __init__(cls, name: str, bases: Tuple[type, ...], attrs: Dict[str, Any]):
        super().__init__(name, bases, attrs)
        # skip for DslBase
        if not hasattr(cls, "_type_shortcut"):
            return
        if not cls.name:
            # abstract base class, register it's shortcut
            cls._types[cls._type_name] = cls._type_shortcut
            # and create a registry for subclasses
            if not hasattr(cls, "_classes"):
                cls._classes = {}
        elif cls.name not in cls._classes:
            # normal class, register it
            cls._classes[cls.name] = cls

    @classmethod
    def get_dsl_type(cls, name: str) -> Type["DslBase"]:
        try:
            return cls._types[name]
        except KeyError:
            raise UnknownDslObject(f"DSL type {name} does not exist.")


class DslBase(metaclass=DslMeta):
    """
    Base class for all DSL objects - queries, filters, aggregations etc. Wraps
    a dictionary representing the object's json.

    Provides several feature:
        - attribute access to the wrapped dictionary (.field instead of ['field'])
        - _clone method returning a copy of self
        - to_dict method to serialize into dict (to be sent via elasticsearch-py)
        - basic logical operators (&, | and ~) using a Bool(Filter|Query) TODO:
          move into a class specific for Query/Filter
        - respects the definition of the class and (de)serializes it's
          attributes based on the `_param_defs` definition (for example turning
          all values in the `must` attribute into Query objects)
    """

    _param_defs: ClassVar[Dict[str, Dict[str, Union[str, bool]]]] = {}

    @classmethod
    def get_dsl_class(
        cls: Type[Self], name: str, default: Optional[str] = None
    ) -> Type[Self]:
        try:
            return cls._classes[name]
        except KeyError:
            if default is not None:
                return cls._classes[default]
            raise UnknownDslObject(
                f"DSL class `{name}` does not exist in {cls._type_name}."
            )

    def __init__(self, _expand__to_dot: Optional[bool] = None, **params: Any) -> None:
        if _expand__to_dot is None:
            _expand__to_dot = EXPAND__TO_DOT
        self._params: Dict[str, Any] = {}
        for pname, pvalue in params.items():
            if pvalue == DEFAULT:
                continue
            # expand "__" to dots
            if "__" in pname and _expand__to_dot:
                pname = pname.replace("__", ".")
            # convert instrumented fields to string
            if type(pvalue).__name__ == "InstrumentedField":
                pvalue = str(pvalue)
            self._setattr(pname, pvalue)

    def _repr_params(self) -> str:
        """Produce a repr of all our parameters to be used in __repr__."""
        return ", ".join(
            f"{n.replace('.', '__')}={v!r}"
            for (n, v) in sorted(self._params.items())
            # make sure we don't include empty typed params
            if "type" not in self._param_defs.get(n, {}) or v
        )

    def __repr__(self) -> str:
        return f"{self.__class__.__name__}({self._repr_params()})"

    def __eq__(self, other: Any) -> bool:
        return isinstance(other, self.__class__) and other.to_dict() == self.to_dict()

    def __ne__(self, other: Any) -> bool:
        return not self == other

    def __setattr__(self, name: str, value: Any) -> None:
        if name.startswith("_"):
            return super().__setattr__(name, value)
        return self._setattr(name, value)

    def _setattr(self, name: str, value: Any) -> None:
        # if this attribute has special type assigned to it...
        name = AttrDict.RESERVED.get(name, name)
        if name in self._param_defs:
            pinfo = self._param_defs[name]

            if "type" in pinfo:
                # get the shortcut used to construct this type (query.Q, aggs.A, etc)
                shortcut = self.__class__.get_dsl_type(str(pinfo["type"]))

                # list of dict(name -> DslBase)
                if pinfo.get("multi") and pinfo.get("hash"):
                    if not isinstance(value, (tuple, list)):
                        value = (value,)
                    value = list(
                        {k: shortcut(v) for (k, v) in obj.items()} for obj in value
                    )
                elif pinfo.get("multi"):
                    if not isinstance(value, (tuple, list)):
                        value = (value,)
                    value = list(map(shortcut, value))

                # dict(name -> DslBase), make sure we pickup all the objs
                elif pinfo.get("hash"):
                    value = {k: shortcut(v) for (k, v) in value.items()}

                # single value object, just convert
                else:
                    value = shortcut(value)
        self._params[name] = value

    def __getattr__(self, name: str) -> Any:
        if name.startswith("_"):
            raise AttributeError(
                f"{self.__class__.__name__!r} object has no attribute {name!r}"
            )

        value = None
        try:
            value = self._params[name]
        except KeyError:
            # compound types should never throw AttributeError and return empty
            # container instead
            if name in self._param_defs:
                pinfo = self._param_defs[name]
                if pinfo.get("multi"):
                    value = self._params.setdefault(name, [])
                elif pinfo.get("hash"):
                    value = self._params.setdefault(name, {})
        if value is None:
            raise AttributeError(
                f"{self.__class__.__name__!r} object has no attribute {name!r}"
            )

        # wrap nested dicts in AttrDict for convenient access
        if isinstance(value, dict):
            return AttrDict(value)
        return value

    def to_dict(self) -> Dict[str, Any]:
        """
        Serialize the DSL object to plain dict
        """
        d = {}
        for pname, value in self._params.items():
            pinfo = self._param_defs.get(pname)

            # typed param
            if pinfo and "type" in pinfo:
                # don't serialize empty lists and dicts for typed fields
                if value in ({}, []):
                    continue

                # list of dict(name -> DslBase)
                if pinfo.get("multi") and pinfo.get("hash"):
                    value = list(
                        {k: v.to_dict() for k, v in obj.items()} for obj in value
                    )

                # multi-values are serialized as list of dicts
                elif pinfo.get("multi"):
                    value = list(map(lambda x: x.to_dict(), value))

                # squash all the hash values into one dict
                elif pinfo.get("hash"):
                    value = {k: v.to_dict() for k, v in value.items()}

                # serialize single values
                else:
                    value = value.to_dict()

            # serialize anything with to_dict method
            elif hasattr(value, "to_dict"):
                value = value.to_dict()

            d[pname] = value
        return {self.name: d}

    def _clone(self) -> Self:
        c = self.__class__()
        for attr in self._params:
            c._params[attr] = copy(self._params[attr])
        return c


if TYPE_CHECKING:
    HitMetaBase = HitBaseType
else:
    HitMetaBase = AttrDict[Any]


class HitMeta(HitMetaBase):
    inner_hits: Mapping[str, Any]

    def __init__(
        self,
        document: Dict[str, Any],
        exclude: Tuple[str, ...] = ("_source", "_fields"),
    ):
        d = {
            k[1:] if k.startswith("_") else k: v
            for (k, v) in document.items()
            if k not in exclude
        }
        if "type" in d:
            # make sure we are consistent everywhere in python
            d["doc_type"] = d.pop("type")
        super().__init__(d)


class ObjectBase(AttrDict[Any]):
    _doc_type: "DocumentOptions"
    _index: "IndexBase"
    meta: HitMeta

    def __init__(self, meta: Optional[Dict[str, Any]] = None, **kwargs: Any):
        meta = meta or {}
        for k in list(kwargs):
            if k.startswith("_") and k[1:] in META_FIELDS:
                meta[k] = kwargs.pop(k)

        super(AttrDict, self).__setattr__("meta", HitMeta(meta))

        # process field defaults
        if hasattr(self, "_defaults"):
            for name in self._defaults:
                if name not in kwargs:
                    value = self._defaults[name]
                    if callable(value):
                        value = value()
                    kwargs[name] = value

        super().__init__(kwargs)

    @classmethod
    def __list_fields(cls) -> Iterator[Tuple[str, "Field", bool]]:
        """
        Get all the fields defined for our class, if we have an Index, try
        looking at the index mappings as well, mark the fields from Index as
        optional.
        """
        for name in cls._doc_type.mapping:
            field = cls._doc_type.mapping[name]
            yield name, field, False

        if hasattr(cls.__class__, "_index"):
            if not cls._index._mapping:
                return
            for name in cls._index._mapping:
                # don't return fields that are in _doc_type
                if name in cls._doc_type.mapping:
                    continue
                field = cls._index._mapping[name]
                yield name, field, True

    @classmethod
    def __get_field(cls, name: str) -> Optional["Field"]:
        try:
            return cls._doc_type.mapping[name]
        except KeyError:
            # fallback to fields on the Index
            if hasattr(cls, "_index") and cls._index._mapping:
                try:
                    return cls._index._mapping[name]
                except KeyError:
                    pass
            return None

    @classmethod
    def from_es(cls, hit: Union[Dict[str, Any], "ObjectApiResponse[Any]"]) -> Self:
        meta = hit.copy()
        data = meta.pop("_source", {})
        doc = cls(meta=meta)
        doc._from_dict(data)
        return doc

    def _from_dict(self, data: Dict[str, Any]) -> None:
        for k, v in data.items():
            f = self.__get_field(k)
            if f and f._coerce:
                v = f.deserialize(v)
            setattr(self, k, v)

    def __getstate__(self) -> Tuple[Dict[str, Any], Dict[str, Any]]:  # type: ignore[override]
        return self.to_dict(), self.meta._d_

    def __setstate__(self, state: Tuple[Dict[str, Any], Dict[str, Any]]) -> None:  # type: ignore[override]
        data, meta = state
        super(AttrDict, self).__setattr__("_d_", {})
        super(AttrDict, self).__setattr__("meta", HitMeta(meta))
        self._from_dict(data)

    def __getattr__(self, name: str) -> Any:
        try:
            return super().__getattr__(name)
        except AttributeError:
            f = self.__get_field(name)
            if f is not None and hasattr(f, "empty"):
                value = f.empty()
                if value not in SKIP_VALUES:
                    setattr(self, name, value)
                    value = getattr(self, name)
                return value
            raise

    def __setattr__(self, name: str, value: Any) -> None:
        if name in self.__class__._doc_type.mapping:
            self._d_[name] = value
        else:
            super().__setattr__(name, value)

    def to_dict(self, skip_empty: bool = True) -> Dict[str, Any]:
        out = {}
        for k, v in self._d_.items():
            # if this is a mapped field,
            f = self.__get_field(k)
            if f and f._coerce:
                v = f.serialize(v)

            # if someone assigned AttrList, unwrap it
            if isinstance(v, AttrList):
                v = v._l_

            if skip_empty:
                # don't serialize empty values
                # careful not to include numeric zeros
                if v in ([], {}, None):
                    continue

            out[k] = v
        return out

    def clean_fields(self, validate: bool = True) -> None:
        errors: Dict[str, List[ValidationException]] = {}
        for name, field, optional in self.__list_fields():
            data = self._d_.get(name, None)
            if data is None and optional:
                continue
            try:
                # save the cleaned value
                data = field.clean(data)
            except ValidationException as e:
                errors.setdefault(name, []).append(e)

            if name in self._d_ or data not in ([], {}, None):
                self._d_[name] = cast(Any, data)

        if validate and errors:
            raise ValidationException(errors)

    def clean(self) -> None:
        pass

    def full_clean(self) -> None:
        self.clean_fields(validate=False)
        self.clean()
        self.clean_fields(validate=True)


def merge(
    data: Union[Dict[str, Any], AttrDict[Any]],
    new_data: Union[Dict[str, Any], AttrDict[Any]],
    raise_on_conflict: bool = False,
) -> None:
    if not (
        isinstance(data, (AttrDict, collections.abc.Mapping))
        and isinstance(new_data, (AttrDict, collections.abc.Mapping))
    ):
        raise ValueError(
            f"You can only merge two dicts! Got {data!r} and {new_data!r} instead."
        )

    for key, value in new_data.items():
        if (
            key in data
            and isinstance(data[key], (AttrDict, collections.abc.Mapping))
            and isinstance(value, (AttrDict, collections.abc.Mapping))
        ):
            merge(data[key], value, raise_on_conflict)  # type: ignore[arg-type]
        elif key in data and data[key] != value and raise_on_conflict:
            raise ValueError(f"Incompatible data for key {key!r}, cannot be merged.")
        else:
            data[key] = value


def recursive_to_dict(data: Any) -> Any:
    """Recursively transform objects that potentially have .to_dict()
    into dictionary literals by traversing AttrList, AttrDict, list,
    tuple, and Mapping types.
    """
    if isinstance(data, AttrList):
        data = list(data._l_)
    elif hasattr(data, "to_dict"):
        data = data.to_dict()
    if isinstance(data, (list, tuple)):
        return type(data)(recursive_to_dict(inner) for inner in data)
    elif isinstance(data, dict):
        return {key: recursive_to_dict(val) for key, val in data.items()}
    return data