3 from __future__ import with_statement
15 import util.coroutines as coroutines
16 import util.misc as misc_utils
19 _moduleLogger = logging.getLogger(__name__)
22 class Conversations(object):
24 OLDEST_COMPATIBLE_FORMAT_VERSION = misc_utils.parse_version("0.8.0")
25 OLDEST_MESSAGE_WINDOW = datetime.timedelta(days=60)
27 def __init__(self, getter):
28 self._get_raw_conversations = getter
29 self._conversations = {}
31 self.updateSignalHandler = coroutines.CoTee()
35 return repr(self._get_raw_conversations.__name__)
38 assert not self._conversations
40 with open(path, "rb") as f:
41 fileVersion, fileBuild, convs = pickle.load(f)
42 except (pickle.PickleError, IOError, EOFError, ValueError):
43 _moduleLogger.exception("While loading for %s" % self._name)
46 if misc_utils.compare_versions(
47 self.OLDEST_COMPATIBLE_FORMAT_VERSION,
48 misc_utils.parse_version(fileVersion),
50 _moduleLogger.info("%s Loaded cache" % (self._name, ))
51 self._conversations = convs
54 "%s Skipping cache due to version mismatch (%s-%s)" % (
55 self._name, fileVersion, fileBuild
61 _moduleLogger.info("%s Saving cache" % (self._name, ))
62 for conv in self._conversations.itervalues():
63 conv.compress(self.OLDEST_MESSAGE_WINDOW)
64 dataToDump = (constants.__version__, constants.__build__, self._conversations)
65 with open(path, "wb") as f:
66 pickle.dump(dataToDump, f, pickle.HIGHEST_PROTOCOL)
67 except (pickle.PickleError, IOError):
68 _moduleLogger.exception("While saving for %s" % self._name)
70 def update(self, force=False):
71 if not force and self._conversations:
74 oldConversationIds = set(self._conversations.iterkeys())
76 updateConversationIds = set()
77 conversations = list(self._get_raw_conversations())
79 for conversation in conversations:
80 key = misc_utils.normalize_number(conversation.number)
82 mergedConversations = self._conversations[key]
84 mergedConversations = MergedConversations()
85 self._conversations[key] = mergedConversations
88 mergedConversations.append_conversation(conversation)
89 isConversationUpdated = True
90 except RuntimeError, e:
92 _moduleLogger.debug("%s Skipping conversation for %r because '%s'" % (self._name, key, e))
93 isConversationUpdated = False
95 if isConversationUpdated:
96 updateConversationIds.add(key)
98 if updateConversationIds:
99 message = (self, updateConversationIds, )
100 self.updateSignalHandler.stage.send(message)
102 def get_conversations(self):
103 return self._conversations.iterkeys()
105 def get_conversation(self, key):
106 return self._conversations[key]
108 def clear_conversation(self, key):
110 del self._conversations[key]
112 _moduleLogger.info("%s Conversation never existed for %r" % (self._name, key, ))
115 self._conversations.clear()
118 class MergedConversations(object):
121 self._conversations = []
123 def append_conversation(self, newConversation):
124 self._validate(newConversation)
126 for similarConversation in self._find_related_conversation(newConversation.id):
127 self._update_previous_related_conversation(similarConversation, newConversation)
128 self._remove_repeats(similarConversation, newConversation)
131 # Hack to reduce a race window with GV marking messages as read
132 # because it thinks we replied when really we replied to the
133 # previous message. Clients of this code are expected to handle
134 # this gracefully. Other race conditions may exist but clients are
135 # responsible for them
136 if newConversation.messages:
137 newConversation.isRead = False
139 newConversation.isRead = True
140 self._conversations.append(newConversation)
144 selfDict["conversations"] = [conv.to_dict() for conv in self._conversations]
148 def conversations(self):
149 return self._conversations
151 def compress(self, timedelta):
152 now = datetime.datetime.now()
153 oldNumConvs = len(self._conversations)
154 oldConvs = self._conversations
155 self._conversations = [
157 for conv in self._conversations
158 if (now - conv.time) < timedelta
160 newNumConvs = len(self._conversations)
161 if oldNumConvs != newNumConvs:
162 _moduleLogger.debug("Compressed conversations from %s to %s" % (oldNumConvs, newNumConvs))
164 _moduleLogger.debug("Did not compress, %s" % (newNumConvs))
166 def _validate(self, newConversation):
167 if not self._conversations:
170 for constantField in ("number", ):
171 assert getattr(self._conversations[0], constantField) == getattr(newConversation, constantField), "Constant field changed, soemthing is seriously messed up: %r v %r" % (
172 getattr(self._conversations[0], constantField),
173 getattr(newConversation, constantField),
176 if newConversation.time <= self._conversations[-1].time:
177 raise RuntimeError("Conversations got out of order")
179 def _find_related_conversation(self, convId):
180 similarConversations = (
182 for conversation in self._conversations
183 if conversation.id == convId
185 return similarConversations
187 def _update_previous_related_conversation(self, relatedConversation, newConversation):
188 for commonField in ("isSpam", "isTrash", "isArchived"):
189 newValue = getattr(newConversation, commonField)
190 setattr(relatedConversation, commonField, newValue)
192 def _remove_repeats(self, relatedConversation, newConversation):
193 newConversationMessages = newConversation.messages
194 newConversation.messages = [
196 for newMessage in newConversationMessages
197 if newMessage not in relatedConversation.messages
199 _moduleLogger.debug("Found %d new messages in conversation %s (%d/%d)" % (
200 len(newConversationMessages) - len(newConversation.messages),
202 len(newConversation.messages),
203 len(newConversationMessages),
205 assert 0 < len(newConversation.messages), "Everything shouldn't have been removed"
208 def filter_out_read(conversations):
211 for conversation in conversations
212 if not conversation.isRead and not conversation.isArchived
216 def is_message_from_self(message):
217 return message.whoFrom == "Me:"
220 def filter_out_self(conversations):
223 for newConversation in conversations
224 if len(newConversation.messages) and any(
225 not is_message_from_self(message)
226 for message in newConversation.messages
231 class FilterOutReported(object):
233 NULL_TIMESTAMP = datetime.datetime(1, 1, 1)
236 self._lastMessageTimestamp = self.NULL_TIMESTAMP
238 def get_last_timestamp(self):
239 return self._lastMessageTimestamp
241 def __call__(self, conversations):
242 filteredConversations = [
244 for conversation in conversations
245 if self._lastMessageTimestamp < conversation.time
247 if filteredConversations and self._lastMessageTimestamp < filteredConversations[0].time:
248 self._lastMessageTimestamp = filteredConversations[0].time
249 return filteredConversations