"""JSON Patch, as per RFC 6902."""

from __future__ import annotations

import copy
import json
from abc import ABC
from abc import abstractmethod
from io import IOBase
from typing import Any
from typing import Dict
from typing import Iterable
from typing import List
from typing import Mapping
from typing import MutableMapping
from typing import MutableSequence
from typing import TypeVar
from typing import Union

from jsonpath._data import load_data
from jsonpath.exceptions import JSONPatchError
from jsonpath.exceptions import JSONPatchTestFailure
from jsonpath.exceptions import JSONPointerError
from jsonpath.exceptions import JSONPointerIndexError
from jsonpath.exceptions import JSONPointerKeyError
from jsonpath.exceptions import JSONPointerTypeError
from jsonpath.pointer import UNDEFINED
from jsonpath.pointer import JSONPointer


class Op(ABC):
    """One of the JSON Patch operations."""

    name = "base"

    @abstractmethod
    def apply(
        self, data: Union[MutableSequence[object], MutableMapping[str, object]]
    ) -> Union[MutableSequence[object], MutableMapping[str, object]]:
        """Apply this patch operation to _data_."""

    @abstractmethod
    def asdict(self) -> Dict[str, object]:
        """Return a dictionary representation of this operation."""


class OpAdd(Op):
    """The JSON Patch _add_ operation."""

    __slots__ = ("path", "value")

    name = "add"

    def __init__(self, path: JSONPointer, value: object) -> None:
        self.path = path
        self.value = value

    def apply(
        self, data: Union[MutableSequence[object], MutableMapping[str, object]]
    ) -> Union[MutableSequence[object], MutableMapping[str, object]]:
        """Apply this patch operation to _data_."""
        parent, obj = self.path.resolve_parent(data)
        if parent is None:
            # Replace the root object.
            # The following op, if any, will raise a JSONPatchError if needed.
            return self.value  # type: ignore

        target = self.path.parts[-1]
        if isinstance(parent, MutableSequence):
            if obj is UNDEFINED:
                if target == "-":
                    parent.append(self.value)
                else:
                    index = self.path._index(target)  # noqa: SLF001
                    if index == len(parent):
                        parent.append(self.value)
                    else:
                        raise JSONPatchError("index out of range")
            else:
                parent.insert(int(target), self.value)
        elif isinstance(parent, MutableMapping):
            parent[str(target)] = self.value
        else:
            raise JSONPatchError(
                f"unexpected operation on {parent.__class__.__name__!r}"
            )
        return data

    def asdict(self) -> Dict[str, object]:
        """Return a dictionary representation of this operation."""
        return {"op": self.name, "path": str(self.path), "value": self.value}


class OpAddNe(OpAdd):
    """A non-standard _add if not exists_ operation.

    This is like _OpAdd_, but only adds object/dict keys/values if they key does
    not already exist.

    **New in version 1.2.0**
    """

    __slots__ = ("path", "value")

    name = "addne"

    def apply(
        self, data: Union[MutableSequence[object], MutableMapping[str, object]]
    ) -> Union[MutableSequence[object], MutableMapping[str, object]]:
        """Apply this patch operation to _data_."""
        parent, obj = self.path.resolve_parent(data)
        if parent is None:
            # Replace the root object.
            # The following op, if any, will raise a JSONPatchError if needed.
            return self.value  # type: ignore

        target = self.path.parts[-1]
        if isinstance(parent, MutableSequence):
            if obj is UNDEFINED:
                parent.append(self.value)
            else:
                parent.insert(int(target), self.value)
        elif isinstance(parent, MutableMapping) and target not in parent:
            parent[target] = self.value
        return data


class OpAddAp(OpAdd):
    """A non-standard add operation that appends to arrays/lists .

    This is like _OpAdd_, but assumes an index of "-" if the path can not
    be resolved rather than raising a JSONPatchError.

    **New in version 1.2.0**
    """

    __slots__ = ("path", "value")

    name = "addap"

    def apply(
        self, data: Union[MutableSequence[object], MutableMapping[str, object]]
    ) -> Union[MutableSequence[object], MutableMapping[str, object]]:
        """Apply this patch operation to _data_."""
        parent, obj = self.path.resolve_parent(data)
        if parent is None:
            # Replace the root object.
            # The following op, if any, will raise a JSONPatchError if needed.
            return self.value  # type: ignore

        target = self.path.parts[-1]
        if isinstance(parent, MutableSequence):
            if obj is UNDEFINED:
                parent.append(self.value)
            else:
                parent.insert(int(target), self.value)
        elif isinstance(parent, MutableMapping):
            parent[target] = self.value
        else:
            raise JSONPatchError(
                f"unexpected operation on {parent.__class__.__name__!r}"
            )
        return data


class OpRemove(Op):
    """The JSON Patch _remove_ operation."""

    __slots__ = ("path",)

    name = "remove"

    def __init__(self, path: JSONPointer) -> None:
        self.path = path

    def apply(
        self, data: Union[MutableSequence[object], MutableMapping[str, object]]
    ) -> Union[MutableSequence[object], MutableMapping[str, object]]:
        """Apply this patch operation to _data_."""
        parent, obj = self.path.resolve_parent(data)
        if parent is None:
            raise JSONPatchError("can't remove root")

        if isinstance(parent, MutableSequence):
            if obj is UNDEFINED:
                raise JSONPatchError("can't remove nonexistent item")
            del parent[int(self.path.parts[-1])]
        elif isinstance(parent, MutableMapping):
            if obj is UNDEFINED:
                raise JSONPatchError("can't remove nonexistent property")
            del parent[str(self.path.parts[-1])]
        else:
            raise JSONPatchError(
                f"unexpected operation on {parent.__class__.__name__!r}"
            )
        return data

    def asdict(self) -> Dict[str, object]:
        """Return a dictionary representation of this operation."""
        return {"op": self.name, "path": str(self.path)}


class OpReplace(Op):
    """The JSON Patch _replace_ operation."""

    __slots__ = ("path", "value")

    name = "replace"

    def __init__(self, path: JSONPointer, value: object) -> None:
        self.path = path
        self.value = value

    def apply(
        self, data: Union[MutableSequence[object], MutableMapping[str, object]]
    ) -> Union[MutableSequence[object], MutableMapping[str, object]]:
        """Apply this patch operation to _data_."""
        parent, obj = self.path.resolve_parent(data)
        if parent is None:
            return self.value  # type: ignore

        if isinstance(parent, MutableSequence):
            if obj is UNDEFINED:
                raise JSONPatchError("can't replace nonexistent item")
            parent[int(self.path.parts[-1])] = self.value
        elif isinstance(parent, MutableMapping):
            if obj is UNDEFINED:
                raise JSONPatchError("can't replace nonexistent property")
            parent[str(self.path.parts[-1])] = self.value
        else:
            raise JSONPatchError(
                f"unexpected operation on {parent.__class__.__name__!r}"
            )
        return data

    def asdict(self) -> Dict[str, object]:
        """Return a dictionary representation of this operation."""
        return {"op": self.name, "path": str(self.path), "value": self.value}


class OpMove(Op):
    """The JSON Patch _move_ operation."""

    __slots__ = ("source", "dest")

    name = "move"

    def __init__(self, from_: JSONPointer, path: JSONPointer) -> None:
        self.source = from_
        self.dest = path

    def apply(
        self, data: Union[MutableSequence[object], MutableMapping[str, object]]
    ) -> Union[MutableSequence[object], MutableMapping[str, object]]:
        """Apply this patch operation to _data_."""
        if self.dest.is_relative_to(self.source):
            raise JSONPatchError("can't move object to one of its own children")

        source_parent, source_obj = self.source.resolve_parent(data)

        if source_obj is UNDEFINED:
            raise JSONPatchError("source object does not exist")

        if isinstance(source_parent, MutableSequence):
            del source_parent[int(self.source.parts[-1])]
        if isinstance(source_parent, MutableMapping):
            del source_parent[str(self.source.parts[-1])]

        dest_parent, _ = self.dest.resolve_parent(data)

        if dest_parent is None:
            # Move source to root
            return source_obj  # type: ignore

        if isinstance(dest_parent, MutableSequence):
            dest_parent.insert(int(self.dest.parts[-1]), source_obj)
        elif isinstance(dest_parent, MutableMapping):
            dest_parent[str(self.dest.parts[-1])] = source_obj
        else:
            raise JSONPatchError(
                f"unexpected operation on {dest_parent.__class__.__name__!r}"
            )

        return data

    def asdict(self) -> Dict[str, object]:
        """Return a dictionary representation of this operation."""
        return {"op": self.name, "from": str(self.source), "path": str(self.dest)}


class OpCopy(Op):
    """The JSON Patch _copy_ operation."""

    __slots__ = ("source", "dest")

    name = "copy"

    def __init__(self, from_: JSONPointer, path: JSONPointer) -> None:
        self.source = from_
        self.dest = path

    def apply(
        self, data: Union[MutableSequence[object], MutableMapping[str, object]]
    ) -> Union[MutableSequence[object], MutableMapping[str, object]]:
        """Apply this patch operation to _data_."""
        source_parent, source_obj = self.source.resolve_parent(data)

        if source_obj is UNDEFINED:
            raise JSONPatchError("source object does not exist")

        dest_parent, dest_obj = self.dest.resolve_parent(data)

        if dest_parent is None:
            # Copy source to root
            return copy.deepcopy(source_obj)  # type: ignore

        if isinstance(dest_parent, MutableSequence):
            dest_parent.insert(int(self.dest.parts[-1]), copy.deepcopy(source_obj))
        elif isinstance(dest_parent, MutableMapping):
            dest_parent[str(self.dest.parts[-1])] = copy.deepcopy(source_obj)
        else:
            raise JSONPatchError(
                f"unexpected operation on {dest_parent.__class__.__name__!r}"
            )

        return data

    def asdict(self) -> Dict[str, object]:
        """Return a dictionary representation of this operation."""
        return {"op": self.name, "from": str(self.source), "path": str(self.dest)}


class OpTest(Op):
    """The JSON Patch _test_ operation."""

    __slots__ = ("path", "value")

    name = "test"

    def __init__(self, path: JSONPointer, value: object) -> None:
        self.path = path
        self.value = value

    def apply(
        self, data: Union[MutableSequence[object], MutableMapping[str, object]]
    ) -> Union[MutableSequence[object], MutableMapping[str, object]]:
        """Apply this patch operation to _data_."""
        _, obj = self.path.resolve_parent(data)
        if not obj == self.value:
            raise JSONPatchTestFailure
        return data

    def asdict(self) -> Dict[str, object]:
        """Return a dictionary representation of this operation."""
        return {"op": self.name, "path": str(self.path), "value": self.value}


Self = TypeVar("Self", bound="JSONPatch")


class JSONPatch:
    """Modify JSON-like data with JSON Patch.

    RFC 6902 defines operations to manipulate a JSON document. `JSONPatch`
    supports parsing and applying standard JSON Patch formatted operations,
    and provides a Python builder API following the same semantics as RFC 6902.

    Arguments:
        ops: A JSON Patch formatted document or equivalent Python objects.
        unicode_escape: If `True`, UTF-16 escape sequences will be decoded
            before parsing JSON pointers.
        uri_decode: If `True`, JSON pointers will be unescaped using _urllib_
            before being parsed.

    Raises:
        JSONPatchError: If _ops_ is given and any of the provided operations
            is malformed.
    """

    def __init__(
        self,
        ops: Union[str, IOBase, Iterable[Mapping[str, object]], None] = None,
        *,
        unicode_escape: bool = True,
        uri_decode: bool = False,
    ) -> None:
        self.ops: List[Op] = []
        self.unicode_escape = unicode_escape
        self.uri_decode = uri_decode
        if ops:
            self._load(ops)

    def _load(self, patch: Union[str, IOBase, Iterable[Mapping[str, object]]]) -> None:
        if isinstance(patch, IOBase):
            _patch = json.loads(patch.read())
        elif isinstance(patch, str):
            _patch = json.loads(patch)
        else:
            _patch = patch

        try:
            self._build(_patch)
        except TypeError as err:
            raise JSONPatchError(
                "expected a sequence of patch operations, "
                f"found {_patch.__class__.__name__!r}"
            ) from err

    def _build(self, patch: Iterable[Mapping[str, object]]) -> None:
        for i, operation in enumerate(patch):
            try:
                op = operation["op"]
            except KeyError as err:
                raise JSONPatchError(f"missing 'op' member at op {i}") from err

            if op == "add":
                self.add(
                    path=self._op_pointer(operation, "path", "add", i),
                    value=self._op_value(operation, "value", "add", i),
                )
            elif op == "addne":
                self.addne(
                    path=self._op_pointer(operation, "path", "addne", i),
                    value=self._op_value(operation, "value", "addne", i),
                )
            elif op == "addap":
                self.addap(
                    path=self._op_pointer(operation, "path", "addap", i),
                    value=self._op_value(operation, "value", "addap", i),
                )
            elif op == "remove":
                self.remove(path=self._op_pointer(operation, "path", "add", i))
            elif op == "replace":
                self.replace(
                    path=self._op_pointer(operation, "path", "replace", i),
                    value=self._op_value(operation, "value", "replace", i),
                )
            elif op == "move":
                self.move(
                    from_=self._op_pointer(operation, "from", "move", i),
                    path=self._op_pointer(operation, "path", "move", i),
                )
            elif op == "copy":
                self.copy(
                    from_=self._op_pointer(operation, "from", "copy", i),
                    path=self._op_pointer(operation, "path", "copy", i),
                )
            elif op == "test":
                self.test(
                    path=self._op_pointer(operation, "path", "test", i),
                    value=self._op_value(operation, "value", "test", i),
                )
            else:
                raise JSONPatchError(
                    "expected 'op' to be one of 'add', 'remove', 'replace', "
                    f"'move', 'copy' or 'test' ({op}:{i})"
                )

    def _op_pointer(
        self, operation: Mapping[str, object], key: str, op: str, i: int
    ) -> JSONPointer:
        try:
            pointer = operation[key]
        except KeyError as err:
            raise JSONPatchError(f"missing property {key!r} ({op}:{i})") from err

        if not isinstance(pointer, str):
            raise JSONPatchError(
                f"expected a JSON Pointer string for {key!r}, "
                f"found {pointer.__class__.__name__!r} "
                f"({op}:{i})"
            )

        try:
            return JSONPointer(
                pointer, unicode_escape=self.unicode_escape, uri_decode=self.uri_decode
            )
        except JSONPointerError as err:
            raise JSONPatchError(f"{err} ({op}:{i})") from err

    def _op_value(
        self, operation: Mapping[str, object], key: str, op: str, i: int
    ) -> object:
        try:
            return operation[key]
        except KeyError as err:
            raise JSONPatchError(f"missing property {key!r} ({op}:{i})") from err

    def _ensure_pointer(self, path: Union[str, JSONPointer]) -> JSONPointer:
        if isinstance(path, str):
            return JSONPointer(
                path,
                unicode_escape=self.unicode_escape,
                uri_decode=self.uri_decode,
            )
        assert isinstance(path, JSONPointer)
        return path

    def add(self: Self, path: Union[str, JSONPointer], value: object) -> Self:
        """Append an _add_ operation to this patch.

        Arguments:
            path: A string representation of a JSON Pointer, or one that has
                already been parsed.
            value: The object to add.

        Returns:
            This `JSONPatch` instance, so we can build a JSON Patch by chaining
                calls to JSON Patch operation methods.
        """
        pointer = self._ensure_pointer(path)
        self.ops.append(OpAdd(path=pointer, value=value))
        return self

    def addne(self: Self, path: Union[str, JSONPointer], value: object) -> Self:
        """Append an _addne_ operation to this patch.

        Arguments:
            path: A string representation of a JSON Pointer, or one that has
                already been parsed.
            value: The object to add.

        Returns:
            This `JSONPatch` instance, so we can build a JSON Patch by chaining
                calls to JSON Patch operation methods.
        """
        pointer = self._ensure_pointer(path)
        self.ops.append(OpAddNe(path=pointer, value=value))
        return self

    def addap(self: Self, path: Union[str, JSONPointer], value: object) -> Self:
        """Append an _addap_ operation to this patch.

        Arguments:
            path: A string representation of a JSON Pointer, or one that has
                already been parsed.
            value: The object to add.

        Returns:
            This `JSONPatch` instance, so we can build a JSON Patch by chaining
                calls to JSON Patch operation methods.
        """
        pointer = self._ensure_pointer(path)
        self.ops.append(OpAddAp(path=pointer, value=value))
        return self

    def remove(self: Self, path: Union[str, JSONPointer]) -> Self:
        """Append a _remove_ operation to this patch.

        Arguments:
            path: A string representation of a JSON Pointer, or one that has
                already been parsed.

        Returns:
            This `JSONPatch` instance, so we can build a JSON Patch by chaining
                calls to JSON Patch operation methods.
        """
        pointer = self._ensure_pointer(path)
        self.ops.append(OpRemove(path=pointer))
        return self

    def replace(self: Self, path: Union[str, JSONPointer], value: object) -> Self:
        """Append a _replace_ operation to this patch.

        Arguments:
            path: A string representation of a JSON Pointer, or one that has
                already been parsed.
            value: The object to add.

        Returns:
            This `JSONPatch` instance, so we can build a JSON Patch by chaining
                calls to JSON Patch operation methods.
        """
        pointer = self._ensure_pointer(path)
        self.ops.append(OpReplace(path=pointer, value=value))
        return self

    def move(
        self: Self, from_: Union[str, JSONPointer], path: Union[str, JSONPointer]
    ) -> Self:
        """Append a _move_ operation to this patch.

        Arguments:
            from_: A string representation of a JSON Pointer, or one that has
                already been parsed.
            path: A string representation of a JSON Pointer, or one that has
                already been parsed.

        Returns:
            This `JSONPatch` instance, so we can build a JSON Patch by chaining
                calls to JSON Patch operation methods.
        """
        source_pointer = self._ensure_pointer(from_)
        dest_pointer = self._ensure_pointer(path)
        self.ops.append(OpMove(from_=source_pointer, path=dest_pointer))
        return self

    def copy(
        self: Self, from_: Union[str, JSONPointer], path: Union[str, JSONPointer]
    ) -> Self:
        """Append a _copy_ operation to this patch.

        Arguments:
            from_: A string representation of a JSON Pointer, or one that has
                already been parsed.
            path: A string representation of a JSON Pointer, or one that has
                already been parsed.

        Returns:
            This `JSONPatch` instance, so we can build a JSON Patch by chaining
                calls to JSON Patch operation methods.
        """
        source_pointer = self._ensure_pointer(from_)
        dest_pointer = self._ensure_pointer(path)
        self.ops.append(OpCopy(from_=source_pointer, path=dest_pointer))
        return self

    def test(self: Self, path: Union[str, JSONPointer], value: object) -> Self:
        """Append a test operation to this patch.

        Arguments:
            path: A string representation of a JSON Pointer, or one that has
                already been parsed.
            value: The object to test.

        Returns:
            This `JSONPatch` instance, so we can build a JSON Patch by chaining
                calls to JSON Patch operation methods.
        """
        pointer = self._ensure_pointer(path)
        self.ops.append(OpTest(path=pointer, value=value))
        return self

    def apply(
        self,
        data: Union[str, IOBase, MutableSequence[Any], MutableMapping[str, Any]],
    ) -> object:
        """Apply all operations from this patch to _data_.

        If _data_ is a string or file-like object, it will be loaded with
        _json.loads_. Otherwise _data_ should be a JSON-like data structure and
        will be modified in place.

        When modifying _data_ in place, we return modified data too. This is
        to allow for replacing _data's_ root element, which is allowed by some
        patch operations.

        Arguments:
            data: The target JSON "document" or equivalent Python objects.

        Returns:
            Modified input data.

        Raises:
            JSONPatchError: When a patch operation fails.
            JSONPatchTestFailure: When a _test_ operation does not pass.
                `JSONPatchTestFailure` is a subclass of `JSONPatchError`.
        """
        _data = load_data(data)

        for i, op in enumerate(self.ops):
            try:
                _data = op.apply(_data)
            except JSONPatchTestFailure as err:
                raise JSONPatchTestFailure(f"test failed ({op.name}:{i})") from err
            except JSONPointerKeyError as err:
                raise JSONPatchError(f"{err} ({op.name}:{i})") from err
            except JSONPointerIndexError as err:
                raise JSONPatchError(f"{err} ({op.name}:{i})") from err
            except JSONPointerTypeError as err:
                raise JSONPatchError(f"{err} ({op.name}:{i})") from err
            except (JSONPointerError, JSONPatchError) as err:
                raise JSONPatchError(f"{err} ({op.name}:{i})") from err

        return _data

    def asdicts(self) -> List[Dict[str, object]]:
        """Return a list of this patch's operations as dictionaries."""
        return [op.asdict() for op in self.ops]


def apply(
    patch: Union[str, IOBase, Iterable[Mapping[str, object]], None],
    data: Union[str, IOBase, MutableSequence[Any], MutableMapping[str, Any]],
    *,
    unicode_escape: bool = True,
    uri_decode: bool = False,
) -> object:
    """Apply the JSON Patch _patch_ to _data_.

    If _data_ is a string or file-like object, it will be loaded with
    _json.loads_. Otherwise _data_ should be a JSON-like data structure and
    will be **modified in-place**.

    When modifying _data_ in-place, we return modified data too. This is
    to allow for replacing _data's_ root element, which is allowed by some
    patch operations.

    Arguments:
        patch: A JSON Patch formatted document or equivalent Python objects.
        data: The target JSON "document" or equivalent Python objects.
        unicode_escape: If `True`, UTF-16 escape sequences will be decoded
            before parsing JSON pointers.
        uri_decode: If `True`, JSON pointers will be unescaped using _urllib_
            before being parsed.

    Returns:
        Modified input data.

    Raises:
        JSONPatchError: When a patch operation fails.
        JSONPatchTestFailure: When a _test_ operation does not pass.
            `JSONPatchTestFailure` is a subclass of `JSONPatchError`.

    """
    return JSONPatch(
        patch,
        unicode_escape=unicode_escape,
        uri_decode=uri_decode,
    ).apply(data)