152 lines
4.6 KiB
Python
152 lines
4.6 KiB
Python
from .integrations import SyncIntegration, NoTokenError
|
|
from rowers.models import User, Rower, Workout, TombStone
|
|
|
|
import django_rq
|
|
queue = django_rq.get_queue('default')
|
|
queuelow = django_rq.get_queue('low')
|
|
queuehigh = django_rq.get_queue('high')
|
|
|
|
from rowers.utils import myqueue, dologging, myqueue
|
|
|
|
import requests
|
|
import urllib
|
|
from rowingdata import rowingdata
|
|
|
|
from rowers.rower_rules import is_workout_user
|
|
import time
|
|
from django_rq import job
|
|
from rowers.mytypes import tpmapping, tcxmapping
|
|
|
|
from rowers.tasks import check_tp_workout_id, handle_workout_tp_upload
|
|
|
|
from rowsandall_app.settings import (
|
|
TP_CLIENT_ID, TP_CLIENT_SECRET,
|
|
TP_REDIRECT_URI, TP_CLIENT_KEY,TP_API_LOCATION,
|
|
TP_OAUTH_LOCATION,
|
|
)
|
|
|
|
import gzip
|
|
|
|
import base64
|
|
from io import BytesIO
|
|
|
|
|
|
tpapilocation = TP_API_LOCATION
|
|
|
|
|
|
|
|
class TPIntegration(SyncIntegration):
|
|
def __init__(self, *args, **kwargs):
|
|
super(TPIntegration, self).__init__(*args, **kwargs)
|
|
self.oauth_data = {
|
|
'client_id': TP_CLIENT_ID,
|
|
'client_secret': TP_CLIENT_SECRET,
|
|
'redirect_uri': TP_REDIRECT_URI,
|
|
'authorization_uri': "https://oauth.trainingpeaks.com/oauth/authorize?",
|
|
'content_type': 'application/x-www-form-urlencoded',
|
|
'tokenname': 'tptoken',
|
|
'refreshtokenname': 'tprefreshtoken',
|
|
'expirydatename': 'tptokenexpirydate',
|
|
'bearer_auth': False,
|
|
'base_url': "https://oauth.trainingpeaks.com/oauth/token",
|
|
'scope': 'write',
|
|
}
|
|
|
|
def get_name(self):
|
|
return "TrainingPeaks"
|
|
|
|
def get_shortname(self):
|
|
return "trainingpeaks"
|
|
|
|
def createworkoutdata(self, w, *args, **kwargs):
|
|
filename = w.csvfilename
|
|
row = rowingdata(csvfile=filename)
|
|
tcxfilename = filename[:-4]+'.tcx'
|
|
try:
|
|
newnotes = w.notes+'\n from '+w.workoutsource+' via rowsandall.com'
|
|
except TypeError:
|
|
newnotes = 'from '+w.workoutsource+' via rowsandall.com'
|
|
|
|
try:
|
|
#row.exporttotcx(tcxfilename, notes=newnotes, sport=tcxmapping[w.workouttype])
|
|
row.exporttotcx(tcxfilename, notes=newnotes, sport=None)
|
|
except KeyError:
|
|
row.exporttotcx(tcxfilename, notes=newnotes, sport='other')
|
|
|
|
return tcxfilename
|
|
|
|
|
|
def workout_export(self, workout, *args, **kwargs) -> str:
|
|
thetoken = self.open()
|
|
tcxfilename = self.createworkoutdata(workout)
|
|
job = myqueue(
|
|
queue,
|
|
handle_workout_tp_upload,
|
|
workout,
|
|
thetoken,
|
|
tcxfilename,
|
|
tpmapping[workout.workouttype]
|
|
)
|
|
return job.id
|
|
|
|
|
|
def get_workouts(self, *args, **kwargs) -> int:
|
|
raise NotImplementedError("not implemented")
|
|
|
|
def get_workout(self, id, *args, **kwargs) -> int:
|
|
raise NotImplementedError("not implemented")
|
|
|
|
def get_workout_list(self, *args, **kwargs) -> list:
|
|
raise NotImplementedError("not implemented")
|
|
|
|
def make_authorization_url(self, *args, **kwargs) -> str: # pragma: no cover
|
|
params = {"client_id": TP_CLIENT_KEY,
|
|
"response_type": "code",
|
|
"redirect_uri": TP_REDIRECT_URI,
|
|
"scope": "file:write",
|
|
}
|
|
url = TP_OAUTH_LOCATION+"oauth/authorize/?" + \
|
|
urllib.parse.urlencode(params)
|
|
|
|
return url
|
|
|
|
|
|
def get_token(self, code, *args, **kwargs) -> (str, int, str):
|
|
# client_auth = requests.auth.HTTPBasicAuth(TP_CLIENT_KEY, TP_CLIENT_SECRET)
|
|
post_data = {
|
|
"client_id": TP_CLIENT_KEY,
|
|
"grant_type": "authorization_code",
|
|
"code": code,
|
|
"redirect_uri": TP_REDIRECT_URI,
|
|
"client_secret": TP_CLIENT_SECRET,
|
|
}
|
|
|
|
response = requests.post(
|
|
TP_OAUTH_LOCATION+"/oauth/token/",
|
|
data=post_data, verify=False,
|
|
)
|
|
|
|
if response.status_code != 200:
|
|
raise NoTokenError("no token")
|
|
|
|
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
|
|
thetoken = ""
|
|
expires_in = 0
|
|
refresh_token = ""
|
|
|
|
return thetoken, expires_in, refresh_token
|
|
|
|
|
|
def open(self, *args, **kwargs) -> str:
|
|
return super(TPIntegration, self).open(*args, **kwargs)
|
|
|
|
def token_refresh(self, *args, **kwargs) -> str:
|
|
return super(TPIntegration, self).token_refresh(*args, **kwargs)
|
|
|
|
|