Source code for hangups.conversation_event

"""ConversationEvent base class and subclasses.

These classes are wrappers for hangouts_pb2.Event instances. Parsing is done
through property methods, which prefer logging warnings to raising exceptions.
"""

import logging

from hangups import parsers, message_parser, user, hangouts_pb2

logger = logging.getLogger(__name__)
chat_message_parser = message_parser.ChatMessageParser()


[docs]class ConversationEvent: """An event which becomes part of the permanent record of a conversation. This is a wrapper for the ``Event`` message, which may contain one of many subtypes, represented here as other subclasses. Args: event: ``Event`` message. """ def __init__(self, event): self._event = event # Event @property def timestamp(self): """When the event occurred (:class:`datetime.datetime`).""" return parsers.from_timestamp(self._event.timestamp) @property def user_id(self): """Who created the event (:class:`~hangups.user.UserID`).""" return user.UserID(chat_id=self._event.sender_id.chat_id, gaia_id=self._event.sender_id.gaia_id) @property def conversation_id(self): """ID of the conversation containing the event (:class:`str`).""" return self._event.conversation_id.id @property def id_(self): """ID of this event (:class:`str`).""" return self._event.event_id
[docs]class ChatMessageSegment: """A segment of a chat message in :class:`ChatMessageEvent`. Args: text (str): Text of the segment. segment_type: (optional) One of ``SEGMENT_TYPE_TEXT``, ``SEGMENT_TYPE_LINE_BREAK``, or ``SEGMENT_TYPE_LINK``. Defaults to ``SEGMENT_TYPE_TEXT``, or ``SEGMENT_TYPE_LINK`` if ``link_target`` is specified. is_bold (bool): (optional) Whether the text is bold. Defaults to ``False``. is_italic (bool): (optional) Whether the text is italic. Defaults to ``False``. is_strikethrough (bool): (optional) Whether the text is struck through. Defaults to ``False``. is_underline (bool): (optional) Whether the text is underlined. Defaults to ``False``. link_target (str): (option) URL to link to. Defaults to ``None``. """ def __init__(self, text, segment_type=None, is_bold=False, is_italic=False, is_strikethrough=False, is_underline=False, link_target=None): """Create a new chat message segment.""" if segment_type is not None: self.type_ = segment_type elif link_target is not None: self.type_ = hangouts_pb2.SEGMENT_TYPE_LINK else: self.type_ = hangouts_pb2.SEGMENT_TYPE_TEXT self.text = text self.is_bold = is_bold self.is_italic = is_italic self.is_strikethrough = is_strikethrough self.is_underline = is_underline self.link_target = link_target
[docs] @staticmethod def from_str(text): """Construct :class:`ChatMessageSegment` list parsed from a string. Args: text (str): Text to parse. May contain line breaks, URLs and formatting markup (simplified Markdown and HTML) to be converted into equivalent segments. Returns: List of :class:`ChatMessageSegment` objects. """ segment_list = chat_message_parser.parse(text) return [ChatMessageSegment(segment.text, **segment.params) for segment in segment_list]
[docs] @staticmethod def deserialize(segment): """Construct :class:`ChatMessageSegment` from ``Segment`` message. Args: segment: ``Segment`` message to parse. Returns: :class:`ChatMessageSegment` object. """ link_target = segment.link_data.link_target return ChatMessageSegment( segment.text, segment_type=segment.type, is_bold=segment.formatting.bold, is_italic=segment.formatting.italic, is_strikethrough=segment.formatting.strikethrough, is_underline=segment.formatting.underline, link_target=None if link_target == '' else link_target )
[docs] def serialize(self): """Serialize this segment to a ``Segment`` message. Returns: ``Segment`` message. """ segment = hangouts_pb2.Segment( type=self.type_, text=self.text, formatting=hangouts_pb2.Formatting( bold=self.is_bold, italic=self.is_italic, strikethrough=self.is_strikethrough, underline=self.is_underline, ), ) if self.link_target is not None: segment.link_data.link_target = self.link_target return segment
[docs]class ChatMessageEvent(ConversationEvent): """An event that adds a new message to a conversation. Corresponds to the ``ChatMessage`` message. """ @property def text(self): """Text of the message without formatting (:class:`str`).""" lines = [''] for segment in self.segments: if segment.type_ == hangouts_pb2.SEGMENT_TYPE_TEXT: lines[-1] += segment.text elif segment.type_ == hangouts_pb2.SEGMENT_TYPE_LINK: lines[-1] += segment.text elif segment.type_ == hangouts_pb2.SEGMENT_TYPE_LINE_BREAK: lines.append('') else: logger.warning('Ignoring unknown chat message segment type: {}' .format(segment.type_)) lines.extend(self.attachments) return '\n'.join(lines) @property def segments(self): """List of :class:`ChatMessageSegment` in message (:class:`list`).""" seg_list = self._event.chat_message.message_content.segment return [ChatMessageSegment.deserialize(seg) for seg in seg_list] @property def attachments(self): """List of attachments in the message (:class:`list`).""" raw_attachments = self._event.chat_message.message_content.attachment if raw_attachments is None: raw_attachments = [] attachments = [] for attachment in raw_attachments: for embed_item_type in attachment.embed_item.type: known_types = [ hangouts_pb2.ITEM_TYPE_PLUS_PHOTO, hangouts_pb2.ITEM_TYPE_PLACE_V2, hangouts_pb2.ITEM_TYPE_PLACE, hangouts_pb2.ITEM_TYPE_THING, ] if embed_item_type not in known_types: logger.warning('Received chat message attachment with ' 'unknown embed type: %r', embed_item_type) if attachment.embed_item.HasField('plus_photo'): attachments.append( attachment.embed_item.plus_photo.thumbnail.image_url ) return attachments
[docs]class OTREvent(ConversationEvent): """An event that changes a conversation's OTR (history) mode. Corresponds to the ``OTRModification`` message. """ @property def new_otr_status(self): """The conversation's new OTR status. May be either ``OFF_THE_RECORD_STATUS_OFF_THE_RECORD`` or ``OFF_THE_RECORD_STATUS_ON_THE_RECORD``. """ return self._event.otr_modification.new_otr_status @property def old_otr_status(self): """The conversation's old OTR status. May be either ``OFF_THE_RECORD_STATUS_OFF_THE_RECORD`` or ``OFF_THE_RECORD_STATUS_ON_THE_RECORD``. """ return self._event.otr_modification.old_otr_status
[docs]class RenameEvent(ConversationEvent): """An event that renames a conversation. Corresponds to the ``ConversationRename`` message. """ @property def new_name(self): """The conversation's new name (:class:`str`). May be an empty string if the conversation's name was cleared. """ return self._event.conversation_rename.new_name @property def old_name(self): """The conversation's old name (:class:`str`). May be an empty string if the conversation had no previous name. """ return self._event.conversation_rename.old_name
[docs]class MembershipChangeEvent(ConversationEvent): """An event that adds or removes a conversation participant. Corresponds to the ``MembershipChange`` message. """ @property def type_(self): """The type of membership change. May be either ``MEMBERSHIP_CHANGE_TYPE_JOIN`` or ``MEMBERSHIP_CHANGE_TYPE_LEAVE``. """ return self._event.membership_change.type @property def participant_ids(self): """:class:`~hangups.user.UserID` of users involved (:class:`list`).""" return [user.UserID(chat_id=id_.chat_id, gaia_id=id_.gaia_id) for id_ in self._event.membership_change.participant_ids]
[docs]class HangoutEvent(ConversationEvent): """An event that is related to a Hangout voice or video call. Corresponds to the ``HangoutEvent`` message. """ @property def event_type(self): """The Hangout event type. May be one of ``HANGOUT_EVENT_TYPE_START``, ``HANGOUT_EVENT_TYPE_END``, ``HANGOUT_EVENT_TYPE_JOIN``, ``HANGOUT_EVENT_TYPE_LEAVE``, ``HANGOUT_EVENT_TYPE_COMING_SOON``, or ``HANGOUT_EVENT_TYPE_ONGOING``. """ return self._event.hangout_event.event_type
[docs]class GroupLinkSharingModificationEvent(ConversationEvent): """An event that modifies a conversation's group link sharing status. Corresponds to the ``GroupLinkSharingModification`` message. """ @property def new_status(self): """The new group link sharing status. May be either ``GROUP_LINK_SHARING_STATUS_ON`` or ``GROUP_LINK_SHARING_STATUS_OFF``. """ return self._event.group_link_sharing_modification.new_status