sander/rowsandall
Private
Public Access
1
0
Fork 0
Code Activity

first attempt on c2

Browse Source
This commit is contained in:
Sander Roosendaal
2023-02-10 16:55:47 +01:00
parent d661f861c4
commit 2981c59a5d
19 changed files with 873 additions and 1129 deletions
Download Patch File Download Diff File Expand all files Collapse all files

179
rowers/views/importviews.py
View File

@@ -7,6 +7,8 @@ from rowers.views.statements import *
from rowers.plannedsessions import get_dates_timeperiod
from rowers.tasks import fetch_strava_workout
from rowers.integrations.c2 import C2Integration
import numpy
@@ -109,9 +111,10 @@ def workout_c2_upload_view(request, id=0):
)
dologging('c2_log.log', s)
c2_integration = C2Integration(request.user)
try:
message, c2id = c2stuff.workout_c2_upload(
request.user, w, asynchron=True)
c2id = c2_integration.workout_export(w)
except NoTokenError: # pragma: no cover
return HttpResponseRedirect("/rowers/me/c2authorize/")
@@ -187,18 +190,9 @@ def rower_nk_authorize(request): # pragma: no cover
# Concept2 authorization
@login_required()
def rower_c2_authorize(request): # pragma: no cover
# Generate a random string for the state parameter
# Save it for use later to prevent xsrf attacks
# state = str(uuid4())
scope = "user:read,results:write"
params = {"client_id": C2_CLIENT_ID,
"response_type": "code",
"redirect_uri": C2_REDIRECT_URI}
url = "http://log.concept2.com/oauth/authorize?" + \
urllib.parse.urlencode(params)
url += "&scope="+scope
c2_integration = C2Integration(request.user)
url = c2_integration.make_authorization_url()
return HttpResponseRedirect(url)
# Garmin authorization
@@ -312,26 +306,12 @@ def rower_tp_authorize(request): # pragma: no cover
# Concept2 token refresh. URL for manual refresh. Not visible to users
@login_required()
def rower_c2_token_refresh(request):
r = getrower(request.user)
res = c2stuff.do_refresh_token(r.c2refreshtoken)
if res[0] is not None:
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.c2token = access_token
r.tokenexpirydate = expirydatetime
r.c2refreshtoken = refresh_token
r.save()
successmessage = "Tokens refreshed. Good to go"
messages.info(request, successmessage)
else: # pragma: no cover
message = "Something went wrong (refreshing tokens). Please reauthorize:"
messages.error(request, message)
c2_integration = C2Integration(request.user)
try:
token = c2_integration.token_refresh()
messages.info(request, "Tokens refreshed. Good to go")
except NoTokenError:
messages.error(request, "Something went wrong refreshing C2 tokens. Please reauthorize")
url = reverse('workouts_view')
@@ -395,10 +375,11 @@ def rower_sporttracks_token_refresh(request):
# Concept2 Callback
@login_required()
def rower_process_callback(request):
c2_integration = C2Integration(request.user)
try:
code = request.GET['code']
res = c2stuff.get_token(code)
except MultiValueDictKeyError: # pragma: no cover
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)
@@ -1677,61 +1658,18 @@ def workout_sporttracksimport_view(request, message="", userid=0):
# List of workouts on Concept2 logbook. This view only used for debugging
@login_required()
def c2listdebug_view(request, page=1, message=""): # pragma: no cover
try:
_ = c2_open(request.user)
except NoTokenError: # pragma: no cover
return HttpResponseRedirect("/rowers/me/c2authorize/")
r = getrower(request.user)
res = c2stuff.get_c2_workout_list(request.user, page=page)
if (res.status_code != 200):
message = "Something went wrong in workout_c2import_view (C2 token renewal)"
messages.error(request, message)
if settings.DEBUG:
return HttpResponse(res)
else:
url = reverse('workouts_view')
return HttpResponseRedirect(url)
else:
workouts = []
for item in res.json()['data']:
d = item['distance']
i = item['id']
ttot = item['time_formatted']
s = item['date']
r = item['type']
s2 = item['source']
c = item['comments']
keys = ['id', 'distance', 'duration',
'starttime', 'rowtype', 'source', 'comment']
values = [i, d, ttot, s, r, s2, c]
res = dict(zip(keys, values))
workouts.append(res)
return render(request,
'c2_list_import2.html',
{'workouts': workouts,
'teams': get_my_teams(request.user),
})
# 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 = C2Integration(request.user)
try:
_ = c2_open(request.user)
_ = c2_integration.open()
except NoTokenError: # pragma: no cover
return HttpResponseRedirect("/rowers/me/c2authorize/")
r = getrequestrower(request)
result = c2stuff.get_c2_workouts(r, page=page, do_async=True)
result = c2_integration.get_workouts(page=page)
if result:
messages.info(
@@ -1777,76 +1715,23 @@ def workout_c2import_view(request, page=1, userid=0, message=""):
'userid': request.user.id})
return HttpResponseRedirect(url)
c2_integration = C2Integration(request.user)
try:
_ = c2_open(request.user)
_ = c2_integration.open()
except NoTokenError: # pragma: no cover
return HttpResponseRedirect("/rowers/me/c2authorize/")
res = c2stuff.get_c2_workout_list(request.user, page=page)
workouts = c2_integration.get_workout_list(page=1)
if (res.status_code != 200): # pragma: no cover
message = "Something went wrong in workout_c2import_view (C2 token refresh)"
messages.error(request, message)
url = reverse('workouts_view')
return HttpResponseRedirect(url)
workouts = []
c2ids = [item['id'] for item in res.json()['data']]
knownc2ids = uniqify([
w.uploadedtoc2 for w in Workout.objects.filter(user=rower)
])
tombstones = [
t.uploadedtoc2 for t in TombStone.objects.filter(user=rower)
]
parkedids = []
try:
with open('c2blocked.json', 'r') as c2blocked:
jsondata = json.load(c2blocked)
parkedids = jsondata['ids']
except: # pragma: no cover
pass
knownc2ids = uniqify(knownc2ids+tombstones+parkedids)
for item in res.json()['data']:
d = item['distance']
i = item['id']
ttot = item['time_formatted']
s = item['date']
r = item['type']
s2 = item['source']
c = item['comments']
if i in knownc2ids:
nnn = ''
else: # pragma: no cover
nnn = 'NEW'
keys = ['id', 'distance', 'duration', 'starttime',
'rowtype', 'source', 'comment', 'new']
values = [i, d, ttot, s, r, s2, c, nnn]
ress = dict(zip(keys, values))
workouts.append(ress)
if request.method == "POST":
try: # pragma: no cover
tdict = dict(request.POST.lists())
ids = tdict['workoutid']
c2ids = [int(id) for id in ids]
alldata = {}
for item in res.json()['data']:
alldata[item['id']] = item
counter = 0
for c2id in c2ids:
_ = myqueue(
queue,
handle_c2_async_workout,
alldata,
rower.user.id,
rower.c2token,
c2id,
counter,
rower.defaulttimezone
)
counter = counter+1
c2_integration.get_workout(c2id)
# done, redirect to workouts list
messages.info(
request,
@@ -1877,7 +1762,7 @@ def workout_c2import_view(request, page=1, userid=0, message=""):
checknew = request.GET.get('selectallnew', False)
return render(request,
'c2_list_import2.html',
'list_import.html',
{'workouts': workouts,
'rower': rower,
'active': 'nav-workouts',
@@ -1885,6 +1770,7 @@ def workout_c2import_view(request, page=1, userid=0, message=""):
'teams': get_my_teams(request.user),
'page': page,
'checknew': checknew,
'integration': 'C2 Logbook'
})
@@ -1909,7 +1795,7 @@ importauthorizeviews = {
}
importsources = {
'c2': c2stuff,
'c2': C2Integration,
'strava': stravastuff,
'polar': polarstuff,
'ownapi': ownapistuff,
@@ -1953,19 +1839,18 @@ def workout_getimportview(request, externalid, source='c2', do_async=True):
startdate = request.session.get('startdate')
enddate = request.session.get('enddate')
integration = importsources[source](request.user)
try:
result = importsources[source].get_workout(request.user, externalid, do_async=do_async,
startdate=startdate, enddate=enddate)
result = integration.get_workout(externalid, 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)
result = integration.get_workout(externalid)
except NoTokenError:
return HttpResponseRedirect(reverse(importauthorizeviews[source]))
if result: # pragma: no cover