import logging
import inspect
+from xml.sax import saxutils
from xml.etree import ElementTree
try:
return "%s (%s): %s" % (
self.whoFrom,
self.when,
- "".join(str(part) for part in self.body)
+ "".join(unicode(part) for part in self.body)
)
def to_dict(self):
self._XML_ACCOUNT_URL = SECURE_URL_BASE + "contacts/"
# HACK really this redirects to the main pge and we are grabbing some javascript
self._XML_CONTACTS_URL = "http://www.google.com/voice/inbox/search/contact"
+ self._CSV_CONTACTS_URL = "http://mail.google.com/mail/contacts/data/export"
+ self._JSON_CONTACTS_URL = SECURE_URL_BASE + "b/0/request/user"
self._XML_RECENT_URL = SECURE_URL_BASE + "inbox/recent/"
self.XML_FEEDS = (
self._XML_TRASH_URL = SECURE_URL_BASE + "inbox/recent/trash"
self._XML_VOICEMAIL_URL = SECURE_URL_BASE + "inbox/recent/voicemail/"
self._XML_SMS_URL = SECURE_URL_BASE + "inbox/recent/sms/"
+ self._JSON_SMS_URL = SECURE_URL_BASE + "b/0/request/messages/"
+ self._JSON_SMS_COUNT_URL = SECURE_URL_BASE + "b/0/request/unread"
self._XML_RECORDED_URL = SECURE_URL_BASE + "inbox/recent/recorded/"
self._XML_PLACED_URL = SECURE_URL_BASE + "inbox/recent/placed/"
self._XML_RECEIVED_URL = SECURE_URL_BASE + "inbox/recent/received/"
self._accountNumRe = re.compile(r"""<b class="ms\d">(.{14})</b></div>""")
self._callbackRe = re.compile(r"""\s+(.*?):\s*(.*?)<br\s*/>\s*$""", re.M)
- self._contactsBodyRe = re.compile(r"""gcData\s*=\s*({.*?});""", re.MULTILINE | re.DOTALL)
self._seperateVoicemailsRegex = re.compile(r"""^\s*<div id="(\w+)"\s* class=".*?gc-message.*?">""", re.MULTILINE | re.DOTALL)
self._exactVoicemailTimeRegex = re.compile(r"""<span class="gc-message-time">(.*?)</span>""", re.MULTILINE)
self._relativeVoicemailTimeRegex = re.compile(r"""<span class="gc-message-relative">(.*?)</span>""", re.MULTILINE)
Attempts to detect a current session
@note Once logged in try not to reauth more than once a minute.
@returns If authenticated
+ @blocks
"""
isRecentledAuthed = (time.time() - self._lastAuthed) < 120
isPreviouslyAuthed = self._token is not None
"""
Attempt to login to GoogleVoice
@returns Whether login was successful or not
+ @blocks
"""
self.logout()
galxToken = self._get_token()
self._lastAuthed = time.time()
return True
+ def persist(self):
+ self._browser.save_cookies()
+
+ def shutdown(self):
+ self._browser.save_cookies()
+ self._token = None
+ self._lastAuthed = 0.0
+
def logout(self):
self._browser.clear_cookies()
self._browser.save_cookies()
self._lastAuthed = 0.0
def is_dnd(self):
+ """
+ @blocks
+ """
isDndPage = self._get_page(self._isDndURL)
dndGroup = self._isDndRe.search(isDndPage)
return isDnd
def set_dnd(self, doNotDisturb):
+ """
+ @blocks
+ """
dndPostData = {
"doNotDisturb": 1 if doNotDisturb else 0,
}
def call(self, outgoingNumber):
"""
This is the main function responsible for initating the callback
+ @blocks
"""
outgoingNumber = self._send_validation(outgoingNumber)
subscriberNumber = None
"""
Cancels a call matching outgoing and forwarding numbers (if given).
Will raise an error if no matching call is being placed
+ @blocks
"""
page = self._get_page_with_token(
self._callCancelURL,
self._parse_with_validation(page)
def send_sms(self, phoneNumbers, message):
+ """
+ @blocks
+ """
validatedPhoneNumbers = [
self._send_validation(phoneNumber)
for phoneNumber in phoneNumbers
self._sendSmsURL,
{
'phoneNumber': flattenedPhoneNumbers,
- 'text': message
+ 'text': unicode(message).encode("utf-8"),
},
)
self._parse_with_validation(page)
"""
Search your Google Voice Account history for calls, voicemails, and sms
Returns ``Folder`` instance containting matching messages
+ @blocks
"""
page = self._get_page(
self._XML_SEARCH_URL,
return json
def get_feed(self, feed):
+ """
+ @blocks
+ """
actualFeed = "_XML_%s_URL" % feed.upper()
feedUrl = getattr(self, actualFeed)
which can either be a ``Message`` instance, or a SHA1 identifier.
Saves files to ``adir`` (defaults to current directory).
Message hashes can be found in ``self.voicemail().messages`` for example.
- Returns location of saved file.
+ @returns location of saved file.
+ @blocks
"""
page = self._get_page(self._downloadVoicemailURL, {"id": messageId})
fn = os.path.join(adir, '%s.mp3' % messageId)
def get_recent(self):
"""
@returns Iterable of (personsName, phoneNumber, exact date, relative date, action)
+ @blocks
"""
- for action, url in (
- ("Received", self._XML_RECEIVED_URL),
- ("Missed", self._XML_MISSED_URL),
- ("Placed", self._XML_PLACED_URL),
- ):
- flatXml = self._get_page(url)
-
- allRecentHtml = self._grab_html(flatXml)
- allRecentData = self._parse_history(allRecentHtml)
- for recentCallData in allRecentData:
- recentCallData["action"] = action
- yield recentCallData
+ recentPages = [
+ (action, self._get_page(url))
+ for action, url in (
+ ("Received", self._XML_RECEIVED_URL),
+ ("Missed", self._XML_MISSED_URL),
+ ("Placed", self._XML_PLACED_URL),
+ )
+ ]
+ return self._parse_recent(recentPages)
def get_contacts(self):
"""
@returns Iterable of (contact id, contact name)
+ @blocks
"""
- page = self._get_page(self._XML_CONTACTS_URL)
- contactsBody = self._contactsBodyRe.search(page)
- if contactsBody is None:
- raise RuntimeError("Could not extract contact information")
- accountData = _fake_parse_json(contactsBody.group(1))
- for contactId, contactDetails in accountData["contacts"].iteritems():
- # A zero contact id is the catch all for unknown contacts
- if contactId != "0":
- yield contactId, contactDetails
+ page = self._get_page(self._JSON_CONTACTS_URL)
+ return self._process_contacts(page)
+
+ def get_csv_contacts(self):
+ data = {
+ "groupToExport": "mine",
+ "exportType": "ALL",
+ "out": "OUTLOOK_CSV",
+ }
+ encodedData = urllib.urlencode(data)
+ contacts = self._get_page(self._CSV_CONTACTS_URL+"?"+encodedData)
+ return contacts
def get_voicemails(self):
+ """
+ @blocks
+ """
voicemailPage = self._get_page(self._XML_VOICEMAIL_URL)
voicemailHtml = self._grab_html(voicemailPage)
voicemailJson = self._grab_json(voicemailPage)
+ if voicemailJson is None:
+ return ()
parsedVoicemail = self._parse_voicemail(voicemailHtml)
voicemails = self._merge_conversation_sources(parsedVoicemail, voicemailJson)
return voicemails
def get_texts(self):
+ """
+ @blocks
+ """
smsPage = self._get_page(self._XML_SMS_URL)
smsHtml = self._grab_html(smsPage)
smsJson = self._grab_json(smsPage)
+ if smsJson is None:
+ return ()
parsedSms = self._parse_sms(smsHtml)
smss = self._merge_conversation_sources(parsedSms, smsJson)
return smss
+ def get_unread_counts(self):
+ countPage = self._get_page(self._JSON_SMS_COUNT_URL)
+ counts = parse_json(countPage)
+ counts = counts["unreadCounts"]
+ return counts
+
def mark_message(self, messageId, asRead):
+ """
+ @blocks
+ """
postData = {
"read": 1 if asRead else 0,
"id": messageId,
markPage = self._get_page(self._markAsReadURL, postData)
def archive_message(self, messageId):
+ """
+ @blocks
+ """
postData = {
"id": messageId,
}
raise RuntimeError("Not Authenticated")
return number
+ def _parse_recent(self, recentPages):
+ for action, flatXml in recentPages:
+ allRecentHtml = self._grab_html(flatXml)
+ allRecentData = self._parse_history(allRecentHtml)
+ for recentCallData in allRecentData:
+ recentCallData["action"] = action
+ yield recentCallData
+
+ def _process_contacts(self, page):
+ accountData = parse_json(page)
+ for contactId, contactDetails in accountData["contacts"].iteritems():
+ # A zero contact id is the catch all for unknown contacts
+ if contactId != "0":
+ yield contactId, contactDetails
+
def _parse_history(self, historyHtml):
splitVoicemail = self._seperateVoicemailsRegex.split(historyHtml)
for messageId, messageHtml in itergroup(splitVoicemail[1:], 2):
yield {
"id": messageId.strip(),
"contactId": contactId,
- "name": name,
+ "name": unescape(name),
"time": exactTime,
"relTime": relativeTime,
"prettyNumber": prettyNumber,
"number": number,
- "location": location,
+ "location": unescape(location),
}
@staticmethod
relativeTimeGroup = self._relativeVoicemailTimeRegex.search(messageHtml)
conv.relTime = relativeTimeGroup.group(1).strip() if relativeTimeGroup else ""
locationGroup = self._voicemailLocationRegex.search(messageHtml)
- conv.location = locationGroup.group(1).strip() if locationGroup else ""
+ conv.location = unescape(locationGroup.group(1).strip() if locationGroup else "")
nameGroup = self._voicemailNameRegex.search(messageHtml)
- conv.name = nameGroup.group(1).strip() if nameGroup else ""
+ conv.name = unescape(nameGroup.group(1).strip() if nameGroup else "")
numberGroup = self._voicemailNumberRegex.search(messageHtml)
conv.number = numberGroup.group(1).strip() if numberGroup else ""
prettyNumberGroup = self._prettyVoicemailNumberRegex.search(messageHtml)
message = Message()
message.body = messageParts
message.whoFrom = conv.name
- message.when = conv.time.strftime("%I:%M %p")
+ try:
+ message.when = conv.time.strftime("%I:%M %p")
+ except ValueError:
+ _moduleLogger.exception("Confusing time provided: %r" % conv.time)
+ message.when = "Unknown"
conv.messages = (message, )
yield conv
conv.location = ""
nameGroup = self._voicemailNameRegex.search(messageHtml)
- conv.name = nameGroup.group(1).strip() if nameGroup else ""
+ conv.name = unescape(nameGroup.group(1).strip() if nameGroup else "")
numberGroup = self._voicemailNumberRegex.search(messageHtml)
conv.number = numberGroup.group(1).strip() if numberGroup else ""
prettyNumberGroup = self._prettyVoicemailNumberRegex.search(messageHtml)
def _parse_with_validation(self, page):
json = parse_json(page)
- validate_response(json)
+ self._validate_response(json)
return json
+ def _validate_response(self, response):
+ """
+ Validates that the JSON response is A-OK
+ """
+ try:
+ assert response is not None, "Response not provided"
+ assert 'ok' in response, "Response lacks status"
+ assert response['ok'], "Response not good"
+ except AssertionError:
+ try:
+ if response["data"]["code"] == 20:
+ raise RuntimeError(
+"""Ambiguous error 20 returned by Google Voice.
+Please verify you have configured your callback number (currently "%s"). If it is configured some other suspected causes are: non-verified callback numbers, and Gizmo5 callback numbers.""" % self._callbackNumber)
+ except KeyError:
+ pass
+ raise RuntimeError('There was a problem with GV: %s' % response)
+
+
+_UNESCAPE_ENTITIES = {
+ """: '"',
+ " ": " ",
+ "'": "'",
+}
+
+
+def unescape(text):
+ plain = saxutils.unescape(text, _UNESCAPE_ENTITIES)
+ return plain
+
def google_strptime(time):
"""
"""
abbrevTime = time[:-3]
parsedTime = datetime.datetime.strptime(abbrevTime, "%m/%d/%y %I:%M")
- if time[-2] == "PN":
+ if time.endswith("PM"):
parsedTime += datetime.timedelta(hours=12)
return parsedTime
def safe_eval(s):
_TRUE_REGEX = re.compile("true")
_FALSE_REGEX = re.compile("false")
+ _COMMENT_REGEX = re.compile("^\s+//.*$", re.M)
s = _TRUE_REGEX.sub("True", s)
s = _FALSE_REGEX.sub("False", s)
- return eval(s, {}, {})
+ s = _COMMENT_REGEX.sub("#", s)
+ try:
+ results = eval(s, {}, {})
+ except SyntaxError:
+ _moduleLogger.exception("Oops")
+ results = None
+ return results
def _fake_parse_json(flattened):
return jsonTree, flatHtml
-def validate_response(response):
- """
- Validates that the JSON response is A-OK
- """
- try:
- assert 'ok' in response and response['ok']
- except AssertionError:
- raise RuntimeError('There was a problem with GV: %s' % response)
-
-
def guess_phone_type(number):
if number.startswith("747") or number.startswith("1747") or number.startswith("+1747"):
return GVoiceBackend.PHONE_TYPE_GIZMO
("login", backend._loginURL),
("isdnd", backend._isDndURL),
("account", backend._XML_ACCOUNT_URL),
- ("contacts", backend._XML_CONTACTS_URL),
+ ("html_contacts", backend._XML_CONTACTS_URL),
+ ("contacts", backend._JSON_CONTACTS_URL),
+ ("csv", backend._CSV_CONTACTS_URL),
("voicemail", backend._XML_VOICEMAIL_URL),
- ("sms", backend._XML_SMS_URL),
+ ("html_sms", backend._XML_SMS_URL),
+ ("sms", backend._JSON_SMS_URL),
+ ("count", backend._JSON_SMS_COUNT_URL),
("recent", backend._XML_RECENT_URL),
("placed", backend._XML_PLACED_URL),
projects / theonering / blobdiff