1230 lines
43 KiB
Python
1230 lines
43 KiB
Python
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',
|
|
'garmin': 'rower_garmin_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 == 'garmin':
|
|
return rower_garmin_authorize(request)
|
|
if source == 'rojabo':
|
|
return rower_rojabo_authorize(request)
|
|
if source == 'polar':
|
|
return rower_polar_authorize(request)
|
|
url = integration.make_authorization_url()
|
|
return HttpResponseRedirect(url)
|
|
|
|
|
|
|
|
@login_required()
|
|
def rower_garmin_authorize(request): # pragma: no cover
|
|
authorization_url, token, secret = garmin_stuff.garmin_authorize()
|
|
request.session['garmin_owner_key'] = token
|
|
request.session['garmin_owner_secret'] = secret
|
|
return HttpResponseRedirect(authorization_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<id>\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 Garmin callback
|
|
@login_required()
|
|
def rower_process_garmincallback(request): # pragma: no cover
|
|
r = getrower(request.user)
|
|
absoluteurl = request.build_absolute_uri()
|
|
|
|
try:
|
|
key = request.session['garmin_owner_key']
|
|
secret = request.session['garmin_owner_secret']
|
|
except KeyError:
|
|
authorization_url, key, secret = garmin_stuff.garmin_authorize()
|
|
garmintoken, garminrefreshtoken = garmin_stuff.garmin_processcallback(
|
|
absoluteurl, key, secret)
|
|
r.garmintoken = garmintoken
|
|
r.garminrefreshtoken = garminrefreshtoken
|
|
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 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()
|
|
@permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True)
|
|
@permission_required('rower.is_not_freecoach', fn=get_user_by_userid, raise_exception=True)
|
|
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)
|
|
|
|
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
|
|
)
|
|
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(isplanmember, login_url="/rowers/paidplans/",
|
|
message="This functionality requires a Self-coach 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:
|
|
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
|
|
|
|
|
|
@csrf_exempt
|
|
def garmin_summaries_view(request): # pragma: no cover
|
|
if request.method != 'POST':
|
|
return HttpResponse(status=200)
|
|
|
|
t = time.localtime()
|
|
timestamp = time.strftime('%b-%d-%Y_%H%M', t)
|
|
dologging("garminlog.log",request.body)
|
|
# POST request
|
|
data = json.loads(request.body)
|
|
activities = data['activities']
|
|
result = garmin_stuff.garmin_workouts_from_summaries(activities)
|
|
|
|
if result:
|
|
return HttpResponse(status=200)
|
|
|
|
return HttpResponse(status=200)
|
|
|
|
|
|
@csrf_exempt
|
|
def garmin_newfiles_ping(request): # pragma: no cover
|
|
if request.method != 'POST':
|
|
return HttpResponse(status=200)
|
|
|
|
data = json.loads(request.body)
|
|
for file in data['activityFiles']:
|
|
try:
|
|
garmintoken = file['userAccessToken']
|
|
try:
|
|
r = Rower.objects.get(garmintoken=garmintoken)
|
|
callbackURL = file['callbackURL']
|
|
starttime = file['startTimeInSeconds']
|
|
fileType = file['fileType']
|
|
_ = garmin_stuff.get_garmin_file(r, callbackURL, starttime, fileType)
|
|
except Rower.DoesNotExist:
|
|
pass
|
|
except KeyError:
|
|
pass
|
|
|
|
return HttpResponse(status=200) # pragma: no cover
|
|
|
|
|
|
@csrf_exempt
|
|
def garmin_deregistration_view(request):
|
|
if request.method != 'POST': # pragma: no cover
|
|
return HttpResponse(status=200)
|
|
|
|
data = json.loads(request.body)
|
|
deregistrations = data['deregistrations']
|
|
for deregistration in deregistrations:
|
|
try:
|
|
garmintoken = deregistration['userAccessToken']
|
|
try:
|
|
r = Rower.objects.get(garmintoken=garmintoken)
|
|
r.garmintoken = ''
|
|
r.save()
|
|
except Rower.DoesNotExist: # pragma: no cover
|
|
pass
|
|
except KeyError: # pragma: no cover
|
|
pass
|
|
|
|
return HttpResponse(status=200)
|
|
|
|
|
|
@csrf_exempt
|
|
def garmin_details_view(request):
|
|
if request.method != 'POST': # pragma: no cover
|
|
return HttpResponse(status=200)
|
|
|
|
# POST request
|
|
data = json.loads(request.body)
|
|
_ = garmin_stuff.garmin_workouts_from_details(data)
|
|
|
|
return HttpResponse(status=200)
|
|
|
|
|
|
|
|
@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)
|
|
|