Source code for hangups.parsers
"""Parsing helper functions."""
from collections import namedtuple
import datetime
import logging
from hangups import user, hangouts_pb2
logger = logging.getLogger(__name__)
##############################################################################
# Message parsing utils
##############################################################################
def from_timestamp(microsecond_timestamp):
"""Convert a microsecond timestamp to a UTC datetime instance."""
# Create datetime without losing precision from floating point (yes, this
# is actually needed):
return datetime.datetime.fromtimestamp(
microsecond_timestamp // 1000000, datetime.timezone.utc
).replace(microsecond=(microsecond_timestamp % 1000000))
def to_timestamp(datetime_timestamp):
"""Convert UTC datetime to microsecond timestamp used by Hangouts."""
return int(datetime_timestamp.timestamp() * 1000000)
def from_participantid(participant_id):
"""Convert hangouts_pb2.ParticipantId to UserID."""
return user.UserID(
chat_id=participant_id.chat_id,
gaia_id=participant_id.gaia_id
)
def to_participantid(user_id):
"""Convert UserID to hangouts_pb2.ParticipantId."""
return hangouts_pb2.ParticipantId(
chat_id=user_id.chat_id,
gaia_id=user_id.gaia_id
)
##############################################################################
# Message types and parsers
##############################################################################
TypingStatusMessage = namedtuple(
'TypingStatusMessage', ['conv_id', 'user_id', 'timestamp', 'status']
)
"""
A notification about a user's typing status in a conversation.
Args:
conv_id (str): ID of the conversation.
user_id (hangups.user.UserID): ID of the affected user.
timestamp (datetime.datetime): When the notification was generated.
status: The new status; one of ``TYPING_TYPE_STARTED``,
``TYPING_TYPE_PAUSED``, or ``TYPING_TYPE_STOPPED``.
"""
def parse_typing_status_message(p):
"""Return TypingStatusMessage from hangouts_pb2.SetTypingNotification.
The same status may be sent multiple times consecutively, and when a
message is sent the typing status will not change to stopped.
"""
return TypingStatusMessage(
conv_id=p.conversation_id.id,
user_id=from_participantid(p.sender_id),
timestamp=from_timestamp(p.timestamp),
status=p.type,
)
WatermarkNotification = namedtuple(
'WatermarkNotification', ['conv_id', 'user_id', 'read_timestamp']
)
"""A notification about a user's watermark (read timestamp).
Args:
conv_id (str): ID of the conversation.
user_id (hangups.user.UserID): ID of the affected user.
read_timestamp (datetime.datetime): The new watermark.
"""
def parse_watermark_notification(p):
"""Return WatermarkNotification from hangouts_pb2.WatermarkNotification."""
return WatermarkNotification(
conv_id=p.conversation_id.id,
user_id=from_participantid(p.sender_id),
read_timestamp=from_timestamp(
p.latest_read_timestamp
),
)