forked from yihong0618/running_page
-
Notifications
You must be signed in to change notification settings - Fork 0
/
garmin_sync.py
341 lines (295 loc) · 11.9 KB
/
garmin_sync.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Python 3 API wrapper for Garmin Connect to get your statistics.
Copy most code from https://github.com/cyberjunky/python-garminconnect
"""
import argparse
import asyncio
import logging
import os
import re
import json
import sys
import time
import traceback
import aiofiles
import cloudscraper
import httpx
from config import GPX_FOLDER, JSON_FILE, SQL_FILE, config
from utils import make_activities_file
# logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
TIME_OUT = httpx.Timeout(240.0, connect=360.0)
GARMIN_COM_URL_DICT = {
"BASE_URL": "https://connect.garmin.com",
"SSO_URL_ORIGIN": "https://sso.garmin.com",
"SSO_URL": "https://sso.garmin.com/sso",
# "MODERN_URL": "https://connect.garmin.com/modern",
"MODERN_URL": "https://connect.garmin.com",
"SIGNIN_URL": "https://sso.garmin.com/sso/signin",
"CSS_URL": "https://static.garmincdn.com/com.garmin.connect/ui/css/gauth-custom-v1.2-min.css",
"UPLOAD_URL": "https://connect.garmin.com/modern/proxy/upload-service/upload/.gpx",
"ACTIVITY_URL": "https://connect.garmin.com/proxy/activity-service/activity/{activity_id}",
}
GARMIN_CN_URL_DICT = {
"BASE_URL": "https://connect.garmin.cn",
"SSO_URL_ORIGIN": "https://sso.garmin.com",
"SSO_URL": "https://sso.garmin.cn/sso",
# "MODERN_URL": "https://connect.garmin.cn/modern",
"MODERN_URL": "https://connect.garmin.cn",
"SIGNIN_URL": "https://sso.garmin.cn/sso/signin",
"CSS_URL": "https://static.garmincdn.cn/cn.garmin.connect/ui/css/gauth-custom-v1.2-min.css",
"UPLOAD_URL": "https://connect.garmin.cn/modern/proxy/upload-service/upload/.gpx",
"ACTIVITY_URL": "https://connect.garmin.cn/proxy/activity-service/activity/{activity_id}",
}
class Garmin:
def __init__(self, email, password, auth_domain, is_only_running=False):
"""
Init module
"""
self.email = email
self.password = password
self.req = httpx.AsyncClient(timeout=TIME_OUT)
self.cf_req = cloudscraper.CloudScraper()
self.URL_DICT = (
GARMIN_CN_URL_DICT
if auth_domain and str(auth_domain).upper() == "CN"
else GARMIN_COM_URL_DICT
)
self.modern_url = self.URL_DICT.get("MODERN_URL")
self.headers = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/79.0.3945.88 Safari/537.36",
"origin": self.URL_DICT.get("SSO_URL_ORIGIN"),
}
self.is_only_running = is_only_running
self.upload_url = self.URL_DICT.get("UPLOAD_URL")
self.activity_url = self.URL_DICT.get("ACTIVITY_URL")
self.is_login = False
def login(self):
"""
Login to portal
"""
params = {
"webhost": self.URL_DICT.get("BASE_URL"),
"service": self.modern_url,
"source": self.URL_DICT.get("SIGNIN_URL"),
"redirectAfterAccountLoginUrl": self.modern_url,
"redirectAfterAccountCreationUrl": self.modern_url,
"gauthHost": self.URL_DICT.get("SSO_URL"),
"locale": "en_US",
"id": "gauth-widget",
"cssUrl": self.URL_DICT.get("CSS_URL"),
"clientId": "GarminConnect",
"rememberMeShown": "true",
"rememberMeChecked": "false",
"createAccountShown": "true",
"openCreateAccount": "false",
"usernameShown": "false",
"displayNameShown": "false",
"consumeServiceTicket": "false",
"initialFocus": "true",
"embedWidget": "false",
"generateExtraServiceTicket": "false",
}
data = {
"username": self.email,
"password": self.password,
"embed": "true",
"lt": "e1s1",
"_eventId": "submit",
"displayNameRequired": "false",
}
try:
self.cf_req.get(
self.URL_DICT.get("SIGNIN_URL"), headers=self.headers, params=params
)
response = self.cf_req.post(
self.URL_DICT.get("SIGNIN_URL"),
headers=self.headers,
params=params,
data=data,
)
except Exception as err:
raise GarminConnectConnectionError("Error connecting") from err
response_url = re.search(r'"(https:[^"]+?ticket=[^"]+)"', response.text)
if not response_url:
raise GarminConnectAuthenticationError("Authentication error")
response_url = re.sub(r"\\", "", response_url.group(1))
try:
response = self.cf_req.get(response_url)
self.req.cookies = self.cf_req.cookies
if response.status_code == 429:
raise GarminConnectTooManyRequestsError("Too many requests")
response.raise_for_status()
self.is_login = True
except Exception as err:
raise GarminConnectConnectionError("Error connecting") from err
async def fetch_data(self, url, retrying=False):
"""
Fetch and return data
"""
try:
response = await self.req.get(url, headers=self.headers)
if response.status_code == 429:
raise GarminConnectTooManyRequestsError("Too many requests")
logger.debug(f"fetch_data got response code {response.status_code}")
response.raise_for_status()
return response.json()
except Exception as err:
if retrying:
logger.debug(
"Exception occurred during data retrieval, relogin without effect: %s"
% err
)
raise GarminConnectConnectionError("Error connecting") from err
else:
logger.debug(
"Exception occurred during data retrieval - perhaps session expired - trying relogin: %s"
% err
)
self.login()
await self.fetch_data(url, retrying=True)
async def get_activities(self, start, limit):
"""
Fetch available activities
"""
if not self.is_login:
self.login()
url = f"{self.modern_url}/proxy/activitylist-service/activities/search/activities?start={start}&limit={limit}"
if self.is_only_running:
url = url + "&activityType=running"
return await self.fetch_data(url)
async def download_activity(self, activity_id):
url = f"{self.modern_url}/proxy/download-service/export/gpx/activity/{activity_id}"
logger.info(f"Download activity from {url}")
response = await self.req.get(url, headers=self.headers)
response.raise_for_status()
return response.read()
async def upload_activities(self, files):
if not self.is_login:
self.login()
for file, garmin_type in files:
files = {"data": ("file.gpx", file)}
try:
res = await self.req.post(
self.upload_url, files=files, headers={"nk": "NT"}
)
except Exception as e:
print(str(e))
# just pass for now
continue
try:
resp = res.json()["detailedImportResult"]
except Exception as e:
print(e)
raise Exception("failed to upload")
# change the type
if resp["successes"]:
activity_id = resp["successes"][0]["internalId"]
print(f"id {activity_id} uploaded...")
data = {"activityTypeDTO": {"typeKey": garmin_type}}
encoding_headers = {"Content-Type": "application/json; charset=UTF-8"}
r = await self.req.put(
self.activity_url.format(activity_id=activity_id),
data=json.dumps(data),
headers=encoding_headers,
)
r.raise_for_status()
await self.req.aclose()
class GarminConnectHttpError(Exception):
def __init__(self, status):
super(GarminConnectHttpError, self).__init__(status)
self.status = status
class GarminConnectConnectionError(Exception):
"""Raised when communication ended in error."""
def __init__(self, status):
"""Initialize."""
super(GarminConnectConnectionError, self).__init__(status)
self.status = status
class GarminConnectTooManyRequestsError(Exception):
"""Raised when rate limit is exceeded."""
def __init__(self, status):
"""Initialize."""
super(GarminConnectTooManyRequestsError, self).__init__(status)
self.status = status
class GarminConnectAuthenticationError(Exception):
"""Raised when login returns wrong result."""
def __init__(self, status):
"""Initialize."""
super(GarminConnectAuthenticationError, self).__init__(status)
self.status = status
async def download_garmin_gpx(client, activity_id):
try:
gpx_data = await client.download_activity(activity_id)
file_path = os.path.join(GPX_FOLDER, f"{activity_id}.gpx")
async with aiofiles.open(file_path, "wb") as fb:
await fb.write(gpx_data)
except:
print(f"Failed to download activity {activity_id}: ")
traceback.print_exc()
pass
async def get_activity_id_list(client, start=0):
activities = await client.get_activities(start, 100)
if len(activities) > 0:
ids = list(map(lambda a: str(a.get("activityId", "")), activities))
print(f"Syncing Activity IDs")
return ids + await get_activity_id_list(client, start + 100)
else:
return []
async def gather_with_concurrency(n, tasks):
semaphore = asyncio.Semaphore(n)
async def sem_task(task):
async with semaphore:
return await task
return await asyncio.gather(*(sem_task(task) for task in tasks))
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("email", nargs="?", help="email of garmin")
parser.add_argument("password", nargs="?", help="password of garmin")
parser.add_argument(
"--is-cn",
dest="is_cn",
action="store_true",
help="if garmin accout is cn",
)
parser.add_argument(
"--only-run",
dest="only_run",
action="store_true",
help="if is only for running",
)
options = parser.parse_args()
email = options.email or config("sync", "garmin", "email")
password = options.password or config("sync", "garmin", "password")
auth_domain = (
"CN" if options.is_cn else config("sync", "garmin", "authentication_domain")
)
is_only_running = options.only_run
if email == None or password == None:
print("Missing argument nor valid configuration file")
sys.exit(1)
# make gpx dir
if not os.path.exists(GPX_FOLDER):
os.mkdir(GPX_FOLDER)
async def download_new_activities():
client = Garmin(email, password, auth_domain, is_only_running)
client.login()
# because I don't find a para for after time, so I use garmin-id as filename
# to find new run to generage
downloaded_ids = [
i.split(".")[0] for i in os.listdir(GPX_FOLDER) if not i.startswith(".")
]
activity_ids = await get_activity_id_list(client)
to_generate_garmin_ids = list(set(activity_ids) - set(downloaded_ids))
print(f"{len(to_generate_garmin_ids)} new activities to be downloaded")
start_time = time.time()
await gather_with_concurrency(
10, [download_garmin_gpx(client, id) for id in to_generate_garmin_ids]
)
print(f"Download finished. Elapsed {time.time()-start_time} seconds")
make_activities_file(SQL_FILE, GPX_FOLDER, JSON_FILE)
await client.req.aclose()
loop = asyncio.get_event_loop()
future = asyncio.ensure_future(download_new_activities())
loop.run_until_complete(future)