Private
Public Access
1
0

testing full refactor, excluding garmin, polar, rojabo

This commit is contained in:
Sander Roosendaal
2023-02-17 17:14:56 +01:00
parent f17fb560c5
commit 20a7ebea46
8 changed files with 97 additions and 499 deletions

View File

@@ -10,6 +10,7 @@ from rowers.tasks import fetch_strava_workout
import rowers.integrations.strava as strava
from rowers.integrations import importsources
from rowers.utils import NoTokenError
import numpy
@@ -94,7 +95,11 @@ def rower_polar_authorize(request): # pragma: no cover
@login_required()
def rower_integration_token_refresh(request, source='c2'):
integration = importsource[source](request.user)
try:
integration = importsources[source](request.user)
except KeyError:
url = reverse('workouts_view')
return HttpResponseRedirect(url)
try:
token = integration.token_refresh()
messages.info(request, "Tokens refreshed. Good to go")
@@ -320,58 +325,57 @@ def rower_process_nkcallback(request): # pragma: no cover
return HttpResponseRedirect(url)
@login_required()
def workout_getnkworkout_all(request, startdatestring='', enddatestring=''):
@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, startdatestring=startdatestring, enddatestring=enddatestring)
request, defaulttimeperiod='last30')
startdate = startdate.date()
enddate = enddate.date()
r = getrequestrower(request)
integration = importsources[source](request.user)
try:
_ = integration.open()
except NoTokenError: # pragma: no cover
return HttpResponseRedirect("/rowers/me/nkauthorize/")
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))
nk_integration = importsources['nk'](request.user)
try:
_ = nk_integration.open()
except NoTokenError: # pragma: no cover
return HttpResponseRedirect("rower_nk_authorize")
r = getrequestrower(request)
result = nk_integration.get_workouts(before=before, after=after)
if result:
messages.info(
request, "Your NK workouts will be imported in the coming few minutes")
else: # pragma: no cover
messages.error(request, "Your NK workouts import failed")
url = reverse('workouts_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_nkimport_view(request, userid=0, after=0, before=0):
startdate, enddate = get_dates_timeperiod(
request, defaulttimeperiod='last30')
startdate = startdate.date()
enddate = enddate.date()
r = getrequestrower(request, userid=userid)
nk_integration = importsources['nk'](request.user)
try:
_ = nk_integration.open()
except NoTokenError: # pragma: no cover
return HttpResponseRedirect("/rowers/me/nkauthorize/")
workouts = nk_integration.get_workout_list(before=0, after=0)
workouts = integration.get_workout_list(before=before, after=after, startdate=startdate, enddate=enddate)
if request.method == 'POST': # pragma: no cover
@@ -380,11 +384,11 @@ def workout_nkimport_view(request, userid=0, after=0, before=0):
ids = tdict['workoutid']
nkids = [int(id) for id in ids]
for nkid in nkids:
_ = nk_integration.get_workout(nkid)
_ = integration.get_workout(nkid)
messages.info(
request,
'Your NK logbook workouts will be imported in the background.'
' It may take a few minutes before they appear.')
'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:
@@ -396,27 +400,29 @@ def workout_nkimport_view(request, userid=0, after=0, before=0):
'name': 'Workouts'
},
{
'url': reverse('workout_nkimport_view'),
'name': 'NK Logbook'
'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,
'startdate': startdate,
'enddate': enddate,
'dateform':dateform,
'active': 'nav-workouts',
'breadcrumbs': breadcrumbs,
'teams': get_my_teams(request.user),
'checknew': checknew,
'integration': 'NK Logbook'
'startdate':startdate,
'enddate': enddate,
'integration': integration.get_name()
})
# Process Strava Callback
@login_required()
@@ -568,67 +574,6 @@ def rower_process_testcallback(request): # pragma: no cover
return HttpResponse(text)
@login_required()
@permission_required('rower.is_coach', fn=get_user_by_userid, raise_exception=True)
def workout_rp3import_view(request, userid=0):
r = getrequestrower(request, userid=userid)
rp3_integration = importsources['rp3'](request.user)
try:
_ = rp3_integration.open()
except NoTokenError: # pragma: no cover
url = reverse('rower_rp3_authorize')
return HttpResponseRedirect(url)
workouts = rp3_integration.get_workout_list()
datedict = {}
for workout in workouts:
datedict[workout['id']] = workout['starttime']
if request.method == "POST":
try: # pragma: no cover
tdict = dict(request.POST.lists())
ids = tdict['workoutid']
rp3ids = [int(id) for id in ids]
for rp3id in rp3ids:
rp3_integration.get_workout(rp3id,startdatetime=datedict[rp3id])
# done, redirect to workouts list
messages.info(
request,
'Your RP3 workouts will be imported in the background.'
' It may take a few minutes before they appear.')
url = reverse('workouts_view')
return HttpResponseRedirect(url)
except KeyError: # pragma: no cover
pass
breadcrumbs = [
{
'url': '/rowers/list-workouts/',
'name': 'Workouts'
},
{
'url': reverse('workout_rp3import_view'),
'name': 'RP3'
},
]
checknew = request.GET.get('selectallnew', False)
return render(request, 'list_import.html',
{
'workouts': workouts,
'rower': r,
'active': 'nav-workouts',
'breadcrumbs': breadcrumbs,
'teams': get_my_teams(request.user),
'integration': 'RP3',
'checknew': checknew,
})
# The page where you select which Strava workout to import
@login_required()
@@ -791,72 +736,6 @@ def workout_rojaboimport_view(request, message="", userid=0):
'checknew': checknew,
})
# The page where you select which Strava workout to import
@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_stravaimport_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 the NK Logbook, not those of your athletes')
url = reverse('workout_stravaimport_view',
kwargs={'userid': request.user.id})
return HttpResponseRedirect(url)
strava_integration = importsources['strava'](request.user)
try:
_ = strava_integration.open()
except NoTokenError: # pragma: no cover
return HttpResponseRedirect("/rowers/me/stravaauthorize/")
workouts = strava_integration.get_workout_list()
if request.method == "POST":
try: # pragma: no cover
tdict = dict(request.POST.lists())
ids = tdict['workoutid']
stravaids = [int(id) for id in ids]
alldata = {}
for stravaid in stravaids:
res = strava_integration.get_workout(id)
# done, redirect to workouts list
messages.info(request,
'Your Strava workouts will be imported in the background.'
' It may take a few minutes before they appear.')
url = reverse('workouts_view')
return HttpResponseRedirect(url)
except KeyError: # pragma: no cover
pass
breadcrumbs = [
{
'url': '/rowers/list-workouts/',
'name': 'Workouts'
},
{
'url': reverse('workout_stravaimport_view'),
'name': 'Strava'
},
]
checknew = request.GET.get('selectallnew', False)
# 2022-10-24 sorting the results
workouts = sorted(workouts, key = lambda d:d['starttime'], reverse=True)
return render(request, 'list_import.html',
{'workouts': workouts,
'rower': r,
'active': 'nav-workouts',
'breadcrumbs': breadcrumbs,
'teams': get_my_teams(request.user),
'checknew': checknew,
'integration': 'Strava'
})
# for Strava webhook request validation
@@ -1122,261 +1001,24 @@ def workout_polarimport_view(request, userid=0): # pragma: no cover
'teams': get_my_teams(request.user),
})
# The page where you select which SportTracks workout to import
@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_sporttracksimport_view(request, message="", userid=0):
st_integration = importsources['sporttracks'](request.user)
try:
_ = st_integration.open()
except NoTokenError:
return HttpResponseRedirect("/rowers/me/sporttracksauthorize/")
workouts = st_integration.get_workout_list()
r = getrower(request.user)
if request.method == "POST":
try: # pragma: no cover
tdict = dict(request.POST.lists())
ids = tdict['workoutid']
stids = [int(id) for id in ids]
alldata = {}
for id in stids:
res = st_integration.get_workout(id)
# done, redirect to workouts list
messages.info(request,
'Your SportTracks workouts will be imported in the background.'
' It may take a few minutes before they appear.')
url = reverse('workouts_view')
return HttpResponseRedirect(url)
except KeyError: # pragma: no cover
pass
breadcrumbs = [
{
'url': '/rowers/list-workouts/',
'name': 'Workouts'
},
{
'url': reverse('workout_sporttracksimport_view'),
'name': 'SportTracks'
},
]
checknew = request.GET.get('selectallnew', False)
return render(request, 'list_import.html',
{'workouts': workouts,
'breadcrumbs': breadcrumbs,
'active': 'nav-workouts',
'rower': r,
'teams': get_my_teams(request.user),
'integration':'SportTracks',
'checknew': checknew,
})
return HttpResponse(res) # pragma: no cover
# List of workouts on Concept2 logbook. This view only used for debugging
# Import all unknown workouts available on Concept2 logbook
@login_required()
def workout_getc2workout_all(request, page=1, message=""): # pragma: no cover
r = getrequestrower(request)
c2_integration = importsources['c2'](request.user)
try:
_ = c2_integration.open()
except NoTokenError: # pragma: no cover
return HttpResponseRedirect("/rowers/me/c2authorize/")
result = c2_integration.get_workouts(page=page)
if result:
messages.info(
request, 'Your C2 workouts will be imported in the coming few minutes')
else:
messages.error(request, 'Your C2 workouts import failed')
url = reverse('workouts_view')
return HttpResponseRedirect(url)
@login_required()
def workout_getrp3workout_all(request): # pragma: no cover
try:
_ = rp3_open(request.user)
except NoTokenError: # pragma: no cover
return HttpResponseRedirect("/rowers/me/rp3authorize/")
r = getrequestrower(request)
rp3_integration = importsources['rp3'](request.user)
result = rp3_integration.get_workouts()
if result:
messages.info(
request, 'Your RP3 workouts will be imported in the coming few minutes')
else:
messages.error(request, 'Your RP3 workouts import failed')
url = reverse('workouts_view')
return HttpResponseRedirect(url)
# List of workouts available on Concept2 logbook - for import
@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_c2import_view(request, page=1, userid=0, message=""):
rower = getrequestrower(request, userid=userid)
if rower.user != request.user:
messages.error(
request, 'You can only access your own workouts on the Concept2 Logbook, not those of your athletes')
url = reverse('workout_c2import_view', kwargs={
'userid': request.user.id})
return HttpResponseRedirect(url)
c2_integration = importsources['c2'](request.user)
try:
_ = c2_integration.open()
except NoTokenError: # pragma: no cover
return HttpResponseRedirect("/rowers/me/c2authorize/")
workouts = c2_integration.get_workout_list(page=1)
if request.method == "POST":
try: # pragma: no cover
tdict = dict(request.POST.lists())
ids = tdict['workoutid']
c2ids = [int(id) for id in ids]
for c2id in c2ids:
c2_integration.get_workout(c2id)
# done, redirect to workouts list
messages.info(
request,
'Your Concept2 workouts will be imported in the background.'
' It may take a few minutes before they appear.')
url = reverse('workouts_view')
return HttpResponseRedirect(url)
except KeyError: # pragma: no cover
pass
breadcrumbs = [
{
'url': '/rowers/list-workouts/',
'name': 'Workouts'
},
{
'url': reverse('workout_c2import_view'),
'name': 'Concept2'
},
{
'url': reverse('workout_c2import_view', kwargs={'page': page}),
'name': 'Page '+str(page)
}
]
rower = getrower(request.user)
checknew = request.GET.get('selectallnew', False)
return render(request,
'list_import.html',
{'workouts': workouts,
'rower': rower,
'active': 'nav-workouts',
'breadcrumbs': breadcrumbs,
'teams': get_my_teams(request.user),
'page': page,
'checknew': checknew,
'integration': 'C2 Logbook'
})
importlistviews = {
'c2': 'workout_c2import_view',
'strava': 'workout_stravaimport_view',
'polar': 'workout_polarimport_view',
'ownapi': 'workout_view',
'sporttracks': 'workout_sporttracksimport_view',
'trainingpeaks': 'workout_view',
'nk': 'workout_nkimport_view',
'rp3': 'workout_rp3import_view',
}
importauthorizeviews = {
'c2': 'rower_c2_authorize',
'strava': 'rower_strava_authorize',
'c2': 'rower_integration_authorize',
'strava': 'rower_integration_authorize',
'polar': 'rower_polar_authorize',
'ownapi': 'workout_view',
'sporttracks': 'rower_sporttracks_authorize',
'trainingpeaks': 'rower_tp_authorize',
'nk': 'rower_nk_authorize',
'rp3': 'rower_rp3_authorize',
'sporttracks': 'rower_integration_authorize',
'trainingpeaks': 'rower_integration_authorize',
'nk': 'rower_integration_authorize',
'rp3': 'rower_integration_authorize',
}
@login_required()
def workout_getimportview_old(request, externalid, source='c2', do_async=True):
if 'startdate' in request.session and source == 'nk': # pragma: no cover
startdate = request.session.get('startdate')
enddate = request.session.get('enddate')
try:
result = importsources[source].get_workout(request.user, externalid, do_async=do_async,
startdate=startdate, enddate=enddate)
except NoTokenError:
return HttpResponseRedirect(reverse(importauthorizeviews[source]))
url = reverse(importlistviews[source])
return HttpResponseRedirect(url)
try:
result = importsources[source].get_workout(request.user, externalid,
do_async=do_async)
except NoTokenError:
return HttpResponseRedirect(reverse(importauthorizeviews[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(importlistviews[source])
return HttpResponseRedirect(url)
# Imports all new workouts from SportTracks
@login_required()
def workout_getsporttracksworkout_all(request):
st_integration = importsources['sporttracks'](request.user)
st_integration.get_workouts()
messages.info(request,"Your SportTracks workouts will be imported in the background")
url = reverse('workouts_view')
return HttpResponseRedirect(url)
@login_required()
def workout_getimportview(request, externalid, source='c2', do_async=True):
try:
integration = importsources[source](request.user)
except (TypeError, NotImplementedError, KeyError):
return workout_getimportview_old(request, externalid, source=source, do_async=True)
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')
@@ -1385,14 +1027,14 @@ def workout_getimportview(request, externalid, source='c2', do_async=True):
try:
result = integration.get_workout(externalid, startdate=startdate, enddate=enddate)
except NoTokenError:
return HttpResponseRedirect(reverse(importauthorizeviews[source]))
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]))
return HttpResponseRedirect(reverse(importauthorizeviews[source],kwargs={'source':source}))
if result: # pragma: no cover
messages.info(
@@ -1401,7 +1043,7 @@ def workout_getimportview(request, externalid, source='c2', do_async=True):
else: # pragma: no cover
messages.error(request, 'Error getting the workout')
url = reverse(importlistviews[source])
url = reverse("workout_import_view", kwargs={'source':source})
return HttpResponseRedirect(url)