import re
import http.cookiejar
import getpass
import json
import logging
import sys
import time
import urllib.request, urllib.parse, urllib.error # for urlencode
import urllib.request, urllib.error, urllib.parse # for urlopen
import time
from optparse import OptionParser
logger = None
def log_http_request(request):
logger.info('> %s %s', request.get_method(), request.get_full_url())
for k, v in request.header_items():
logger.debug('> %s: %s', k, v)
logger.debug('>')
def log_http_response(response):
logger.info('< %s %s', response.getcode(), response.geturl())
for k, v in response.headers.items():
logger.debug('< %s: %s', k, v)
logger.debug('<')
def fetch_json_from_url(url, **kwargs):
request = urllib.request.Request(url)
request.add_header('Accept', 'application/json')
# build up the request
if len(kwargs) != 0:
request.data = urllib.parse.urlencode(kwargs).encode('utf-8')
with urllib.request.urlopen(request, timeout=30) as response:
# read and then decode the response
raw_data = response.read().decode(response.info().get_content_charset())
# log the response
log_http_request(request)
log_http_response(response)
# load the json data
return json.loads(raw_data)
def fetch_html_from_url(url, **kwargs):
request = urllib.request.Request(url)
#request.add_header('Accept-Encoding', 'gzip'),
# build up the request
if len(kwargs) != 0:
request.data = urllib.parse.urlencode(kwargs).encode('utf-8')
with urllib.request.urlopen(request, timeout=30) as response:
response.debuglevel = 10
# read and then decode the response
raw_data = response.read().decode(response.info().get_content_charset())
# log the response
log_http_request(request)
log_http_response(response);
return raw_data
def main():
optparser = OptionParser()
optparser.add_option("-u", "--username",
dest="username",
help="Garmin Connect Username")
optparser.add_option("-p", "--password",
dest="password",
help="Garmin Connect Password")
optparser.add_option("-f", "--filter",
dest="filters", action="append",
help="Additional query filter parameters as described in "
"connect.garmin.com/.../rest.activities.html")
# private, subscriber, groups, public
optparser.add_option("-s", "--private",
dest="privacy", action="store_const", const="private",
help="Make all matching activities private")
optparser.add_option("-c", "--connections",
dest="privacy", action="store_const", const="subscribers",
help="Make all matching activities visible to connections and yourself only")
optparser.add_option("-g", "--groups",
dest="privacy", action="store_const", const="groups",
help="Make all matching activities visible to connections, groups, and yourself only")
optparser.add_option("-w", "--public",
dest="privacy", action="store_const", const="public",
help="Make all matching activities public")
optparser.set_defaults(log_level=logging.WARNING)
optparser.add_option('-i', '--info',
dest='log_level', action='store_const', const=logging.INFO,
help='Log info events')
optparser.add_option('-d', '--debug',
dest='log_level', action='store_const', const=logging.DEBUG,
help='Log debug events')
optparser.add_option('-e', '--error',
dest='log_level', action='store_const', const=logging.ERROR,
help='Log only errors')
(options, args) = optparser.parse_args()
if not options.username:
options.username = input('Username: ')
if not options.password:
options.password = getpass.getpass('Password: ')
# configure the logger
logging.basicConfig(level=options.log_level)
global logger
logger = logging.getLogger()
# configure a url opener that uses cookies
opener = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar()))
urllib.request.install_opener(opener)
query_args = urllib.parse.urlencode({
'service': 'connect.garmin.com/.../login',
'clientId': 'GarminConnect',
'consumeServiceTicket': 'false'
})
url = '?'.join(('sso.garmin.com/.../login', query_args))
fetch_html_from_url(url)
kwargs = {
'username': options.username,
'password': options.password,
'embed' : 'true',
'_eventId': 'submit',
'lt' : 'e1s1'
}
response = fetch_html_from_url(url, **kwargs)
ticket = re.search("ticket=([^']+)'", response).groups(1)[0]
logger.debug('ticket=%s' % ticket)
query_args = urllib.parse.urlencode({
'ticket': ticket
})
url = '?'.join(('connect.garmin.com/.../login', query_args))
fetch_html_from_url(url)
if logger.isEnabledFor(logging.INFO):
json_data = fetch_json_from_url('connect.garmin.com/.../account')
userId = json_data['account']['userId']
logger.info('Authenticated %s [%s]' % (json_data['account']['email'], userId))
currentPage = 1
totalPages = 0
# convert list of filters into a string
if options.filters != None:
options.filters = '&'.join(options.filters)
while True:
url = 'connect.garmin.com/.../activities
# append the query filter options
if options.filters != None:
url = '&'.join((url, options.filters))
json_data = fetch_json_from_url(url)
# parse the json data
results = json_data['results']
if totalPages == 0:
totalPages = results['totalPages']
# list of matching activities, each of which is a map with one
# element named 'activity'
for activities in results['activities']:
activity = activities['activity']
activityId = activity['activityId']
activityName = activity['activityName']
activityPrivacy = activity['privacy']['key']
if options.privacy == None:
print('Activity {0} is currently {1}.'.format(activityId, activityPrivacy))
elif options.privacy != activityPrivacy:
# be nice to garmin's servers
time.sleep(.250)
fetch_json_from_url('connect.garmin.com/.../{0}'.format(activityId), value=options.privacy)
if currentPage != totalPages:
# be nice to garmin's servers
time.sleep(1)
currentPage = currentPage + 1
else:
break
if __name__ == "__main__":
main()