Files
zwift-offline/online_sync.py
2024-08-09 12:28:46 -03:00

209 lines
7.4 KiB
Python
Executable File

#!/usr/bin/env python3
#
# Adapted from https://github.com/jlemon/zlogger/blob/master/get_riders.py
#
# The MIT License (MIT)
#
# Copyright (c) 2016 Jonathan Lemon
#
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
#
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NON INFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
import json
import requests
import profile_pb2
def post_credentials(session, username, password):
# Credentials POSTing and tokens retrieval
# POST https://secure.zwift.com/auth/realms/zwift/tokens/access/codes
try:
response = session.post(
url="https://secure.zwift.com/auth/realms/zwift/tokens/access/codes",
headers={
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate",
"Connection": "keep-alive",
"Content-Type": "application/x-www-form-urlencoded",
"Host": "secure.zwift.com",
"User-Agent": "Zwift/1.5 (iPhone; iOS 9.0.2; Scale/2.00)",
"Accept-Language": "en-US;q=1",
},
data={
"client_id": "Zwift_Mobile_Link",
"username": username,
"password": password,
"grant_type": "password",
},
allow_redirects=False,
)
json_dict = json.loads(response.content)
return (json_dict["access_token"], json_dict["refresh_token"], json_dict["expires_in"])
except requests.exceptions.RequestException as e:
print('HTTP Request failed: %s' % e)
except KeyError as e:
print('Invalid uname and/or password')
def query(session, access_token, route):
try:
response = session.get(
url="https://us-or-rly101.zwift.com/%s" % route,
headers={
"Accept-Encoding": "gzip, deflate",
"Accept": "application/x-protobuf-lite",
"Connection": "keep-alive",
"Host": "us-or-rly101.zwift.com",
"User-Agent": "Zwift/115 CFNetwork/758.0.2 Darwin/15.0.0",
"Authorization": "Bearer %s" % access_token,
"Accept-Language": "en-us",
},
)
return response.content
except requests.exceptions.RequestException as e:
print('HTTP Request failed: %s' % e)
def api_login(session, access_token, login_request):
try:
response = session.post(
url="https://us-or-rly101.zwift.com/api/users/login",
headers={
"Content-Type": "application/x-protobuf-lite",
"Accept": "application/x-protobuf-lite",
"Connection": "keep-alive",
"Host": "us-or-rly101.zwift.com",
"User-Agent": "Zwift/115 CFNetwork/758.0.2 Darwin/15.0.0",
"Authorization": "Bearer %s" % access_token,
"Accept-Language": "en-us",
},
data=login_request.SerializeToString(),
)
return response.content
except requests.exceptions.RequestException as e:
print('HTTP Request failed: %s' % e)
def logout(session, refresh_token):
# Logout
# POST https://secure.zwift.com/auth/realms/zwift/tokens/logout
try:
response = session.post(
url="https://secure.zwift.com/auth/realms/zwift/tokens/logout",
headers={
"Accept": "*/*",
"Accept-Encoding": "gzip, deflate",
"Connection": "keep-alive",
"Content-Type": "application/x-www-form-urlencoded",
"Host": "secure.zwift.com",
"User-Agent": "Zwift/1.5 (iPhone; iOS 9.0.2; Scale/2.00)",
"Accept-Language": "en-US;q=1",
},
data={
"client_id": "Zwift_Mobile_Link",
"refresh_token": refresh_token,
},
)
except requests.exceptions.RequestException as e:
print('HTTP Request failed: %s' % e)
def login(session, user, password):
access_token, refresh_token, expired_in = post_credentials(session, user, password)
return access_token, refresh_token
def create_activity(session, access_token, activity):
try:
response = session.post(
url="https://us-or-rly101.zwift.com/api/profiles/%s/activities" % activity.player_id,
headers={
"Content-Type": "application/x-protobuf-lite",
"Accept": "application/json",
"Connection": "keep-alive",
"Host": "us-or-rly101.zwift.com",
"User-Agent": "Zwift/115 CFNetwork/758.0.2 Darwin/15.0.0",
"Authorization": "Bearer %s" % access_token,
"Accept-Language": "en-us",
},
data=activity.SerializeToString(),
)
json_dict = json.loads(response.content)
return json_dict["id"]
except requests.exceptions.RequestException as e:
print('HTTP Request failed: %s' % e)
def upload_activity(session, access_token, activity):
try:
response = session.put(
url="https://us-or-rly101.zwift.com/api/profiles/%s/activities/%s" % (activity.player_id, activity.id),
headers={
"Content-Type": "application/x-protobuf-lite",
"Accept": "application/json",
"Connection": "keep-alive",
"Host": "us-or-rly101.zwift.com",
"User-Agent": "Zwift/115 CFNetwork/758.0.2 Darwin/15.0.0",
"Authorization": "Bearer %s" % access_token,
"Accept-Language": "en-us",
},
data=activity.SerializeToString(),
params={"upload-to-strava": "true"}
)
return response.status_code
except requests.exceptions.RequestException as e:
print('HTTP Request failed: %s' % e)
def get_player_id(session, access_token):
try:
response = session.get(
url="https://us-or-rly101.zwift.com/api/profiles/me",
headers={
"Accept-Encoding": "gzip, deflate",
"Accept": "application/x-protobuf-lite",
"Connection": "keep-alive",
"Host": "us-or-rly101.zwift.com",
"User-Agent": "Zwift/115 CFNetwork/758.0.2 Darwin/15.0.0",
"Authorization": "Bearer %s" % access_token,
"Accept-Language": "en-us",
},
)
profile = profile_pb2.PlayerProfile()
profile.ParseFromString(response.content)
return profile.id
except requests.exceptions.RequestException as e:
print('HTTP Request failed: %s' % e)