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

commit

Browse Source
This commit is contained in:
Sander Roosendaal
2022-02-15 08:05:12 +01:00
parent 5b3d7fcf2c
commit 8af7ac8af4
71 changed files with 19992 additions and 19476 deletions
Download Patch File Download Diff File Expand all files Collapse all files

111
rowers/tpstuff.py
View File

@@ -1,4 +1,8 @@
from __future__ import absolute_import
from celery import Celery, app
from rowers.rower_rules import is_workout_user
import time
from django_rq import job
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
@@ -17,14 +21,11 @@ from rowsandall_app.settings import (
C2_CLIENT_ID, C2_REDIRECT_URI, C2_CLIENT_SECRET,
STRAVA_CLIENT_ID, STRAVA_REDIRECT_URI, STRAVA_CLIENT_SECRET,
TP_CLIENT_ID, TP_CLIENT_SECRET,
TP_REDIRECT_URI,TP_CLIENT_KEY,
)
TP_REDIRECT_URI, TP_CLIENT_KEY,
)
tpapilocation = "https://api.trainingpeaks.com"
from celery import Celery,app
from django_rq import job
import time
#from async_messages import message_user,messages
oauth_data = {
@@ -33,16 +34,14 @@ oauth_data = {
'redirect_uri': TP_REDIRECT_URI,
'autorization_uri': "https://oauth.trainingpeaks.com/oauth/authorize?",
'content_type': 'application/x-www-form-urlencoded',
# 'content_type': 'application/json',
# 'content_type': 'application/json',
'tokenname': 'tptoken',
'refreshtokenname': 'tprefreshtoken',
'expirydatename': 'tptokenexpirydate',
'bearer_auth': False,
'base_url': "https://oauth.trainingpeaks.com/oauth/token",
'scope':'write',
}
from rowers.rower_rules import is_workout_user
'scope': 'write',
}
# Checks if user has UnderArmour token, renews them if they are expired
@@ -50,47 +49,52 @@ def tp_open(user):
return imports_open(user, oauth_data)
# Refresh ST token using refresh token
def do_refresh_token(refreshtoken):
return imports_do_refresh_token(refreshtoken, oauth_data)
# Exchange access code for long-lived access token
def get_token(code):
client_auth = requests.auth.HTTPBasicAuth(TP_CLIENT_KEY, TP_CLIENT_SECRET)
post_data = {
"client_id":TP_CLIENT_KEY,
"client_id": TP_CLIENT_KEY,
"grant_type": "authorization_code",
"code": code,
"redirect_uri":TP_REDIRECT_URI,
"client_secret": TP_CLIENT_SECRET,
"redirect_uri": TP_REDIRECT_URI,
"client_secret": TP_CLIENT_SECRET,
}
headers = {
'Content-Type': 'application/x-www-form-urlencoded',
'Content-Type': 'application/x-www-form-urlencoded',
}
response = requests.post(
"https://oauth.trainingpeaks.com/oauth/token",
data=post_data,verify=False,
"https://oauth.trainingpeaks.com/oauth/token",
data=post_data, verify=False,
)
try:
token_json = response.json()
thetoken = token_json['access_token']
expires_in = token_json['expires_in']
refresh_token = token_json['refresh_token']
except KeyError: # pragma: no cover
except KeyError: # pragma: no cover
thetoken = 0
expires_in = 0
refresh_token = 0
return thetoken,expires_in,refresh_token
return thetoken, expires_in, refresh_token
# Make authorization URL including random string
def make_authorization_url(request): # pragma: no cover
def make_authorization_url(request): # pragma: no cover
return imports_make_authorization_url(oauth_data)
def getidfromresponse(response): # pragma: no cover
def getidfromresponse(response): # pragma: no cover
t = json.loads(response.text)
links = t["_links"]
@@ -99,6 +103,7 @@ def getidfromresponse(response): # pragma: no cover
return int(id)
def createtpworkoutdata(w):
filename = w.csvfilename
row = rowingdata(csvfile=filename)
@@ -108,64 +113,62 @@ def createtpworkoutdata(w):
except TypeError:
newnotes = 'from '+w.workoutsource+' via rowsandall.com'
row.exporttotcx(tcxfilename,notes=newnotes)
row.exporttotcx(tcxfilename, notes=newnotes)
return tcxfilename
def tp_check(access_token): # pragma: no cover
def tp_check(access_token): # pragma: no cover
headers = {
"Content-Type": "application/json",
'Accept': 'application/json',
'Accept': 'application/json',
'authorization': 'Bearer %s' % access_token
}
resp = requests.post(tpapilocation+"/v1/info/version",
headers=headers,verify=False)
headers=headers, verify=False)
return resp
def uploadactivity(access_token,filename,description='',
def uploadactivity(access_token, filename, description='',
name='Rowsandall.com workout'):
data_gz = BytesIO()
with open(filename,'rb') as inF:
with open(filename, 'rb') as inF:
s = inF.read()
with gzip.GzipFile(fileobj=data_gz,mode="w") as gzf:
with gzip.GzipFile(fileobj=data_gz, mode="w") as gzf:
gzf.write(s)
headers = {
'Content-Type': 'application/json',
'Accept': 'application/json',
'Accept': 'application/json',
'Authorization': 'Bearer %s' % access_token
}
data = {
"UploadClient": "rowsandall",
"Filename": filename,
"SetWorkoutPublic": True,
"Title":name,
"Title": name,
"Type": "rowing",
"Comment": description,
"Data": base64.b64encode(data_gz.getvalue()).decode("ascii")
}
}
resp = requests.post(tpapilocation+"/v1/file",
data = json.dumps(data),
headers=headers,verify=False)
data=json.dumps(data),
headers=headers, verify=False)
if resp.status_code != 200: # pragma: no cover
return 0,resp.reason,resp.status_code,headers
if resp.status_code != 200: # pragma: no cover
return 0, resp.reason, resp.status_code, headers
else:
return resp.json()[0]["Id"],"ok",200,""
return resp.json()[0]["Id"], "ok", 200, ""
return 0,0,0,0 # pragma: no cover
return 0, 0, 0, 0 # pragma: no cover
def workout_tp_upload(user,w): # pragma: no cover
def workout_tp_upload(user, w): # pragma: no cover
message = "Uploading to TrainingPeaks"
tpid = 0
r = w.user
@@ -174,40 +177,40 @@ def workout_tp_upload(user,w): # pragma: no cover
# need some code if token doesn't refresh
if (is_workout_user(user,w)):
if (is_workout_user(user, w)):
tcxfile = createtpworkoutdata(w)
if tcxfile:
res,reason,status_code,headers = uploadactivity(
thetoken,tcxfile,
res, reason, status_code, headers = uploadactivity(
thetoken, tcxfile,
name=w.name
)
if res == 0:
message = "Upload to TrainingPeaks failed with status code "+str(status_code)+": "+reason
message = "Upload to TrainingPeaks failed with status code " + \
str(status_code)+": "+reason
w.tpid = -1
try:
os.remove(tcxfile)
except WindowsError:
pass
return message,tpid
return message, tpid
else: # res != 0
else: # res != 0
w.uploadedtotp = res
tpid = res
w.save()
os.remove(tcxfile)
return 'Successfully synchronized to TrainingPeaks',tpid
return 'Successfully synchronized to TrainingPeaks', tpid
else: # no tcxfile
else: # no tcxfile
message = "Upload to TrainingPeaks failed"
w.uploadedtotp = -1
tpid = -1
w.save()
return message,tpid
else: # not allowed to upload
return message, tpid
else: # not allowed to upload
message = "You are not allowed to export this workout to TP"
tpid = 0
return message,tpid
return message, tpid
return message,tpid
return message, tpid