/usr/share/pytrainer/extensions/stravaupload/stravaupload.py is in pytrainer 1.11.0-1.
This file is owned by root:root, with mode 0o755.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 | #!/usr/bin/env python
import json
import urllib
import urllib2
import gtk
import logging
import webbrowser
from subprocess import Popen, PIPE
LOGIN_URL = "https://www.strava.com/api/v2/authentication/login"
UPLOAD_URL = "http://www.strava.com/api/v2/upload"
STATUS_URL = "http://www.strava.com/api/v2/upload/status/%s"
ACTIVITY_URL = "http://app.strava.com/activities/"
class ConfigError(Exception): pass
class ProgressDialog():
def __init__(self, text):
self.progress = None
self.text = text
def __enter__(self):
self.progress = Popen(["zenity", "--text", "Strava Upload", "--percentage=0", "--auto-close=True", "--pulsate=True", "--progress", "--no-cancel"], stdin=PIPE)
self.progress.stdin.write('# %s\n' % self.text)
return self
def __exit__(self, type, value, traceback):
self.progress.stdin.write('100\n')
self.progress.stdin.close()
self.progress.wait()
class StravaUpload:
def __init__(self, parent = None, pytrainer_main = None, conf_dir = None, options = None):
self.pytrainer_main = pytrainer_main
self.conf_dir = conf_dir
self.strava_token = "%s/.strava_token" % self.conf_dir
self.strava_uploads = "%s/.strava_uploads" % self.conf_dir
self.email = None
self.password = None
if options:
self.email = options['stravauploademail']
self.password = options['stravauploadpassword']
def __enter__(self):
return self
def __exit__(self, type, value, traceback):
pass
def get_web_data(self, url, values, text):
with ProgressDialog(text) as p:
data = urllib.urlencode(values)
req = urllib2.Request(url, data)
response = urllib2.urlopen(req)
return json.loads(response.read())
def login_token(self):
token = None
try:
with open(self.strava_token) as f:
token = f.readline()
except:
pass
if token is None or token.strip() == '' :
if self.email and self.password:
values = { 'email' : self.email, 'password' : self.password }
result = self.get_web_data(LOGIN_URL, values, "Validating user...")
token = result['token']
else:
raise ConfigError, "Username or password missing"
try:
with open(self.strava_token, 'w') as f:
f.write(token)
except:
# didn't write token but that's ok, get another next time...
pass
return token
def find_upload(self, id):
upload_id = None
try:
with open(self.strava_uploads) as f:
for line in f:
upload = line.strip().split(',')
if upload[0] == str(id):
upload_id = upload[1]
break
except IOError, e:
logging.debug("Failed to read uploads file: %s" % e)
return upload_id
def store_upload_id(self, id, upload_id):
try:
with open(self.strava_uploads, 'a') as f:
f.write('%s,%s\n' % (id, upload_id))
except IOError, e:
# log failure but continue...
logging.debug("Failed to write upload id: %s" % e)
def get_status(self, token, id):
status = None
values = { 'token': token }
result = self.get_web_data(STATUS_URL % id, values, "Getting status...")
if result.has_key('activity_id'):
status = result['activity_id']
return status
def upload(self, token, gpx_file):
gpx = None
upload_id = 0
try:
with open(gpx_file) as f:
gpx = f.read()
except:
pass
if gpx is not None and gpx.strip() != '':
values = { 'token': token, 'type': 'gpx', 'data': gpx }
result = self.get_web_data(UPLOAD_URL, values, "Uploading record...")
upload_id = result['upload_id']
return upload_id
def run(self, id, activity = None):
logging.debug(">>")
log = "Strava Upload: "
gpx_file = "%s/gpx/%s.gpx" % (self.conf_dir, id)
try:
logging.debug("Getting user token")
user_token = self.login_token();
if user_token is not None:
existing_id = self.find_upload(id)
if existing_id is None:
logging.debug("Uploading GPX: %s" % gpx_file)
upload_id = self.upload(user_token, gpx_file)
if upload_id > 0:
self.store_upload_id(id, upload_id)
log = log + "success (upload: %s)!" % upload_id
else:
log = log + "failed to upload!"
else:
# already uploaded - get status
log = "Record already uploaded."
status = self.get_status(user_token, existing_id)
if status:
url = '%s%s' % (ACTIVITY_URL, status)
logging.debug(url)
log = log + "\n\n%s" % url
webbrowser.open(url)
else:
# don't know status yet, assume it is processing
log = log + " Processing... check again soon!"
except (ValueError, KeyError), e:
log = log + ("JSON error: %s." % e)
except ConfigError, e:
log = log + ("config error: %s." % e)
except Exception, e:
log = "Unknown exception: %s." % e
logging.debug(log)
md = gtk.MessageDialog(self.pytrainer_main.windowmain.window1, gtk.DIALOG_DESTROY_WITH_PARENT, gtk.MESSAGE_INFO, gtk.BUTTONS_OK, log)
md.set_title(_("Strava Upload"))
md.set_modal(False)
md.run()
md.destroy()
logging.debug("<<")
|