import json
import warnings
from dataclasses import dataclass, field
from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Type, Union

from xsdata.exceptions import ConverterWarning, ParserError
from xsdata.formats.bindings import AbstractParser, T
from xsdata.formats.converter import converter
from xsdata.formats.dataclass.context import XmlContext
from xsdata.formats.dataclass.models.elements import XmlMeta, XmlVar
from xsdata.formats.dataclass.parsers.config import ParserConfig
from xsdata.formats.dataclass.parsers.utils import ParserUtils
from xsdata.formats.dataclass.typing import get_args, get_origin
from xsdata.utils import collections
from xsdata.utils.constants import EMPTY_MAP


@dataclass
class JsonParser(AbstractParser):
    """
    Json parser for dataclasses.

    :param config: Parser configuration
    :param context: Model context provider
    :param load_factory: Replace the default json.load call with another
        implementation
    """

    config: ParserConfig = field(default_factory=ParserConfig)
    context: XmlContext = field(default_factory=XmlContext)
    load_factory: Callable = field(default=json.load)

    def parse(self, source: Any, clazz: Optional[Type[T]] = None) -> T:
        """Parse the input stream or filename and return the resulting object
        tree."""

        data = self.load_json(source)
        tp = self.verify_type(clazz, data)

        with warnings.catch_warnings():
            if self.config.fail_on_converter_warnings:
                warnings.filterwarnings("error", category=ConverterWarning)

            try:
                if not isinstance(data, list):
                    return self.bind_dataclass(data, tp)

                return [self.bind_dataclass(obj, tp) for obj in data]  # type: ignore
            except ConverterWarning as e:
                raise ParserError(e)

    def load_json(self, source: Any) -> Union[Dict, List]:
        if not hasattr(source, "read"):
            with open(source, "rb") as fp:
                return self.load_factory(fp)

        return self.load_factory(source)

    def verify_type(self, clazz: Optional[Type[T]], data: Union[Dict, List]) -> Type[T]:
        if clazz is None:
            return self.detect_type(data)

        try:
            origin = get_origin(clazz)
            list_type = False
            if origin is list:
                list_type = True
                args = get_args(clazz)

                if len(args) != 1 or not self.context.class_type.is_model(args[0]):
                    raise TypeError()

                clazz = args[0]
            elif origin is not None:
                raise TypeError()
        except TypeError:
            raise ParserError(f"Invalid clazz argument: {clazz}")

        if list_type != isinstance(data, list):
            if list_type:
                raise ParserError("Document is object, expected array")
            raise ParserError("Document is array, expected object")

        return clazz  # type: ignore

    def detect_type(self, data: Union[Dict, List]) -> Type[T]:
        if not data:
            raise ParserError("Document is empty, can not detect type")

        keys = data[0].keys() if isinstance(data, list) else data.keys()
        clazz: Optional[Type[T]] = self.context.find_type_by_fields(set(keys))

        if clazz:
            return clazz

        raise ParserError(f"Unable to locate model with properties({list(keys)})")

    def bind_dataclass(self, data: Dict, clazz: Type[T]) -> T:
        """Recursively build the given model from the input dict data."""
        if set(data.keys()) == self.context.class_type.derived_keys:
            return self.bind_derived_dataclass(data, clazz)

        meta = self.context.build(clazz)
        xml_vars = meta.get_all_vars()

        params = {}
        for key, value in data.items():
            is_array = collections.is_array(value)
            var = self.find_var(xml_vars, key, is_array)

            if var is None and self.config.fail_on_unknown_properties:
                raise ParserError(f"Unknown property {clazz.__qualname__}.{key}")

            if var and var.init:
                params[var.name] = self.bind_value(meta, var, value)

        try:
            return self.config.class_factory(clazz, params)
        except TypeError as e:
            raise ParserError(e)

    def bind_derived_dataclass(self, data: Dict, clazz: Type[T]) -> Any:
        qname = data["qname"]
        xsi_type = data["type"]
        params = data["value"]

        generic = self.context.class_type.derived_element

        if clazz is generic:
            real_clazz: Optional[Type[T]] = None
            if xsi_type:
                real_clazz = self.context.find_type(xsi_type)

            if real_clazz is None:
                raise ParserError(
                    f"Unable to locate derived model "
                    f"with properties({list(params.keys())})"
                )

            value = self.bind_dataclass(params, real_clazz)
        else:
            value = self.bind_dataclass(params, clazz)

        return generic(qname=qname, type=xsi_type, value=value)

    def bind_best_dataclass(self, data: Dict, classes: Iterable[Type[T]]) -> T:
        """Attempt to bind the given data to one possible models, if more than
        one is successful return the object with the highest score."""
        obj = None
        keys = set(data.keys())
        max_score = -1.0
        for clazz in classes:
            if not self.context.class_type.is_model(clazz):
                continue

            if self.context.local_names_match(keys, clazz):
                candidate = self.bind_optional_dataclass(data, clazz)
                score = self.context.class_type.score_object(candidate)
                if score > max_score:
                    max_score = score
                    obj = candidate

        if obj:
            return obj

        raise ParserError(
            f"Failed to bind object with properties({list(data.keys())}) "
            f"to any of the {[cls.__qualname__ for cls in classes]}"
        )

    def bind_optional_dataclass(self, data: Dict, clazz: Type[T]) -> Optional[T]:
        """Recursively build the given model from the input dict data but fail
        on any converter warnings."""
        try:
            with warnings.catch_warnings():
                warnings.filterwarnings("error", category=ConverterWarning)
                return self.bind_dataclass(data, clazz)
        except Exception:
            return None

    def bind_value(
        self, meta: XmlMeta, var: XmlVar, value: Any, recursive: bool = False
    ) -> Any:
        """Main entry point for binding values."""

        # xs:anyAttributes get it out of the way, it's the mapping exception!
        if var.is_attributes:
            return dict(value)

        # Repeating element, recursively bind the values
        if not recursive and var.list_element and isinstance(value, list):
            assert var.factory is not None
            return var.factory(self.bind_value(meta, var, val, True) for val in value)

        # If not dict this is an text or tokens value.
        if not isinstance(value, dict):
            return self.bind_text(meta, var, value)

        keys = value.keys()
        if keys == self.context.class_type.any_keys:
            # Bind data to AnyElement dataclass
            return self.bind_dataclass(value, self.context.class_type.any_element)

        if keys == self.context.class_type.derived_keys:
            # Bind data to AnyElement dataclass
            return self.bind_derived_value(meta, var, value)

        # Bind data to a user defined dataclass
        return self.bind_complex_type(meta, var, value)

    def bind_text(self, meta: XmlMeta, var: XmlVar, value: Any) -> Any:
        """Bind text/tokens value entrypoint."""
        if var.is_elements:
            # Compound field we need to match the value to one of the choice elements
            check_subclass = self.context.class_type.is_model(value)
            choice = var.find_value_choice(value, check_subclass)
            if choice:
                return self.bind_text(meta, choice, value)

            if value is None:
                return value

            raise ParserError(
                f"Failed to bind '{value}' "
                f"to {meta.clazz.__qualname__}.{var.name} field"
            )

        if var.any_type or var.is_wildcard:
            # field can support any object return the value as it is
            return value

        value = converter.serialize(value)

        # Convert value according to the field types
        return ParserUtils.parse_value(
            value=value,
            types=var.types,
            default=var.default,
            ns_map=EMPTY_MAP,
            tokens_factory=var.tokens_factory,
            format=var.format,
        )

    def bind_complex_type(self, meta: XmlMeta, var: XmlVar, data: Dict) -> Any:
        """Bind data to a user defined dataclass."""
        if var.is_clazz_union:
            # Union of dataclasses
            return self.bind_best_dataclass(data, var.types)
        if var.elements:
            # Compound field with multiple choices
            return self.bind_best_dataclass(data, var.element_types)
        if var.any_type or var.is_wildcard:
            # xs:anyType element, check all meta classes
            return self.bind_best_dataclass(data, meta.element_types)

        assert var.clazz is not None

        subclasses = set(self.context.get_subclasses(var.clazz))
        if subclasses:
            # field annotation is an abstract/base type
            subclasses.add(var.clazz)
            return self.bind_best_dataclass(data, subclasses)

        return self.bind_dataclass(data, var.clazz)

    def bind_derived_value(self, meta: XmlMeta, var: XmlVar, data: Dict) -> Any:
        """Bind derived element entry point."""
        qname = data["qname"]
        xsi_type = data["type"]
        params = data["value"]

        if var.elements:
            choice = var.find_choice(qname)
            if choice is None:
                raise ParserError(
                    f"Unable to locate compound element"
                    f" {meta.clazz.__qualname__}.{var.name}[{qname}]"
                )
            return self.bind_derived_value(meta, choice, data)

        if not isinstance(params, dict):
            value = self.bind_text(meta, var, params)
        elif xsi_type:
            clazz: Optional[Type] = self.context.find_type(xsi_type)

            if clazz is None:
                raise ParserError(f"Unable to locate xsi:type `{xsi_type}`")

            value = self.bind_dataclass(params, clazz)
        elif var.clazz:
            value = self.bind_complex_type(meta, var, params)
        else:
            value = self.bind_best_dataclass(params, meta.element_types)

        generic = self.context.class_type.derived_element
        return generic(qname=qname, value=value, type=xsi_type)

    @classmethod
    def find_var(
        cls, xml_vars: Sequence[XmlVar], local_name: str, is_list: bool = False
    ) -> Optional[XmlVar]:
        for var in xml_vars:
            if var.local_name == local_name:
                var_is_list = var.list_element or var.tokens
                if is_list == var_is_list or var.clazz is None:
                    return var

        return None


@dataclass
class DictConverter(JsonParser):
    def convert(self, data: Dict, clazz: Type[T]) -> T:
        return self.bind_dataclass(data, clazz)