# Source code for nio.events.account_data

    # -*- coding: utf-8 -*-

    # Copyright © 2018-2019 Damir Jelić <poljar@termina.org.uk>
    #
    # Permission to use, copy, modify, and/or distribute this software for
    # any purpose with or without fee is hereby granted, provided that the
    # above copyright notice and this permission notice appear in all copies.
    #
    # THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL WARRANTIES
    # WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED WARRANTIES OF
    # MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE AUTHOR BE LIABLE FOR ANY
    # SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL DAMAGES OR ANY DAMAGES WHATSOEVER
    # RESULTING FROM LOSS OF USE, DATA OR PROFITS, WHETHER IN AN ACTION OF
    # CONTRACT, NEGLIGENCE OR OTHER TORTIOUS ACTION, ARISING OUT OF OR IN
    # CONNECTION WITH THE USE OR PERFORMANCE OF THIS SOFTWARE.

    """nio Account data events.

    Clients can store custom config data for their account on their homeserver.

    This account data will be synced between different devices and can persist
    across installations on a particular device.

    """

    from __future__ import unicode_literals

    import re

    from fnmatch import fnmatchcase
    from typing import Any, Dict, List, Optional, Union, TYPE_CHECKING

    from dataclasses import dataclass, field

    from ..api import PushRuleKind
    from ..schemas import Schemas
    from .misc import verify, verify_or_none
    from .room_events import Event

    if TYPE_CHECKING:
        from ..rooms import MatrixRoom

    @dataclass
    class AccountDataEvent:
        """Abstract class for account data events."""

        @classmethod
        @verify(Schemas.account_data)
        def parse_event(
            cls,
            event_dict,  # type: Dict[Any, Any]
        ):

            if event_dict["type"] == "m.fully_read":
                return FullyReadEvent.from_dict(event_dict)
            elif event_dict["type"] == "m.tag":
                return TagEvent.from_dict(event_dict)
            elif event_dict["type"] == "m.push_rules":
                return PushRulesEvent.from_dict(event_dict)

            return UnknownAccountDataEvent.from_dict(event_dict)

    @dataclass
    class FullyReadEvent(AccountDataEvent):
        """Read marker location event.

        The current location of the user's read marker in a room.
        This event appears in the user's room account data for the room the marker
        is applicable for.

        Attributes:
            event_id (str): The event id the user's read marker is located
                at in the room.

        """

        event_id: str = field()

        @classmethod
        @verify(Schemas.fully_read)
        def from_dict(cls, event_dict):
            """Construct a FullyReadEvent from a dictionary."""
            content = event_dict.pop("content")
            return cls(
                content["event_id"],
            )

    @dataclass
    class TagEvent(AccountDataEvent):
        """Event representing the tags of a room.

           Room tags may include:

            - m.favourite for favourite rooms
            - m.lowpriority for low priority room

           A tag may have an order between 0 and 1, indicating the
           room's position towards other rooms with the same tag.

        Attributes:
            tags (Dict[str, Optional[Dict[str, float]]]): The tags of the room
            and their contents.
        """

        tags: Dict[str, Optional[Dict[str, float]]] = field()

        @classmethod
        @verify(Schemas.tags)
        def from_dict(cls, event_dict):
            """Construct a TagEvent from a dictionary."""
            content = event_dict.pop("content")
            return cls(
                content["tags"]
            )

    @dataclass
    class PushCondition:
        """A condition for a push rule to match an event."""

        @classmethod
        def from_dict(cls, condition: Dict[str, Any]) -> "PushCondition":
            cnd = condition

            if cnd["kind"] == "event_match" and "key" in cnd and "pattern" in cnd:
                return PushEventMatch(cnd["key"], cnd["pattern"])

            if cnd["kind"] == "contains_display_name":
                return PushContainsDisplayName()

            if cnd["kind"] == "room_member_count":
                return PushRoomMemberCount.from_dict(cnd)

            if cnd["kind"] == "sender_notification_permission" and "key" in cnd:
                return PushSenderNotificationPermission(cnd["key"])

            return PushUnknownCondition(cnd)

        @property
        def as_value(self) -> Dict[str, Any]:
            raise NotImplementedError()

        def matches(
            self, event: Event, room: "MatrixRoom", display_name: str,
        ) -> bool:
            """Return whether this condition holds true for a room event.

            Args:
                event (Event): The room event to check the condition for.
                room (MatrixRoom): The room that this event is part of.
                display_name (str): The display name of our own user in the room.
            """
            return False

    @dataclass
    class PushEventMatch(PushCondition):
        """Require a field of the event to match a glob-style pattern.

        Attributes:
            key (str): The dot-separated field of the event to match,
                e.g. ``"type"`` or ``"content.body"``.

            pattern (str): Glob-style pattern to match the field's value against.
                Patterns with no special glob characters should be treated as
                starting and ending with an asterisk.
        """

        key: str = field()
        pattern: str = field()

        @property
        def as_value(self) -> Dict[str, Any]:
            return {
                "kind": "event_match", "key": self.key, "pattern": self.pattern,
            }

        def matches(
            self, event: Event, room: "MatrixRoom", display_name: str,
        ) -> bool:

            if self.key == "room_id":
                return fnmatchcase(room.room_id, self.pattern)

            value = event.flattened().get(self.key)

            if not isinstance(value, str):
                return False

            if self.key == "content.body":
                pattern = f"*[!a-z0-9]{self.pattern.lower()}[!a-z0-9]*"
                return fnmatchcase(f" {value.lower()} ", pattern)

            return fnmatchcase(value.lower(), self.pattern.lower())

    @dataclass
    class PushContainsDisplayName(PushCondition):
        """Require a message's ``content.body`` to contain our display name.

        This rule can only match unencrypted messages.
        """

        @property
        def as_value(self) -> Dict[str, Any]:
            return {"kind": "contains_display_name"}

        def matches(
            self, event: Event, room: "MatrixRoom", display_name: str,
        ) -> bool:
            body = event.source.get("content", {}).get("body")

            if not isinstance(body, str):
                return False

            pattern = rf"(^|\W){re.escape(display_name)}(\W|$)"
            return bool(re.match(pattern, body, re.IGNORECASE))

    @dataclass
    class PushRoomMemberCount(PushCondition):
        """Require a certain member count for the room the event is posted in.

        Attributes:
            count (int): A number of members
            operator (str): Whether the room's member count should be
                equal (``"=="``) to ``count``, inferior (``"<"``),
                superior (``">"``), inferior or equal (``"<="``),
                or superior or equal (``">="``).
        """

        count: int = field()
        operator: str = "=="

        @classmethod
        def from_dict(cls, condition: Dict[str, Any]) -> "PushRoomMemberCount":
            op, num = re.findall(r"(==|<|>|<=|>=)?([0-9.-]+)", condition["is"])[0]
            return cls(int(num), op or "==")

        @property
        def as_value(self) -> Dict[str, Any]:
            operator = "" if self.operator == "==" else self.operator
            return {"kind": "room_member_count", "is": f"{operator}{self.count}"}

        def matches(
            self, event: Event, room: "MatrixRoom", display_name: str,
        ) -> bool:
            if self.operator == "==":
                return room.joined_count == self.count
            elif self.operator == "<":
                return room.joined_count < self.count
            elif self.operator == ">":
                return room.joined_count > self.count
            elif self.operator == "<=":
                return room.joined_count <= self.count
            else:
                return room.joined_count >= self.count

    @dataclass
    class PushSenderNotificationPermission(PushCondition):
        """Require the event's sender to have a high enough power level.

        Attributes:
            key (str): Which key from the ``notifications`` dict in
            power levels event
            (https://matrix.org/docs/spec/client_server/latest#m-room-power-levels)
            should be refered to as the required level for the event's sender,
            e.g. ``room``.
        """

        key: str = field()

        @property
        def as_value(self) -> Dict[str, Any]:
            return {
                "kind": "sender_notification_permission", "key": self.key,
            }

        def matches(
            self, event: Event, room: "MatrixRoom", display_name: str,
        ) -> bool:
            return room.power_levels.can_user_notify(event.sender, self.key)

    @dataclass
    class PushUnknownCondition(PushCondition):
        """An unknown kind of push rule condition.

        Attributes:
            condition (Dict[str, Any]): The condition as a dict from the
                source event.
        """
        condition: Dict[str, Any] = field()

        @property
        def as_value(self) -> Dict[str, Any]:
            return self.condition

    @dataclass
    class PushAction:
        """An action to apply for a push rule when matching."""

        @classmethod
        def from_dict(cls, action: Union[str, Dict[str, Any]]) -> "PushAction":
            # isinstance() to make mypy happy

            if isinstance(action, str) and action == "notify":
                return PushNotify()

            if isinstance(action, str) and action == "dont_notify":
                return PushDontNotify()

            if isinstance(action, str) and action == "coalesce":
                return PushCoalesce()

            if isinstance(action, dict) and "set_tweak" in action:
                value = action.get("value")

                if action["set_tweak"] == "sound" and value is None:
                    value = "default"

                if action["set_tweak"] == "highlight" and value is None:
                    value = True

                return PushSetTweak(action["set_tweak"], value)

            return PushUnknownAction(action)

        @property
        def as_value(self) -> Union[str, Dict[str, Any]]:
            raise NotImplementedError()

    @dataclass
    class PushNotify(PushAction):
        """Cause the matching event to generate a notification."""

        @property
        def as_value(self) -> str:
            return "notify"

    @dataclass
    class PushDontNotify(PushAction):
        """Prevents the matching event from generating a notification."""

        @property
        def as_value(self) -> str:
            return "dont_notify"

    @dataclass
    class PushCoalesce(PushAction):
        """Causes multiple matching events to be joined into a single notification.

        The behavior is homeserver-dependent. Homeservers not supporting this
        action should treat it as a ``PushNotify`` action.
        """

        @property
        def as_value(self) -> str:
            return "coalesce"

    @dataclass
    class PushSetTweak(PushAction):
        """Set a particular tweak for the notification.

        These tweaks are defined by the Matrix specification:

        - ``sound``: The sound to be played when the notification arrives,
          e.g. a file path.
          A ``value`` of ``"default"`` means to play the client's default sound.
          A device may choose to alert the user by some other means if appropriate,
          e.g. vibration.

        - ``highlight``: Whether this message should be highlighted in the UI.
          This typically takes the form of presenting the message with a different
          color or style. The UI might also be adjusted to draw particular
          attention to the room in which the event occurred.

        Attributes:
            tweak (str): The name of the tweak to set
            valeu (Any): The tweak's value.
        """

        tweak: str = field()
        value: Any = None

        @property
        def as_value(self) -> Dict[str, Any]:
            return {"set_tweak": self.tweak, "value": self.value}

    @dataclass
    class PushUnknownAction(PushAction):
        """An unknown kind of push rule action.

        Attributes:
            action (Union[str, Dict[str, Any]]): The action as a string or dict
                from the source event.
        """
        action: Union[str, Dict[str, Any]] = field()

        @property
        def as_value(self) -> Union[str, Dict[str, Any]]:
            return self.action

    @dataclass
    class PushRule:
        """Rule stating how to notify the user for events matching some conditions.

        Attributes:
            kind (PushRuleKind): The kind of rule this is.

            id (str): A unique (within its ruleset) string identifying this rule.
                The ``id`` for default rules set by the server starts with a ``.``.
                For rules of ``room`` kind, this will be the room ID to match for.
                For rules of ``sender`` kind, this will be the user ID to match.

            default (bool): Whether this is a default rule set by the server,
                or one that the user created explicitely.

            enabled (bool): Whether this rule is currently enabled, or
                disabled and to be ignored.

            pattern (str): Only applies to ``content`` rules.
                The glob-style pattern to match message text against.

            conditions (List[PushCondition]):
                Only applies to ``override`` and ``underride`` rules.
                The conditions that must be true for an event in order for
                this rule to be applied to it.
                A rule with no condition always matches.

            actions (List[PushAction]):
                The actions to perform when this rule matches.
        """

        kind: PushRuleKind = field()
        id: str = field()
        default: bool = field()
        enabled: bool = True
        pattern: str = ""
        conditions: List[PushCondition] = field(default_factory=list)
        actions: List[PushAction] = field(default_factory=list)

        def matches(
            self, event: Event, room: "MatrixRoom", display_name: str,
        ) -> bool:
            """Return whether this push rule matches a room event.

            Args:
                event (Event): The room event to match.
                room (MatrixRoom): The room that this event is part of.
                display_name (str): The display name of our own user in the room.
            """

            if not self.enabled:
                return False

            conditions = self.conditions

            if self.kind == PushRuleKind.content:
                conditions = [PushEventMatch("content.body", self.pattern)]
            elif self.kind == PushRuleKind.room:
                conditions = [PushEventMatch("room_id", self.id)]
            elif self.kind == PushRuleKind.sender:
                conditions = [PushEventMatch("sender", self.id)]

            return all(c.matches(event, room, display_name) for c in conditions)

        @classmethod
        @verify_or_none(Schemas.push_rule)
        def from_dict(cls, rule: Dict[str, Any], kind: PushRuleKind) -> "PushRule":
            return cls(
                kind,
                rule["rule_id"],
                rule["default"],
                rule["enabled"],
                rule.get("pattern", ""),
                [PushCondition.from_dict(c) for c in rule.get("conditions", [])],
                [PushAction.from_dict(a) for a in rule.get("actions", [])],
            )

    @dataclass
    class PushRuleset:
        """A set of different kinds of push rules under a same scope.

        Attributes:
            override (List[PushRule]): Highest priority rules

            content (List[PushRule]): Rules that configure behaviors for messages
                with text matching certain patterns.

            room (List[PushRule]): Rules that configure behaviors for all messages
                in a certain room. Their ``id`` is the room's ID.

            sender (List[PushRule]): Rules that configure behaviors for all
                messages sent by a specific user. Their ``id`` is the user's ID.

            underride (List[PushRule]): Identical the ``override`` rules, but have
                a lower priority than ``content``, ``room`` and ``sender`` rules.
        """

        override: List[PushRule] = field(default_factory=list)
        content: List[PushRule] = field(default_factory=list)
        room: List[PushRule] = field(default_factory=list)
        sender: List[PushRule] = field(default_factory=list)
        underride: List[PushRule] = field(default_factory=list)

        def matching_rule(
            self, event: Event, room: "MatrixRoom", display_name: str,
        ) -> Optional[PushRule]:
            """Return the push rule in this set that matches a room event, if any.

            Args:
                event (Event): The room event to match.
                room (MatrixRoom): The room that this event is part of.
                display_name (str): The display name of our own user in the room.
            """

            for kind in PushRuleKind:
                for rule in getattr(self, kind.value):
                    if rule.matches(event, room, display_name):
                        return rule

            return None

        @classmethod
        @verify_or_none(Schemas.push_ruleset)
        def from_dict(cls, ruleset: Dict[str, Any]) -> "PushRuleset":
            kwargs = {}

            for kind in PushRuleKind:
                rules = [
                    PushRule.from_dict(rule_dict, kind)
                    for rule_dict in ruleset.get(kind.value, [])
                ]
                # PushRule.from_dict returns None if the schema verification fails
                kwargs[kind.value] = [r for r in rules if r]

            return cls(**kwargs)

        def __bool__(self) -> bool:
            return bool(
                self.override or self.content or self.room or self.sender or
                self.underride,
            )

    @dataclass
    class PushRulesEvent(AccountDataEvent):
        """Configured push rule sets for an account. Each set belongs to a scope.

        Attributes:
            global_rules (PushRuleset): Rulesets applying to all devices
            device_rules (PushRuleset): Rulesets applying to current device only
        """

        global_rules: PushRuleset = field(default_factory=PushRuleset)
        device_rules: PushRuleset = field(default_factory=PushRuleset)

        @classmethod
        @verify(Schemas.push_rules)
        def from_dict(cls, event: Dict[str, Any]) -> "PushRulesEvent":
            content = event["content"]

            return cls(
                PushRuleset.from_dict(content.get("global", {})) or PushRuleset(),
                PushRuleset.from_dict(content.get("device", {})) or PushRuleset(),
            )

        def __bool__(self) -> bool:
            return bool(self.global_rules or self.device_rules)

    @dataclass
    class UnknownAccountDataEvent(AccountDataEvent):
        """Account data event of an unknown type.

        Attributes:
            type (str): The type of the event.
            content (Dict): The content of the event.

        """

        type: str = field()
        content: Dict[str, Any] = field()

        @classmethod
        def from_dict(cls, event_dict):
            """Construct an UnknownAccountDataEvent from a dictionary."""
            content = event_dict.pop("content")
            return cls(
                event_dict["type"],
                content
            )