http://posttopic.com/topic/google-voice-add-on-development
"""
+from __future__ import with_statement
import os
import re
return itertools.izip(*nIterators)
+class NetworkError(RuntimeError):
+ pass
+
+
class GVDialer(object):
"""
This class encapsulates all of the knowledge necessary to interact with the GoogleVoice servers
self._callbackNumber = ""
self._callbackNumbers = {}
+ _forwardURL = "https://www.google.com/voice/mobile/phones"
+
def is_authed(self, force = False):
"""
Attempts to detect a current session
self._lastAuthed = time.time()
return True
+ _tokenURL = "http://www.google.com/voice/m"
_loginURL = "https://www.google.com/accounts/ServiceLoginAuth"
+ _galxRe = re.compile(r"""<input.*?name="GALX".*?value="(.*?)".*?/>""", re.MULTILINE | re.DOTALL)
- def login(self, username, password):
- """
- Attempt to login to GoogleVoice
- @returns Whether login was successful or not
- """
+ def _get_token(self):
+ try:
+ tokenPage = self._browser.download(self._tokenURL)
+ except urllib2.URLError, e:
+ _moduleLogger.exception("Translating error: %s" % str(e))
+ raise NetworkError("%s is not accesible" % self._loginURL)
+ galxTokens = self._galxRe.search(tokenPage)
+ if galxTokens is not None:
+ galxToken = galxTokens.group(1)
+ else:
+ galxToken = ""
+ _moduleLogger.debug("Could not grab GALX token")
+ return galxToken
+
+ def _login(self, username, password, token):
loginPostData = urllib.urlencode({
'Email' : username,
'Passwd' : password,
"ltmpl": "mobile",
"btmpl": "mobile",
"PersistentCookie": "yes",
+ "GALX": token,
"continue": self._forwardURL,
})
try:
loginSuccessOrFailurePage = self._browser.download(self._loginURL, loginPostData)
except urllib2.URLError, e:
- _moduleLogger.exception(str(e))
- raise RuntimeError("%s is not accesible" % self._loginURL)
+ _moduleLogger.exception("Translating error: %s" % str(e))
+ raise NetworkError("%s is not accesible" % self._loginURL)
+ return loginSuccessOrFailurePage
+
+ def login(self, username, password):
+ """
+ Attempt to login to GoogleVoice
+ @returns Whether login was successful or not
+ """
+ self.logout()
+ galxToken = self._get_token()
+ loginSuccessOrFailurePage = self._login(username, password, galxToken)
try:
self._grab_account_info(loginSuccessOrFailurePage)
except Exception, e:
- _moduleLogger.exception(str(e))
- return False
+ # Retry in case the redirect failed
+ # luckily is_authed does everything we need for a retry
+ loggedIn = self.is_authed(True)
+ if not loggedIn:
+ _moduleLogger.exception(str(e))
+ return False
+ _moduleLogger.info("Redirection failed on initial login attempt, auto-corrected for this")
self._browser.cookies.save()
self._lastAuthed = time.time()
}
callSuccessPage = self._browser.download(self._clicktocallURL, clickToCallData, None, otherData)
except urllib2.URLError, e:
- _moduleLogger.exception(str(e))
- raise RuntimeError("%s is not accesible" % self._clicktocallURL)
+ _moduleLogger.exception("Translating error: %s" % str(e))
+ raise NetworkError("%s is not accesible" % self._clicktocallURL)
if self._gvDialingStrRe.search(callSuccessPage) is None:
raise RuntimeError("Google Voice returned an error")
}
smsSuccessPage = self._browser.download(self._sendSmsURL, smsData, None, otherData)
except urllib2.URLError, e:
- _moduleLogger.exception(str(e))
- raise RuntimeError("%s is not accesible" % self._sendSmsURL)
+ _moduleLogger.exception("Translating error: %s" % str(e))
+ raise NetworkError("%s is not accesible" % self._sendSmsURL)
return True
return {}
return self._callbackNumbers
- _setforwardURL = "https://www.google.com//voice/m/setphone"
-
def set_callback_number(self, callbacknumber):
"""
Set the number that GoogleVoice calls
try:
flatXml = self._browser.download(url)
except urllib2.URLError, e:
- _moduleLogger.exception(str(e))
- raise RuntimeError("%s is not accesible" % url)
+ _moduleLogger.exception("Translating error: %s" % str(e))
+ raise NetworkError("%s is not accesible" % url)
allRecentHtml = self._grab_html(flatXml)
allRecentData = self._parse_voicemail(allRecentHtml)
try:
contactsPage = self._browser.download(contactsPageUrl)
except urllib2.URLError, e:
- _moduleLogger.exception(str(e))
- raise RuntimeError("%s is not accesible" % contactsPageUrl)
+ _moduleLogger.exception("Translating error: %s" % str(e))
+ raise NetworkError("%s is not accesible" % contactsPageUrl)
for contact_match in self._contactsRe.finditer(contactsPage):
contactId = contact_match.group(1)
contactName = saxutils.unescape(contact_match.group(2))
try:
detailPage = self._browser.download(self._contactDetailURL + '/' + contactId)
except urllib2.URLError, e:
- _moduleLogger.exception(str(e))
- raise RuntimeError("%s is not accesible" % self._contactDetailURL)
+ _moduleLogger.exception("Translating error: %s" % str(e))
+ raise NetworkError("%s is not accesible" % self._contactDetailURL)
for detail_match in self._contactDetailPhoneRe.finditer(detailPage):
phoneNumber = detail_match.group(1)
_voicemailURL = "https://www.google.com/voice/inbox/recent/voicemail/"
_smsURL = "https://www.google.com/voice/inbox/recent/sms/"
+ @staticmethod
+ def _merge_messages(parsedMessages, json):
+ for message in parsedMessages:
+ id = message["id"]
+ jsonItem = json["messages"][id]
+ message["isRead"] = jsonItem["isRead"]
+ message["isSpam"] = jsonItem["isSpam"]
+ message["isTrash"] = jsonItem["isTrash"]
+ yield message
+
def get_messages(self):
try:
voicemailPage = self._browser.download(self._voicemailURL)
except urllib2.URLError, e:
- _moduleLogger.exception(str(e))
- raise RuntimeError("%s is not accesible" % self._voicemailURL)
+ _moduleLogger.exception("Translating error: %s" % str(e))
+ raise NetworkError("%s is not accesible" % self._voicemailURL)
voicemailHtml = self._grab_html(voicemailPage)
+ voicemailJson = self._grab_json(voicemailPage)
parsedVoicemail = self._parse_voicemail(voicemailHtml)
- decoratedVoicemails = self._decorate_voicemail(parsedVoicemail)
+ voicemails = self._merge_messages(parsedVoicemail, voicemailJson)
+ decoratedVoicemails = self._decorate_voicemail(voicemails)
try:
smsPage = self._browser.download(self._smsURL)
except urllib2.URLError, e:
- _moduleLogger.exception(str(e))
- raise RuntimeError("%s is not accesible" % self._smsURL)
+ _moduleLogger.exception("Translating error: %s" % str(e))
+ raise NetworkError("%s is not accesible" % self._smsURL)
smsHtml = self._grab_html(smsPage)
+ smsJson = self._grab_json(smsPage)
parsedSms = self._parse_sms(smsHtml)
- decoratedSms = self._decorate_sms(parsedSms)
+ smss = self._merge_messages(parsedSms, smsJson)
+ decoratedSms = self._decorate_sms(smss)
allMessages = itertools.chain(decoratedVoicemails, decoratedSms)
return allMessages
_tokenRe = re.compile(r"""<input.*?name="_rnr_se".*?value="(.*?)"\s*/>""")
_accountNumRe = re.compile(r"""<b class="ms\d">(.{14})</b></div>""")
_callbackRe = re.compile(r"""\s+(.*?):\s*(.*?)<br\s*/>\s*$""", re.M)
- _forwardURL = "https://www.google.com/voice/mobile/phones"
def _grab_account_info(self, page):
tokenGroup = self._tokenRe.search(page)
callbackNumber = match.group(2)
callbackName = match.group(1)
self._callbackNumbers[callbackNumber] = callbackName
+ if len(self._callbackNumbers) == 0:
+ _moduleLogger.debug("Could not extract callback numbers from GoogleVoice (the troublesome page follows):\n%s" % page)
def _send_validation(self, number):
if not self.is_valid_syntax(number):
"number": number,
"location": location,
"messageParts": messageParts,
+ "type": "Voicemail",
}
def _decorate_voicemail(self, parsedVoicemails):
"number": number,
"location": "",
"messageParts": messageParts,
+ "type": "Texts",
}
def _decorate_sms(self, parsedTexts):
"""
@returns (personsName, phoneNumber, date, action)
"""
+ contactId = recentCallData["contactId"]
if recentCallData["name"]:
header = recentCallData["name"]
elif recentCallData["prettyNumber"]:
number = recentCallData["number"]
relTime = recentCallData["relTime"]
action = recentCallData["action"]
- return header, number, relTime, action
+ return contactId, header, number, relTime, action
def decorate_message(messageData):
+ contactId = messageData["contactId"]
exactTime = messageData["time"]
if messageData["name"]:
header = messageData["name"]
for messagePart in messageParts
]
- decoratedResults = header, number, relativeTime, messages
+ decoratedResults = contactId, header, number, relativeTime, messages
return decoratedResults
print "Login?: ", backend.login(username, password)
print "Authenticated: ", backend.is_authed()
- #print "Token: ", backend._token
+ print "Token: ", backend._token
#print "Account: ", backend.get_account_number()
#print "Callback: ", backend.get_callback_number()
#print "All Callback: ",
- #import pprint
+ import pprint
#pprint.pprint(backend.get_callback_numbers())
#print "Recent: "
# print contact
# pprint.pprint(list(backend.get_contact_details(contact[0])))
- #print "Messages: ",
- #for message in backend.get_messages():
- # pprint.pprint(message)
+ print "Messages: ",
+ for message in backend.get_messages():
+ message["messageParts"] = list(message["messageParts"])
+ pprint.pprint(message)
#for message in sort_messages(backend.get_messages()):
# pprint.pprint(decorate_message(message))
return backend
+_TEST_WEBPAGES = [
+ ("forward", GVDialer._forwardURL),
+ ("token", GVDialer._tokenURL),
+ ("login", GVDialer._loginURL),
+ ("contacts", GVDialer._contactsURL),
+
+ ("voicemail", GVDialer._voicemailURL),
+ ("sms", GVDialer._smsURL),
+
+ ("recent", GVDialer._recentCallsURL),
+ ("placed", GVDialer._placedCallsURL),
+ ("recieved", GVDialer._receivedCallsURL),
+ ("missed", GVDialer._missedCallsURL),
+]
+
+
+def grab_debug_info(username, password):
+ cookieFile = os.path.join(".", "raw_cookies.txt")
+ try:
+ os.remove(cookieFile)
+ except OSError:
+ pass
+
+ backend = GVDialer(cookieFile)
+ browser = backend._browser
+
+ # Get Pages
+ print "Grabbing pre-login pages"
+ for name, url in _TEST_WEBPAGES:
+ try:
+ page = browser.download(url)
+ except StandardError, e:
+ print e.message
+ continue
+ print "\tWriting to file"
+ with open("not_loggedin_%s.txt" % name, "w") as f:
+ f.write(page)
+
+ # Login
+ print "Attempting login"
+ galxToken = backend._get_token()
+ loginSuccessOrFailurePage = backend._login(username, password, galxToken)
+ with open("loggingin.txt", "w") as f:
+ print "\tWriting to file"
+ f.write(loginSuccessOrFailurePage)
+ try:
+ backend._grab_account_info(loginSuccessOrFailurePage)
+ except Exception:
+ # Retry in case the redirect failed
+ # luckily is_authed does everything we need for a retry
+ loggedIn = backend.is_authed(True)
+ if not loggedIn:
+ raise
+
+ # Get Pages
+ print "Grabbing post-login pages"
+ for name, url in _TEST_WEBPAGES:
+ try:
+ page = browser.download(url)
+ except StandardError, e:
+ print e.message
+ continue
+ print "\tWriting to file"
+ with open("loggedin_%s.txt" % name, "w") as f:
+ f.write(page)
+
+ # Cookies
+ browser.cookies.save()
+ print "\tWriting cookies to file"
+ with open("cookies.txt", "w") as f:
+ f.writelines(
+ "%s: %s\n" % (c.name, c.value)
+ for c in browser.cookies
+ )
+
+
if __name__ == "__main__":
import sys
logging.basicConfig(level=logging.DEBUG)
- test_backend(sys.argv[1], sys.argv[2])
+ #test_backend(sys.argv[1], sys.argv[2])
+ grab_debug_info(sys.argv[1], sys.argv[2])