<commit_before>import time
import typing

import json
import jsonpickle
import logging
import uuid
from dateutil import parser
from typing import List, Dict, Text, Any, Type, Optional

from rasa_core import utils
from rasa_nlu.training_data.formats import MarkdownWriter, MarkdownReader

if typing.TYPE_CHECKING:
    from rasa_core.trackers import DialogueStateTracker

logger = logging.getLogger(__name__)


def deserialise_events(serialized_events: List[Dict[Text, Any]]
                       ) -> List['Event']:
    """Convert a list of dictionaries to a list of corresponding events.

    Example format:
        [{"event": "slot", "value": 5, "name": "my_slot"}]
    """

    deserialised = []

    for e in serialized_events:
        if "event" in e:
            event = Event.from_parameters(e)
            if event:
                deserialised.append(event)
            else:
                logger.warning("Ignoring event ({}) while deserialising "
                               "events. Couldn't parse it.")

    return deserialised


def deserialise_entities(entities):
    if isinstance(entities, str):
        entities = json.loads(entities)

    return [e for e in entities if isinstance(e, dict)]


def md_format_message(text, intent, entities):
    message_from_md = MarkdownReader()._parse_training_example(text)
    deserialised_entities = deserialise_entities(entities)
    return MarkdownWriter()._generate_message_md(
        {"text": message_from_md.text,
         "intent": intent,
         "entities": deserialised_entities}
    )


def first_key(d, default_key):
    if len(d) > 1:
        for k, v in d.items():
            if k != default_key:
                # we return the first key that is not the default key
                return k
    elif len(d) == 1:
        return list(d.keys())[0]
    else:
        return None


# noinspection PyProtectedMember
class Event(object):
    """Events describe everything that occurs in
    a conversation and tell the :class:`rasa_core.trackers.DialogueStateTracker`
    how to update its state."""

    type_name = "event"

    def __init__(self, timestamp=None):
        self.timestamp = timestamp if timestamp else time.time()

    def __ne__(self, other):
        # Not strictly necessary, but to avoid having both x==y and x!=y
        # True at the same time
        return not (self == other)

    def as_story_string(self):
        raise NotImplementedError

    @staticmethod
    def from_story_string(event_name: Text,
                          parameters: Dict[Text, Any],
                          default: Optional[Type['Event']] = None
                          ) -> Optional[List['Event']]:
        event = Event.resolve_by_type(event_name, default)

        if event:
            return event._from_story_string(parameters)
        else:
            return None

    @staticmethod
    def from_parameters(parameters: Dict[Text, Any],
                        default: Optional[Type['Event']] = None
                        ) -> Optional['Event']:

        event_name = parameters.get("event")
        if event_name is not None:
            copied = parameters.copy()
            del copied["event"]

            event = Event.resolve_by_type(event_name, default)
            if event:
                return event._from_parameters(parameters)
            else:
                return None
        else:
            return None

    @classmethod
    def _from_story_string(
        cls,
        parameters: Dict[Text, Any]
    ) -> Optional[List['Event']]:
        """Called to convert a parsed story line into an event."""
        return [cls(parameters.get("timestamp"))]

    def as_dict(self):
        return {
            "event": self.type_name,
            "timestamp": self.timestamp,
        }

    @classmethod
    def _from_parameters(cls, parameters):
        """Called to convert a dictionary of parameters to a single event.

        By default uses the same implementation as the story line
        conversation ``_from_story_string``. But the subclass might
        decide to handle parameters differently if the parsed parameters
        don't origin from a story file."""

        result = cls._from_story_string(parameters)
        if len(result) > 1:
            logger.warning("Event from parameters called with parameters "
                           "for multiple events. This is not supported, "
                           "only the first event will be returned. "
                           "Parameters: {}".format(parameters))
        return result[0] if result else None

    @staticmethod
    def resolve_by_type(
        type_name: Text,
        default: Optional[Text] = None
    ) -> Optional[Type['Event']]:
        """Returns a slots class by its type name."""

        for cls in utils.all_subclasses(Event):
            if cls.type_name == type_name:
                return cls
        if type_name == "topic":
            return None  # backwards compatibility to support old TopicSet evts
        elif default is not None:
            return default
        else:
            raise ValueError("Unknown event name '{}'.".format(type_name))

    def apply_to(self, tracker: 'DialogueStateTracker') -> None:
        pass


# noinspection PyProtectedMember
class UserUttered(Event):
    """The user has said something to the bot.

    As a side effect a new ``Turn`` will be created in the ``Tracker``."""

    type_name = "user"

    def __init__(self, text,
                 intent=None,
                 entities=None,
                 parse_data=None,
                 timestamp=None,
                 input_channel=None):
        self.text = text
        self.intent = intent if intent else {}
        self.entities = entities if entities else []
        self.input_channel = input_channel

        if parse_data:
            self.parse_data = parse_data
        else:
            self.parse_data = {
                "intent": self.intent,
                "entities": self.entities,
                "text": text,
            }

        super(UserUttered, self).__init__(timestamp)

    @staticmethod
    def _from_parse_data(text, parse_data, timestamp=None, input_channel=None):
        return UserUttered(text, parse_data["intent"], parse_data["entities"],
                           parse_data, timestamp, input_channel)

    def __hash__(self):
        return hash((self.text, self.intent.get("name"),<commit_msg>added message id to UserUttered event<commit_after>import time
import typing

import json
import jsonpickle
import logging
import uuid
from dateutil import parser
from typing import List, Dict, Text, Any, Type, Optional

from rasa_core import utils
from rasa_nlu.training_data.formats import MarkdownWriter, MarkdownReader

if typing.TYPE_CHECKING:
    from rasa_core.trackers import DialogueStateTracker

logger = logging.getLogger(__name__)


def deserialise_events(serialized_events: List[Dict[Text, Any]]
                       ) -> List['Event']:
    """Convert a list of dictionaries to a list of corresponding events.

    Example format:
        [{"event": "slot", "value": 5, "name": "my_slot"}]
    """

    deserialised = []

    for e in serialized_events:
        if "event" in e:
            event = Event.from_parameters(e)
            if event:
                deserialised.append(event)
            else:
                logger.warning("Ignoring event ({}) while deserialising "
                               "events. Couldn't parse it.")

    return deserialised


def deserialise_entities(entities):
    if isinstance(entities, str):
        entities = json.loads(entities)

    return [e for e in entities if isinstance(e, dict)]


def md_format_message(text, intent, entities):
    message_from_md = MarkdownReader()._parse_training_example(text)
    deserialised_entities = deserialise_entities(entities)
    return MarkdownWriter()._generate_message_md(
        {"text": message_from_md.text,
         "intent": intent,
         "entities": deserialised_entities}
    )


def first_key(d, default_key):
    if len(d) > 1:
        for k, v in d.items():
            if k != default_key:
                # we return the first key that is not the default key
                return k
    elif len(d) == 1:
        return list(d.keys())[0]
    else:
        return None


# noinspection PyProtectedMember
class Event(object):
    """Events describe everything that occurs in
    a conversation and tell the :class:`rasa_core.trackers.DialogueStateTracker`
    how to update its state."""

    type_name = "event"

    def __init__(self, timestamp=None):
        self.timestamp = timestamp if timestamp else time.time()

    def __ne__(self, other):
        # Not strictly necessary, but to avoid having both x==y and x!=y
        # True at the same time
        return not (self == other)

    def as_story_string(self):
        raise NotImplementedError

    @staticmethod
    def from_story_string(event_name: Text,
                          parameters: Dict[Text, Any],
                          default: Optional[Type['Event']] = None
                          ) -> Optional[List['Event']]:
        event = Event.resolve_by_type(event_name, default)

        if event:
            return event._from_story_string(parameters)
        else:
            return None

    @staticmethod
    def from_parameters(parameters: Dict[Text, Any],
                        default: Optional[Type['Event']] = None
                        ) -> Optional['Event']:

        event_name = parameters.get("event")
        if event_name is not None:
            copied = parameters.copy()
            del copied["event"]

            event = Event.resolve_by_type(event_name, default)
            if event:
                return event._from_parameters(parameters)
            else:
                return None
        else:
            return None

    @classmethod
    def _from_story_string(
        cls,
        parameters: Dict[Text, Any]
    ) -> Optional[List['Event']]:
        """Called to convert a parsed story line into an event."""
        return [cls(parameters.get("timestamp"))]

    def as_dict(self):
        return {
            "event": self.type_name,
            "timestamp": self.timestamp,
        }

    @classmethod
    def _from_parameters(cls, parameters):
        """Called to convert a dictionary of parameters to a single event.

        By default uses the same implementation as the story line
        conversation ``_from_story_string``. But the subclass might
        decide to handle parameters differently if the parsed parameters
        don't origin from a story file."""

        result = cls._from_story_string(parameters)
        if len(result) > 1:
            logger.warning("Event from parameters called with parameters "
                           "for multiple events. This is not supported, "
                           "only the first event will be returned. "
                           "Parameters: {}".format(parameters))
        return result[0] if result else None

    @staticmethod
    def resolve_by_type(
        type_name: Text,
        default: Optional[Text] = None
    ) -> Optional[Type['Event']]:
        """Returns a slots class by its type name."""

        for cls in utils.all_subclasses(Event):
            if cls.type_name == type_name:
                return cls
        if type_name == "topic":
            return None  # backwards compatibility to support old TopicSet evts
        elif default is not None:
            return default
        else:
            raise ValueError("Unknown event name '{}'.".format(type_name))

    def apply_to(self, tracker: 'DialogueStateTracker') -> None:
        pass


# noinspection PyProtectedMember
class UserUttered(Event):
    """The user has said something to the bot.

    As a side effect a new ``Turn`` will be created in the ``Tracker``."""

    type_name = "user"

    def __init__(self, text,
                 intent=None,
                 entities=None,
                 parse_data=None,
                 timestamp=None,
                 input_channel=None,
                 message_id=None):
        self.text = text
        self.intent = intent if intent else {}
        self.entities = entities if entities else []
        self.input_channel = input_channel
        self.message_id = message_id

        if parse_data:
            self.parse_data = parse_data
        else:
            self.parse_data = {
                "intent": self.intent,
                "entities": self.entities,
                "text": text,
            }

        super(UserUttered, self).__init__(timestamp)

    @staticmethod
    def _from_parse_data(text, parse_data, timestamp=None, input_channel=None):
        return UserUttered(text, parse_data["intent"], parse_data["entities"],
                           parse_data, timestamp, input_channel)

    def __hash__(self):
        return hash((self.text, self.intent.get("name"),
{'fraction_non_alphanumeric': 0.08922104938747207, 'fraction_numerical': 0.0009245704599738038, 'mean_word_length': 2.1097268806899856, 'pattern_counts': {'":': 24, '<': 3, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5804157', 'n_tokens_mistral': 3457, 'n_tokens_neox': 3398, 'n_words': 1037}