from rowsandall_app.settings import ( NK_OAUTH_LOCATION, ROJABO_OAUTH_LOCATION, TP_OAUTH_LOCATION, ) from rowers.views.statements import * from rowers.plannedsessions import get_dates_timeperiod from rowers.tasks import fetch_strava_workout from rowers.utils import NoTokenError from rowers.models import PlannedSession import rowers.integrations.strava as strava import rowers.integrations.intervals as intervals from rowers.integrations import importsources from rowers.integrations import IntervalsIntegration from rowers.utils import NoTokenError import numpy importauthorizeviews = { 'c2': 'rower_integration_authorize', 'strava': 'rower_integration_authorize', 'polar': 'rower_polar_authorize', 'ownapi': 'workout_view', 'sporttracks': 'rower_integration_authorize', 'trainingpeaks': 'rower_integration_authorize', 'nk': 'rower_integration_authorize', 'rp3': 'rower_integration_authorize', 'intervals': 'rower_integration_authorize', } def default(o): # pragma: no cover if isinstance(o, numpy.int64): return int(o) raise TypeError @login_required() def workout_export_view(request, id=0, source='c2'): r = getrequestrower(request) w = get_workout_by_opaqueid(request, id) if w.user != request.user.rower: messages.error(request, 'You can only export your own workouts') url = reverse('workouts_view') return HttpResponseRedirect(url) integration = importsources[source](r.user) try: id = integration.workout_export(w) except NoTokenError: # pragma: no cover return HttpResponseRedirect(integration.make_authorization_url()) messages.info( request, "Your workout will be synchronized to {name} in the background".format( name=integration.get_name() ) ) url = reverse(r.defaultlandingpage, kwargs={ 'id': encoder.encode_hex(w.id) }) return HttpResponseRedirect(url) # ROJABO authorization def rower_rojabo_authorize(request): # pragma: no cover state = str(uuid4()) scope = "read" params = { "client_id": ROJABO_CLIENT_ID, "redirect_uri": ROJABO_REDIRECT_URI, } url = ROJABO_OAUTH_LOCATION+'oauth/authorize?'+urllib.parse.urlencode(params) return HttpResponseRedirect(url) @login_required() def rower_integration_authorize(request, source='c2'): # pragma: no cover try: integration = importsources[source](request.user) except KeyError: if source == 'rojabo': return rower_rojabo_authorize(request) if source == 'polar': return rower_polar_authorize(request) url = integration.make_authorization_url() return HttpResponseRedirect(url) # Polar Authorization @login_required() def rower_polar_authorize(request, source='polar'): # pragma: no cover integration = importsources['polar'](request.user) url = integration.make_authorization_url() return HttpResponseRedirect(url) @login_required() def rower_integration_token_refresh(request, source='c2'): try: integration = importsources[source](request.user) except KeyError: # pragma: no cover url = reverse('workouts_view') return HttpResponseRedirect(url) try: token = integration.token_refresh() messages.info(request, "Tokens refreshed. Good to go") except NoTokenError: messages.error(request, "Something went wrong refreshing your access tokens to {name}. Please reauthorize".format( name=integration.get_name() )) url = reverse('workouts_view') return HttpResponseRedirect(url) # Concept2 Callback @login_required() def rower_process_callback(request): c2_integration = importsources['c2'](request.user) try: code = request.GET['code'] res = c2_integration.get_token(code) except (MultiValueDictKeyError, NoTokenError): # pragma: no cover message = "The resource owner or authorization server denied the request" messages.error(request, message) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) access_token = res[0] if access_token == 0: # pragma: no cover message = res[1] message += ' Contact info@rowsandall.com if this behavior persists.' messages.error(request, message) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) expires_in = res[1] refresh_token = res[2] expirydatetime = timezone.now()+datetime.timedelta(seconds=expires_in) r = getrower(request.user) r.c2token = access_token r.tokenexpirydate = expirydatetime r.c2refreshtoken = refresh_token r.save() successmessage = "Tokens stored. Good to go. Please check your import/export settings" messages.info(request, successmessage) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) # dummy @login_required() def rower_process_twittercallback(request): # pragma: no cover return "dummy" # Process Polar Callback @login_required() def rower_process_intervalscallback(request): integration = importsources['intervals'](request.user) r = getrower(request.user) try: code = request.GET['code'] res = integration.get_token(code) except MultiValueDictKeyError: message = "The resource owner or authorization server denied the request" messages.error(request, message) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) access_token = res[0] athlete = res[1] if access_token == 0: message = res[1] message += 'Connection to intervals.icu failed.' messages.error(request, message) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) r.intervals_token = access_token r.intervals_owner_id = athlete['id'] r.save() successmessage = "Tokens stored. Good to go. Please check your import/export settings" messages.info(request, successmessage) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) @login_required() def rower_process_polarcallback(request): integration = importsources['polar'](request.user) error = request.GET.get('error', 'no error') dologging('polar.log', 'Callback: {error}'.format(error=error)) if error != 'no error': # pragma: no cover messages.error(request, error) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) try: code = request.GET['code'] dologging('polar.log', code) except MultiValueDictKeyError: # pragma: no cover try: message = request.GET['error'] except MultiValueDictKeyError: message = "access error" messages.error(request, message) url = reverse('workouts_view') return HttpResponseRedirect(url) access_token, expires_in, user_id = integration.get_token(code) expirydatetime = timezone.now()+datetime.timedelta(seconds=expires_in) r = getrower(request.user) r.polartoken = access_token r.polartokenexpirydate = expirydatetime r.polaruserid = user_id r.save() if user_id: polar_user_data = integration.register_user(access_token) else: # pragma: no cover messages.error(request, 'Polar Flow Authorization Failed') url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) try: error = polar_user_data['error'] if error['error_type'] == 'user_already_registered': # pragma: no cover s = error['message'] tester = re.compile(r'.*userid:(?P\d+)') testresult = tester.match(s) if testresult: user_id2 = testresult.group('id') if user_id2 == str(user_id): messages.info(request, "Tokens stored. Good to go. Please check your import/export settings") url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) except KeyError: pass try: user_id2 = polar_user_data['polar-user-id'] except KeyError: # pragma: no cover user_id2 = 0 if user_id2 != user_id: # pragma: no cover messages.error(request, 'Polar User ID error') if user_id2 == user_id: successmessage = "Tokens stored. Good to go. Please check your import/export settings" messages.info(request, successmessage) else: # pragma: no cover messages.error( request, "Please contact support@rowsandall.com for help.") url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) # Process Rojabo callback @login_required() def rower_process_rojabocallback(request): # pragma: no cover # do stuff try: code = request.GET.get('code', None) res = rojabo_stuff.get_token(code) except MultiValueDictKeyError: message = "The resource owner or authorization server denied the request" messages.error(request, message) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) access_token = res[0] if access_token == 0: message = res[1] message += ' Contact support@rowsandall.com if this behavior persists' messages.error(request, message) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) expires_in = res[1] refresh_token = res[2] expirydatetime = timezone.now()+datetime.timedelta(seconds=expires_in) r = getrower(request.user) r.rojabo_token = access_token r.rojabo_tokenexpirydate = expirydatetime r.rojabo_refreshtoken = refresh_token r.save() successmessage = "Tokens stored. Good to go. Please check your import/export settings" messages.info(request, successmessage) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) # Process NK Callback @login_required() def rower_process_nkcallback(request): # pragma: no cover # do stuff try: code = request.GET.get('code', None) nk_integration = importsources['nk'](request.user) access_token, expires_in, refresh_token = nk_integration.get_token(code) except MultiValueDictKeyError: message = "The resource owner or authorization server denied the request" messages.error(request, message) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) except NoTokenError: message = "Could not fetch token. Please reconnect with NK Logbook" messages.error(request, message) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) if access_token == 0: message = res[1] message += ' Contact support@rowsandall.com if this behavior persists' messages.error(request, message) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) # nk_owner_id = res[3] expirydatetime = timezone.now()+datetime.timedelta(seconds=expires_in) r = getrower(request.user) r.nktoken = access_token r.nktokenexpirydate = expirydatetime r.nkrefreshtoken = refresh_token # r.nk_owner_id = nk_owner_id r.save() successmessage = "Tokens stored. Good to go. Please check your import/export settings" messages.info(request, successmessage) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) @login_required() def workout_import_view(request, source='c2'): startdate, enddate = get_dates_timeperiod( request, defaulttimeperiod='last30') startdate = startdate.date() enddate = enddate.date() r = getrequestrower(request) try: integration = importsources[source](request.user) except KeyError: # pragma: no cover messages.error(request,"This integration is not supported") url = reverse('workouts_view') return HttpResponseRedirect(url) if request.method == 'GET': type_filter = request.GET.get('type', None) else: type_filter = None try: _ = integration.open() except NoTokenError: # pragma: no cover try: theview = importauthorizeviews[source] if theview == 'rower_integration_authorize': url = reverse(theview,kwargs={'source':source}) else: url = reverse(theview) return HttpResponseRedirect(url) except KeyError: messages.error(request,'Sorry, an error occurred. Please reauthorize') url = reverse('rower_export_settings_view') return HttpResponseRedirect(url) if request.method == 'POST': # pragma: no cover dateform = DateRangeForm(request.POST) if dateform.is_valid(): startdate = dateform.cleaned_data['startdate'] enddate = dateform.cleaned_data['enddate'] + \ datetime.timedelta(days=1) else: dateform = DateRangeForm(initial={ 'startdate': startdate, 'enddate': enddate, }) if enddate < startdate: # pragma: no cover s = enddate enddate = startdate startdate = s startdatestring = startdate.strftime('%Y-%m-%d') enddatestring = enddate.strftime('%Y-%m-%d') request.session['startdate'] = startdatestring request.session['enddate'] = enddatestring before = arrow.get(enddate) before = str(int(before.timestamp()*1000)) after = arrow.get(startdate) after = str(int(after.timestamp()*1000)) try: workouts = integration.get_workout_list( before=before, after=after, startdate=startdate, enddate=enddate, type_filter=type_filter ) except NoTokenError: # pragma: no cover messages.error(request,"You must first make the connection to {source}".format( source=source )) url = reverse(importauthorizeviews[source],kwargs={'source':source}) return HttpResponseRedirect(url) if request.method == 'POST': # pragma: no cover try: tdict = dict(request.POST.lists()) ids = tdict['workoutid'] try: nkids = [int(id) for id in ids] except ValueError: nkids = ids for nkid in nkids: try: _ = integration.get_workout(nkid, startdate=startdate, enddate=enddate) except NoTokenError: pass messages.info( request, 'Your {source} workouts will be imported in the background.' ' It may take a few minutes before they appear.'.format(source=integration.get_name())) url = reverse('workouts_view') return HttpResponseRedirect(url) except KeyError: pass breadcrumbs = [ { 'url': '/rowers/list-workouts/', 'name': 'Workouts' }, { 'url': reverse('workout_import_view',kwargs={'source':source}), 'name': integration.get_name() }, ] checknew = request.GET.get('selectallnew', False) return render(request, 'list_import.html', { 'workouts': workouts, 'rower': r, 'dateform':dateform, 'active': 'nav-workouts', 'breadcrumbs': breadcrumbs, 'teams': get_my_teams(request.user), 'checknew': checknew, 'startdate':startdate, 'enddate': enddate, 'integration': integration.get_name() }) @login_required() def rower_process_stravacallback(request): strava_integration = importsources['strava'](request.user) try: code = request.GET['code'] _ = request.GET['scope'] except MultiValueDictKeyError: # pragma: no cover try: message = request.GET['error'] except MultiValueDictKeyError: # pragma: no cover message = "access error" messages.error(request, message) url = reverse('workouts_view') return HttpResponseRedirect(url) try: res = strava_integration.get_token(code,logfile='strava_log.log') except NoTokenError: # pragma: no cover message = "Something went wrong with the Strava authorization" messages.error(request, message) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) if res[0]: access_token = res[0] expires_in = res[1] refresh_token = res[2] expirydatetime = timezone.now()+datetime.timedelta(seconds=expires_in) r = getrower(request.user) r.stravatoken = access_token r.stravatokenexpirydate = expirydatetime r.stravarefreshtoken = refresh_token r.save() try: _ = strava_integration.set_strava_athlete_id() except NoTokenError: # pragma: no cover messages.error(request,'Something went wrong storing your Strava ID. Please authorize again') url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) successmessage = "Tokens stored. Good to go. Please check your import/export settings" messages.info(request, successmessage) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) else: # pragma: no cover message = "Something went wrong with the Strava authorization" messages.error(request, message) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) # pragma: no cover # Process SportTracks callback @login_required() def rower_process_sporttrackscallback(request): try: code = request.GET['code'] except: # pragma: no cover messages.error(request, "Sorry, something went wrong.") url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) st_integration = importsources['sporttracks'](request.user) token = st_integration.get_token(code) successmessage = "Tokens stored. Good to go. Please check your import/export settings" messages.info(request, successmessage) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) # Process RP3 callback @login_required() def rower_process_rp3callback(request): # pragma: no cover try: code = request.GET['code'] except MultiValueDictKeyError: messages.error(request, "There was an error with the callback") try: errormessage = request.GET['error'] messages.error(request, errormessage) except MultiValueDictKeyError: pass url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) rp3_integration = importsources['rp3'](request.user) access_token, expires_in, refresh_token = rp3_integration.get_token(code) expirydatetime = timezone.now()+datetime.timedelta(seconds=expires_in) r = getrower(request.user) r.rp3token = access_token r.rp3tokenexpirydate = expirydatetime r.rp3refreshtoken = refresh_token r.save() successmessage = "Tokens stored. Good to go. Please check your import/export settings" messages.info(request, successmessage) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) # Process TrainingPeaks callback @login_required() def rower_process_tpcallback(request): try: code = request.GET['code'] except MultiValueDictKeyError: # pragma: no cover messages.error(request, "There was an error with the callback") try: errormessage = request.GET['error'] messages.error(request, errormessage) except MultiValueDictKeyError: pass url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) tp_integration = importsources['trainingpeaks'](request.user) access_token, expires_in, refresh_token = tp_integration.get_token(code) expirydatetime = timezone.now()+datetime.timedelta(seconds=expires_in) r = getrower(request.user) r.tptoken = access_token r.tptokenexpirydate = expirydatetime r.tprefreshtoken = refresh_token r.save() successmessage = "Tokens stored. Good to go. Please check your import/export settings" messages.info(request, successmessage) url = reverse('rower_exportsettings_view') return HttpResponseRedirect(url) # Process Own API callback - for API testing purposes @login_required() def rower_process_testcallback(request): # pragma: no cover code = request.GET['code'] res = ownapistuff.get_token(code) access_token = res[0] refresh_token = res[2] text = "Access Token:\n" text += access_token text += "\n\nRefresh Token:\n" text += refresh_token return HttpResponse(text) # view to list planned sessions from intervals.icu @login_required() @user_passes_test(isplanmember, login_url="/rowers/paidplans/", message="This functionality requires a Self-coach plan or higher", redirect_field_name=None) def plannedsession_intervalsimport_view(request, message="", userid=0): r = getrequestrower(request, userid=userid) if r.user != request.user: messages.error( request, 'You can only access your own workouts on Intervals.icu, not those of your athletes') url = reverse('plannedsession_intervalsimport_view', kwargs={'userid': request.user.id}) return HttpResponseRedirect(url) integration = importsources['intervals'](request.user) sessions_list = integration.get_plannedsessions_list() if request.method == 'POST': # pragma: no cover tdict = dict(request.POST.lists()) sessionids = [id for id in tdict['session']] for sessionid in sessionids: sessiondata = integration.get_plannedsession(sessionid) if sessiondata['description'] is None: sessiondata['description'] = '' if sessiondata: timetarget = sessiondata['time_target'] if timetarget is None: timetarget = sessiondata['moving_time'] if timetarget is None: timetarget = 3600 timetarget = int(timetarget)/60. ps = PlannedSession( name=sessiondata['name'], comment=sessiondata['description'], sessionmode='time', sessionvalue=timetarget, startdate=arrow.get(sessiondata['start_date_local']).datetime, enddate=arrow.get(sessiondata['end_date_local']).datetime, preferreddate=arrow.get(sessiondata['start_date_local']).datetime, sessionsport=mytypes.intervalsmappinginv[sessiondata['type']], sessiontype='session', intervals_icu_id=sessiondata['id'], manager=request.user, ) ps.save() ps.rower.add(r) if sessiondata['category'].lower() == 'workout': ps.fitfile = sessiondata['fitfile'] ps.save() ps.update_steps() if sessiondata['category'].lower() == 'target': ps.sessiontype = 'cycletarget' ps.sessionvalue = int(sessiondata['time_target'])/60. ps.enddate = ps.startdate + datetime.timedelta(days=6) ps.save() url = reverse('plannedsessions_view') return HttpResponseRedirect(url) return render(request, 'intervals_list_import.html', { 'sessions': sessions_list, 'rower': r, 'active': 'nav-plans', }) @login_required() @user_passes_test(ispromember, login_url="/rowers/paidplans/", message="This functionality requires a Pro plan or higher", redirect_field_name=None) @permission_required('plannedsession.add_session', fn=get_user_by_userid, raise_exception=True) def workout_rojaboimport_view(request, message="", userid=0): # pragma: no cover r = getrequestrower(request, userid=userid) if r.user != request.user: messages.error( request, 'You can only access your own workouts on Rojabo, not those of your athletes') url = reverse('workout_rojaboimport_view', kwargs={'userid': request.user.id}) return HttpResponseRedirect(url) try: _ = rojabo_open(request.user) except NoTokenError: # pragma: no cover return HttpResponseRedirect("/rowers/me/rojaboauthorize/") res = rojabo_stuff.get_rojabo_workout_list(request.user) if (res.status_code != 200): # pragma: no cover if (res.status_code == 401): r = getrower(request.user) if (r.rojabo_token == '') or (r.rojabo_token is None): s = "Token doesn't exist. Need to authorize" return HttpResponseRedirect("/rowers/me/rojaboauthorize/") message = "Something went wrong in workout_rojaboimport_view" messages.error(request, message) url = reverse('workouts_view') return HttpResponseRedirect(url) sessions = [] r = getrower(request.user) if request.method == "POST": try: tdict = dict(request.POST.lists()) ids = tdict['sessionid'] rojaboids = [int(id) for id in ids] alldata = {} for item in res.json(): alldata[item['training_session']['id']] = item['training_session'] for rojaboid in rojaboids: try: item = alldata[rojaboid] name = item['workout'] spm = item['stroke'] points = item['points'] manager = userid comment = 'ROJABO {name}, SPM: {spm}. Points: {points}'.format( name = name, points = points, spm = spm, ) preferreddate = datetime.datetime.strptime(item['training_date'],"%Y-%m-%d") startdate = preferreddate enddate = preferreddate ps = PlannedSession( name = item['workout'], comment = comment, startdate = startdate, enddate = enddate, preferreddate = preferreddate, sessionsport = 'rower', manager = request.user, rojabo_id = rojaboid, ) ps.save() ps.rower.add(r) ps.tags.add('ROJABO') ps.save() # get steps if there are any steps = [] try: steps = steps+rojabo_stuff.stepsconvert( item['warm_up']['steps'], warmup=True ) except KeyError: # pragma: no cover pass try: steps = steps + rojabo_stuff.stepsconvert( item['primary']['steps'], startid=len(steps) ) except KeyError: # pragma: no cover pass try: steps = steps + rojabo_stuff.stepsconvert( item['cool_down']['steps'], cooldown=True, startid=len(steps)) except KeyError: # pragma: no cover pass if steps: ps.steps = { 'name':'', 'sport':'rowing', 'filename':'', 'steps': steps, } ps.save() messages.info(request,'Saved planned session {id}'.format(id=ps.id)) except KeyError: # pragma: no cover pass except ValidationError: messages.error(request,"You cannot import sessions") except KeyError: pass rojabo_ids = [int(item['training_session']['id']) for item in res.json()] knownrojaboids = uniqify([ ps.rojabo_id for ps in PlannedSession.objects.filter(manager=request.user) ]) for item in res.json(): i = item['training_session']['id'] if i in knownrojaboids: nnn = '' else: nnn = 'NEW' n = item['training_session']['workout'] spm = item['training_session']['stroke'] d = item['training_session']['training_date'] p = item['training_session']['points'] sessions.append({ 'id':i, 'name':n, 'spm':spm, 'new':nnn, 'date': d, 'points': p, }) breadcrumbs = [ { 'url': '/rowers/list-workouts/', 'name': 'Workouts' }, { 'url': reverse('workout_rojaboimport_view'), 'name': 'Rojabo' }, ] checknew = request.GET.get('selectallnew', False) return render(request, 'rojabo_list_import.html', {'sessions': sessions, 'rower': r, 'active': 'nav-plans', 'breadcrumbs': breadcrumbs, 'teams': get_my_teams(request.user), 'checknew': checknew, }) @csrf_exempt def intervals_webhook_view(request): if request.method == 'GET': verificationtoken = request.GET.get('secret') if verificationtoken != intervals.webhookverification: return HttpResponse(status=403) dologging("intervals_webhooks.log","GET request") dologging("intervals_webhooks.log",request.body) else: dologging("intervals_webhooks.log",request.body) data = json.loads(request.body) try: verificationtoken = data['secret'] except KeyError: return HttpResponse(status=403) if verificationtoken != intervals.webhookverification: return HttpResponse(status=403) try: events = data['events'] except KeyError: # return invalid request if no events return HttpResponse(status=200) webhook_type = None for event in events: try: athlete_id = event['athlete_id'] webhook_type = event['type'] r = Rower.objects.get(intervals_owner_id=athlete_id) except Rower.DoesNotExist: return HttpResponse(status=200) except MultipleObjectsReturned: rs = Rower.objects.filter(intervals_owner_id=athlete_id) r = rs[0] except KeyError: return HttpResponse(status=200) integration = IntervalsIntegration(r.user) if webhook_type.lower() == 'calendar_updated': if can_add_session(r.user): integration.update_calendar(r, event) if webhook_type.lower() == 'activity_uploaded': integration.import_activities(event) if webhook_type.lower() == 'activity_updated': integration.update_activities(event) if webhook_type.lower() == 'activity_deleted': integration.delete_activities(event) return HttpResponse(status=200) # for Strava webhook request validation @csrf_exempt def strava_webhook_view(request): if request.method == 'GET': challenge = request.GET.get('hub.challenge') verificationtoken = request.GET.get('hub.verify_token') if verificationtoken != strava.webhookverification: # pragma: no cover return HttpResponse(status=403) data = {"hub.challenge": challenge} return JSONResponse(data) # logging t = time.localtime() timestamp = time.strftime('%b-%d-%Y_%H%M', t) dologging("strava_webhooks.log",request.body) # POST - does nothing so far data = json.loads(request.body) try: aspect_type = data['aspect_type'] object_type = data['object_type'] strava_owner = data['owner_id'] _ = data['event_time'] except KeyError: # pragma: no cover timestamp = time.strftime('%b-%d-%Y_%H%M', t) dologging("strava_webhooks.log","KeyError line 1330") return HttpResponse(status=200) if object_type == 'activity': if aspect_type == 'create': try: stravaid = data['object_id'] except KeyError: # pragma: no cover dologging('strava_webhooks.log', 'KeyError line 1105') return HttpResponse(status=200) try: r = Rower.objects.get(strava_owner_id=strava_owner) except Rower.DoesNotExist: # pragma: no cover dologging('strava_webhooks.log', 'Rower not found') return HttpResponse(status=200) except MultipleObjectsReturned: # pragma: no cover s = 'Multiple rowers found for strava ID {id}'.format( id=strava_owner) dologging('strava_webhooks.log', s) rs = Rower.objects.filter(strava_owner_id=strava_owner) r = rs[0] ws = Workout.objects.filter(uploadedtostrava=stravaid) if ws.count() == 0 and r.strava_auto_import: strava_integration = importsources['strava'](r.user) jobid = strava_integration.get_workout(stravaid) if jobid == 0: # pragma: no cover dologging('strava_webhooks.log', 'Strava strava_open yielded NoTokenError') else: # pragma: no cover dologging('strava_webhooks.log', 'Workouts already existing') for w in ws: dologging('strava_webhooks.log', str(w)) elif aspect_type == 'delete': try: stravaid = data['object_id'] except KeyError: # pragma: no cover dologging('strava_webhooks.log', 'KeyError line 1132') try: ws = Workout.objects.filter(uploadedtostrava=stravaid) if ws.count() == 0: return HttpResponse(status=200) except Workout.DoesNotExist: # pragma: no cover return HttpResponse(status=200) try: # pragma: no cover r = Rower.objects.get(strava_owner_id=strava_owner) except Rower.DoesNotExist: # pragma: no cover dologging('strava_webhooks.log', 'Rower not found') return HttpResponse(status=200) except MultipleObjectsReturned: # pragma: no cover rs = Rower.objects.filter(strava_owner_id=strava_owner) r = rs[0] if r.strava_auto_delete: # pragma: no cover for w in ws: if w.user == r: w.delete() elif aspect_type == 'update': try: updates = data['updates'] stravaid = data['object_id'] except KeyError: # pragma: no cover dologging("strava_webhooks.log","KeyError line 1391") return HttpResponse(status=200) try: ws = Workout.objects.filter(uploadedtostrava=stravaid) if ws.count() == 0: # pragma: no cover return HttpResponse(status=200) except Workout.DoesNotExist: # pragma: no cover timestamp = time.strftime('%b-%d-%Y_%H%M', t) dologging("strava_webhooks.log","workout not found") return HttpResponse(status=200) try: r = Rower.objects.get(strava_owner_id=strava_owner) except Rower.DoesNotExist: # pragma: no cover timestamp = time.strftime('%b-%d-%Y_%H%M', t) dologging("strava_webhooks.log","Rower not found") return HttpResponse(status=200) except MultipleObjectsReturned: # pragma: no cover rs = Rower.objects.filter(strava_owner_id=strava_owner) r = rs[0] for key, value in updates.items(): for w in ws: if key == 'title': w.name = value w.save() if key == 'type' and r.strava_auto_import: try: w.workouttype = mytypes.stravamappinginv[value] w.save() except KeyError: # pragma: no cover dologging("strava_webhooks.log","Workout type not found") return HttpResponse(status=200) return HttpResponse(status=200) # For push notifications from Garmin @login_required() def workout_getimportview(request, externalid, source='c2', do_async=True): try: integration = importsources[source](request.user) except (TypeError, NotImplementedError, KeyError): # pragma: no cover return reverse("workouts_view") if 'startdate' in request.session and source == 'nk': # pragma: no cover startdate = request.session.get('startdate') enddate = request.session.get('enddate') try: result = integration.get_workout(externalid, startdate=startdate, enddate=enddate) except NoTokenError: return HttpResponseRedirect(reverse(importauthorizeviews[source],kwargs={'source':source})) url = reverse(importlistviews[source]) return HttpResponseRedirect(url) try: result = integration.get_workout(externalid) except NoTokenError: return HttpResponseRedirect(reverse(importauthorizeviews[source],kwargs={'source':source})) if result: # pragma: no cover messages.info( request, "Your workout will be imported in the background") # this should return to the respective import list page else: # pragma: no cover messages.error(request, 'Error getting the workout') url = reverse("workout_import_view", kwargs={'source':source}) return HttpResponseRedirect(url) # Imports all new workouts from SportTracks @login_required() def workout_getsporttracksworkout_all(request): # pragma: no cover st_integration = importsources['sporttracks'](request.user) try: _ = st_integration.get_workouts() messages.info(request,"Your SportTracks workouts will be imported in the background") except NoTokenError: messages.error(request,"You have to connect to SportTracks first") url = reverse('workouts_view') return HttpResponseRedirect(url)