piezza/garminexport
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

adapted to new authentication procedure

Browse Source
This commit is contained in:
petergardfjall 2017-05-14 08:54:23 +02:00
parent ef9b875465
commit aaed3f44bb
2 changed files with 77 additions and 128 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

28
garminbackup.py
View File

@ -59,7 +59,7 @@ if __name__ == "__main__":
parser.add_argument(
"-E", "--ignore-errors", action='store_true',
help="Ignore errors and keep going. Default: FALSE")
args = parser.parse_args()
if not args.log_level in LOG_LEVELS:
raise ValueError("Illegal log-level: {}".format(args.log_level))
@ -67,32 +67,31 @@ if __name__ == "__main__":
# if no --format was specified, all formats are to be backed up
args.format = args.format if args.format else export_formats
log.info("backing up formats: %s", ", ".join(args.format))
logging.root.setLevel(LOG_LEVELS[args.log_level])
try:
if not os.path.isdir(args.backup_dir):
os.makedirs(args.backup_dir)
if not args.password:
args.password = getpass.getpass("Enter password: ")
with GarminClient(args.username, args.password) as client:
# get all activity ids and timestamps from Garmin account
log.info("retrieving activities for {} ...".format(args.username))
log.info("scanning activities for %s ...", args.username)
activities = set(client.list_activities())
log.info("account has a total of {} activities.".format(
len(activities)))
log.info("account has a total of %d activities", len(activities))
missing_activities = garminexport.backup.need_backup(
activities, args.backup_dir, args.format)
backed_up = activities - missing_activities
log.info("{} contains {} backed up activities.".format(
args.backup_dir, len(backed_up)))
log.info("%s contains %d backed up activities",
args.backup_dir, len(backed_up))
log.info("activities that aren't backed up: %d",
len(missing_activities))
log.info("activities that aren't backed up: {}".format(
len(missing_activities)))
for index, activity in enumerate(missing_activities):
id, start = activity
log.info("backing up activity %d from %s (%d out of %d) ..." %
@ -106,5 +105,4 @@ if __name__ == "__main__":
raise
except Exception as e:
exc_type, exc_value, exc_traceback = sys.exc_info()
log.error(u"failed with exception: %s", e)
raise
log.error(u"failed with exception: %s", str(e))

177
garminexport/garminclient.py
View File

@ -49,10 +49,10 @@ def require_session(client_function):
client_object = args[0]
if not client_object.session:
raise Exception("Attempt to use GarminClient without being connected. Call connect() before first use.'")
return client_function(*args, **kwargs)
return client_function(*args, **kwargs)
return check_session
class GarminClient(object):
"""A client class used to authenticate with Garmin Connect and
extract data from the user account.
@ -62,13 +62,13 @@ class GarminClient(object):
automatically take care of logging in to Garmin Connect before any
further interactions and logging out after the block completes or
a failure occurs.
Example of use: ::
with GarminClient("my.sample@sample.com", "secretpassword") as client:
ids = client.list_activity_ids()
for activity_id in ids:
gpx = client.get_activity_gpx(activity_id)
"""
def __init__(self, username, password):
@ -86,119 +86,70 @@ class GarminClient(object):
def __enter__(self):
self.connect()
return self
def __exit__(self, exc_type, exc_value, traceback):
self.disconnect()
def connect(self):
self.session = requests.Session()
self._authenticate()
def disconnect(self):
if self.session:
self.session.close()
self.session = None
def _authenticate(self):
log.info("authenticating user ...")
params = {
"service": "http://connect.garmin.com/post-auth/login",
"clientId": "GarminConnect",
"consumeServiceTicket": "false",
"gauthHost": "https://sso.garmin.com/sso"
}
flow_execution_key = self._get_flow_execution_key(params)
log.debug("flow execution key: '{}'".format(flow_execution_key))
validation_url = self._get_auth_ticket(flow_execution_key, params)
# recently, the validation url has started to escape slash characters
# (with a backslash). remove any such occurences.
validation_url = validation_url.replace("\/", "/")
log.debug("auth ticket validation url: {}".format(validation_url))
self._validate_auth_ticket(validation_url)
form_data = {
"username": self.username,
"password": self.password,
"embed": "false"
}
request_params = {
"service": "https://connect.garmin.com/modern"
}
auth_response = self.session.post(
SSO_LOGIN_URL, params=request_params, data=form_data)
log.debug("got auth response: %s", auth_response.text)
if auth_response.status_code != 200:
raise ValueError(
"authentication failure: did you enter valid credentials?")
auth_ticket_url = self._extract_auth_ticket_url(
auth_response.text)
log.debug("auth ticket url: '%s'", auth_ticket_url)
# Referer seems to be a header that is required by the REST API
self.session.headers.update({'Referer': "https://some.random.site"})
def _get_flow_execution_key(self, request_params):
# The flowExecutionKey is embedded in the
# https://sso.garmin.com/sso/login response page. For example:
# <!-- flowExecutionKey: [e3s1] -->
log.debug("get flow execution key ...")
response = self.session.get(SSO_LOGIN_URL, params=request_params)
log.info("claiming auth ticket ...")
response = self.session.get(auth_ticket_url)
if response.status_code != 200:
raise RuntimeError(
"auth failure: %s: code %d: %s" %
(SSO_LOGIN_URL, response.status_code, response.text))
# extract flowExecutionKey
match = re.search(r'<!-- flowExecutionKey: \[(\w+)\]', response.text)
"auth failure: failed to claim auth ticket: %s: %d\n%s" %
(auth_ticket_url, response.status_code, response.text))
# appears like we need to touch base with the old API to initiate
# some form of legacy session. otherwise certain downloads will fail.
self.session.get('https://connect.garmin.com/legacy/session')
def _extract_auth_ticket_url(self, auth_response):
"""Extracts an authentication ticket URL from the response of an
authentication form submission. The auth ticket URL is typically
of form:
https://connect.garmin.com/modern?ticket=ST-0123456-aBCDefgh1iJkLmN5opQ9R-cas
:param auth_response: HTML response from an auth form submission.
"""
match = re.search(
r'response_url\s*=\s*"(https:[^"]+)"', auth_response)
if not match:
raise RuntimeError(
"auth failure: unable to extract flowExecutionKey: %s:\n%s" %
(SSO_LOGIN_URL, response.text))
flow_execution_key = match.group(1)
return flow_execution_key
"auth failure: unable to extract auth ticket URL. did you provide a correct username/password?")
auth_ticket_url = match.group(1).replace("\\", "")
return auth_ticket_url
def _get_auth_ticket(self, flow_execution_key, request_params):
data = {
"username": self.username, "password": self.password,
"_eventId": "submit", "embed": "true", "lt": flow_execution_key
}
log.debug("single sign-on ...")
sso_response = self.session.post(
SSO_LOGIN_URL, params=request_params,
data=data, allow_redirects=False)
# response must contain an SSO ticket
ticket_match = re.search("ticket=([^']+)'", sso_response.text)
if not ticket_match:
raise ValueError("auth failure: unable to get auth ticket: "
"did you enter valid credentials?")
ticket = ticket_match.group(1)
log.debug("SSO ticket: {}".format(ticket))
# response should contain a URL where auth ticket can be validated
validation_url = re.search(
r"response_url\s+=\s+'([^']+)'", sso_response.text)
validation_url = validation_url.group(1)
return validation_url
def _validate_auth_ticket(self, validation_url):
log.debug("validating auth ticket at %s ...", validation_url)
response = self.session.get(validation_url, allow_redirects=False)
# It appears as if from this point on, the User-Agent header needs to
# be set to something similar to the value below for authentication
# to succeed and for downloads of .fit files to work properly.
self.session.headers.update({
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/49.0.2623.112 Safari/537.36'
})
# we will be redirected several times. just follow through ..
while response.status_code == 302:
redirect_url = response.headers['Location']
log.debug("redirected to: '%s'", redirect_url)
response = self.session.get(redirect_url, allow_redirects=False)
if response.status_code != 200:
raise Exception(
u"auth failure: unable to validate auth ticket: {}:\n{}".format(
response.status_code, response.text))
# auth ticket successfully validated.
# our client should now have all necessary cookies set.
# as a final step in the "Garmin login rain dance", it appears
# as though we need to touch on their legacy session page before
# being granted access to some api services (such as the
# activity-search-service).
self.session.get('https://connect.garmin.com/legacy/session')
return
@require_session
def list_activities(self):
"""Return all activity ids stored by the logged in user, along
@ -206,7 +157,7 @@ class GarminClient(object):
:returns: The full list of activity identifiers.
:rtype: tuples of (int, datetime)
"""
"""
ids = []
batch_size = 100
# fetch in batches since the API doesn't allow more than a certain
@ -225,12 +176,12 @@ class GarminClient(object):
Should the index be out of bounds or the account empty, an empty
list is returned.
:param start_index: The index of the first activity to retrieve.
:type start_index: int
:param max_limit: The (maximum) number of activities to retrieve.
:param max_limit: The (maximum) number of activities to retrieve.
:type max_limit: int
:returns: A list of activity identifiers.
:rtype: list of str
"""
@ -253,8 +204,8 @@ class GarminClient(object):
for entry in results["activities"] ]
log.debug("got {} activities.".format(len(entries)))
return entries
@require_session
@require_session
def get_activity_summary(self, activity_id):
"""Return a summary about a given activity. The
summary contains several statistics, such as duration, GPS starting
@ -287,10 +238,10 @@ class GarminClient(object):
response = self.session.get("https://connect.garmin.com/modern/proxy/activity-service-1.3/json/activityDetails/{}".format(activity_id))
if response.status_code != 200:
raise Exception(u"failed to fetch json activityDetails for {}: {}\n{}".format(
activity_id, response.status_code, response.text))
activity_id, response.status_code, response.text))
return json.loads(response.text)
@require_session
@require_session
def get_activity_gpx(self, activity_id):
"""Return a GPX (GPS Exchange Format) representation of a
given activity. If the activity cannot be exported to GPX
@ -308,7 +259,7 @@ class GarminClient(object):
# and is the one used when exporting through the Garmin
# Connect web page.
#response = self.session.get("https://connect.garmin.com/proxy/activity-service-1.1/gpx/activity/{}?full=true".format(activity_id))
# A 404 (Not Found) or 204 (No Content) response are both indicators
# of a gpx file not being available for the activity. It may, for
# example be a manually entered activity without any device data.
@ -316,7 +267,7 @@ class GarminClient(object):
return None
if response.status_code != 200:
raise Exception(u"failed to fetch GPX for activity {}: {}\n{}".format(
activity_id, response.status_code, response.text))
activity_id, response.status_code, response.text))
return response.text
@ -334,13 +285,13 @@ class GarminClient(object):
or ``None`` if the activity cannot be exported to TCX.
:rtype: str
"""
response = self.session.get("https://connect.garmin.com/modern/proxy/download-service/export/tcx/activity/{}".format(activity_id))
if response.status_code == 404:
return None
if response.status_code != 200:
raise Exception(u"failed to fetch TCX for activity {}: {}\n{}".format(
activity_id, response.status_code, response.text))
activity_id, response.status_code, response.text))
return response.text
@ -372,9 +323,9 @@ class GarminClient(object):
fn, ext = os.path.splitext(path)
if fn==str(activity_id):
return ext[1:], zip.open(path).read()
return (None,None)
return (None,None)
def get_activity_fit(self, activity_id):
"""Return a FIT representation for a given activity. If the activity
doesn't have a FIT source (for example, if it was entered manually