git.maemo.org Git - theonering/blob - src/gvoice/conversations.py

   1 #!/usr/bin/python
   2
   3 from __future__ import with_statement
   4
   5 import datetime
   6 import logging
   7
   8 try:
   9         import cPickle
  10         pickle = cPickle
  11 except ImportError:
  12         import pickle
  13
  14 import constants
  15 import util.coroutines as coroutines
  16 import util.misc as misc_utils
  17 import util.go_utils as gobject_utils
  18
  19
  20 _moduleLogger = logging.getLogger(__name__)
  21
  22
  23 class ConversationError(RuntimeError):
  24
  25         pass
  26
  27
  28 class Conversations(object):
  29
  30         OLDEST_COMPATIBLE_FORMAT_VERSION = misc_utils.parse_version("0.8.0")
  31
  32         def __init__(self, getter, asyncPool):
  33                 self._get_raw_conversations = getter
  34                 self._asyncPool = asyncPool
  35                 self._conversations = {}
  36                 self._loadedFromCache = False
  37                 self._hasDoneUpdate = False
  38
  39                 self.updateSignalHandler = coroutines.CoTee()
  40
  41         @property
  42         def _name(self):
  43                 return repr(self._get_raw_conversations.__name__)
  44
  45         def load(self, path):
  46                 _moduleLogger.debug("%s Loading cache" % (self._name, ))
  47                 assert not self._conversations
  48                 try:
  49                         with open(path, "rb") as f:
  50                                 fileVersion, fileBuild, convs = pickle.load(f)
  51                 except (pickle.PickleError, IOError, EOFError, ValueError, Exception):
  52                         _moduleLogger.exception("While loading for %s" % self._name)
  53                         return
  54
  55                 if convs and misc_utils.compare_versions(
  56                         self.OLDEST_COMPATIBLE_FORMAT_VERSION,
  57                         misc_utils.parse_version(fileVersion),
  58                 ) <= 0:
  59                         _moduleLogger.info("%s Loaded cache" % (self._name, ))
  60                         self._conversations = convs
  61                         self._loadedFromCache = True
  62                 else:
  63                         _moduleLogger.debug(
  64                                 "%s Skipping cache due to version mismatch (%s-%s)" % (
  65                                         self._name, fileVersion, fileBuild
  66                                 )
  67                         )
  68
  69         def save(self, path):
  70                 _moduleLogger.info("%s Saving cache" % (self._name, ))
  71                 if not self._conversations:
  72                         _moduleLogger.info("%s Odd, no conversations to cache.  Did we never load the cache?" % (self._name, ))
  73                         return
  74
  75                 try:
  76                         dataToDump = (constants.__version__, constants.__build__, self._conversations)
  77                         with open(path, "wb") as f:
  78                                 pickle.dump(dataToDump, f, pickle.HIGHEST_PROTOCOL)
  79                 except (pickle.PickleError, IOError):
  80                         _moduleLogger.exception("While saving for %s" % self._name)
  81                 _moduleLogger.info("%s Cache saved" % (self._name, ))
  82
  83         def update(self, force=False):
  84                 if not force and self._conversations:
  85                         return
  86
  87                 le = gobject_utils.AsyncLinearExecution(self._asyncPool, self._update)
  88                 le.start()
  89
  90         @misc_utils.log_exception(_moduleLogger)
  91         def _update(self):
  92                 try:
  93                         conversationResult = yield (
  94                                 self._get_raw_conversations,
  95                                 (),
  96                                 {},
  97                         )
  98                 except Exception:
  99                         _moduleLogger.exception("%s While updating conversations" % (self._name, ))
 100                         return
 101
 102                 oldConversationIds = set(self._conversations.iterkeys())
 103
 104                 updateConversationIds = set()
 105                 conversations = list(conversationResult)
 106                 conversations.sort()
 107                 for conversation in conversations:
 108                         key = misc_utils.normalize_number(conversation.number)
 109                         try:
 110                                 mergedConversations = self._conversations[key]
 111                         except KeyError:
 112                                 mergedConversations = MergedConversations()
 113                                 self._conversations[key] = mergedConversations
 114
 115                         if self._loadedFromCache or self._hasDoneUpdate:
 116                                 markAllAsRead = False
 117                         else:
 118                                 markAllAsRead = True
 119
 120                         try:
 121                                 mergedConversations.append_conversation(conversation, markAllAsRead)
 122                                 isConversationUpdated = True
 123                         except ConversationError, e:
 124                                 isConversationUpdated = False
 125                         except AssertionError, e:
 126                                 _moduleLogger.debug("%s Skipping conversation for %r because '%s'" % (self._name, key, e))
 127                                 isConversationUpdated = False
 128                         except RuntimeError, e:
 129                                 _moduleLogger.debug("%s Skipping conversation for %r because '%s'" % (self._name, key, e))
 130                                 isConversationUpdated = False
 131
 132                         if isConversationUpdated:
 133                                 updateConversationIds.add(key)
 134
 135                 for key in updateConversationIds:
 136                         mergedConv = self._conversations[key]
 137                         _moduleLogger.debug("%s \tUpdated %s" % (self._name, key))
 138                         for conv in mergedConv.conversations:
 139                                 message = "%s \t\tUpdated %s (%r) %r %r %r" % (
 140                                         self._name, conv.id, conv.time, conv.isRead, conv.isArchived, len(conv.messages)
 141                                 )
 142                                 _moduleLogger.debug(message)
 143
 144                 if updateConversationIds:
 145                         message = (self, updateConversationIds, )
 146                         self.updateSignalHandler.stage.send(message)
 147                 self._hasDoneUpdate = True
 148
 149         def get_conversations(self):
 150                 return self._conversations.iterkeys()
 151
 152         def get_conversation(self, key):
 153                 return self._conversations[key]
 154
 155         def clear_conversation(self, key):
 156                 try:
 157                         del self._conversations[key]
 158                 except KeyError:
 159                         _moduleLogger.info("%s Conversation never existed for %r" % (self._name, key, ))
 160
 161         def clear_all(self):
 162                 self._conversations.clear()
 163
 164
 165 class MergedConversations(object):
 166
 167         def __init__(self):
 168                 self._conversations = []
 169
 170         def append_conversation(self, newConversation, markAllAsRead):
 171                 self._validate(newConversation)
 172                 for similarConversation in self._find_related_conversation(newConversation.id):
 173                         self._update_previous_related_conversation(similarConversation, newConversation)
 174                         self._remove_repeats(similarConversation, newConversation)
 175
 176                 # HACK: Because GV marks all messages as read when you reply it has
 177                 # the following race:
 178                 # 1. Get all messages
 179                 # 2. Contact sends a text
 180                 # 3. User sends a text marking contacts text as read
 181                 # 4. Get all messages not returning text from step 2
 182                 # This isn't a problem for voicemails but we don't know(?( enough.
 183                 # So we hack around this by:
 184                 # * We cache to disk the history of messages sent/received
 185                 # * On first run we mark all server messages as read due to no cache
 186                 # * If not first load or from cache (disk or in-memory) then it must be unread
 187                 if newConversation.type != newConversation.TYPE_VOICEMAIL:
 188                         if markAllAsRead:
 189                                 newConversation.isRead = True
 190                         else:
 191                                 newConversation.isRead = False
 192
 193                 if newConversation.messages:
 194                         # must not have had all items removed due to duplicates
 195                         self._conversations.append(newConversation)
 196
 197         def to_dict(self):
 198                 selfDict = {}
 199                 selfDict["conversations"] = [conv.to_dict() for conv in self._conversations]
 200                 return selfDict
 201
 202         @property
 203         def conversations(self):
 204                 return self._conversations
 205
 206         def _validate(self, newConversation):
 207                 if not self._conversations:
 208                         return
 209
 210                 for constantField in ("number", ):
 211                         assert getattr(self._conversations[0], constantField) == getattr(newConversation, constantField), "Constant field changed, soemthing is seriously messed up: %r v %r" % (
 212                                 getattr(self._conversations[0], constantField),
 213                                 getattr(newConversation, constantField),
 214                         )
 215
 216                 if newConversation.time <= self._conversations[-1].time:
 217                         raise ConversationError("Conversations got out of order")
 218
 219         def _find_related_conversation(self, convId):
 220                 similarConversations = (
 221                         conversation
 222                         for conversation in self._conversations
 223                         if conversation.id == convId
 224                 )
 225                 return similarConversations
 226
 227         def _update_previous_related_conversation(self, relatedConversation, newConversation):
 228                 for commonField in ("isSpam", "isTrash", "isArchived"):
 229                         newValue = getattr(newConversation, commonField)
 230                         setattr(relatedConversation, commonField, newValue)
 231
 232         def _remove_repeats(self, relatedConversation, newConversation):
 233                 newConversationMessages = newConversation.messages
 234                 newConversation.messages = [
 235                         newMessage
 236                         for newMessage in newConversationMessages
 237                         if newMessage not in relatedConversation.messages
 238                 ]
 239                 _moduleLogger.debug("Found %d new messages in conversation %s (%d/%d)" % (
 240                         len(newConversationMessages) - len(newConversation.messages),
 241                         newConversation.id,
 242                         len(newConversation.messages),
 243                         len(newConversationMessages),
 244                 ))
 245                 assert 0 < len(newConversation.messages), "Everything shouldn't have been removed"
 246
 247
 248 def filter_out_read(conversations):
 249         return (
 250                 conversation
 251                 for conversation in conversations
 252                 if not conversation.isRead and not conversation.isArchived
 253         )
 254
 255
 256 def is_message_from_self(message):
 257         return message.whoFrom == "Me:"
 258
 259
 260 def filter_out_self(conversations):
 261         return (
 262                 newConversation
 263                 for newConversation in conversations
 264                 if len(newConversation.messages) and any(
 265                         not is_message_from_self(message)
 266                         for message in newConversation.messages
 267                 )
 268         )
 269
 270
 271 class FilterOutReported(object):
 272
 273         NULL_TIMESTAMP = datetime.datetime(1, 1, 1)
 274
 275         def __init__(self):
 276                 self._lastMessageTimestamp = self.NULL_TIMESTAMP
 277
 278         def get_last_timestamp(self):
 279                 return self._lastMessageTimestamp
 280
 281         def __call__(self, conversations):
 282                 filteredConversations = [
 283                         conversation
 284                         for conversation in conversations
 285                         if self._lastMessageTimestamp < conversation.time
 286                 ]
 287                 if filteredConversations and self._lastMessageTimestamp < filteredConversations[0].time:
 288                         self._lastMessageTimestamp = filteredConversations[0].time
 289                 return filteredConversations
 290
 291
 292 def print_conversations(path):
 293         import pprint
 294
 295         try:
 296                 with open(path, "rb") as f:
 297                         fileVersion, fileBuild, convs = pickle.load(f)
 298         except (pickle.PickleError, IOError, EOFError, ValueError):
 299                 _moduleLogger.exception("")
 300         else:
 301                 for key, value in convs.iteritems():
 302                         convs[key] = value.to_dict()
 303                 pprint.pprint((fileVersion, fileBuild))
 304                 pprint.pprint(convs)