""" Garmin sync service. Pulls daily summaries + activities through the `garminconnect` library and upserts them. The library and real Garmin credentials are required to actually run a sync; without them the endpoint reports a clear error instead of crashing. Garmin credentials: the app only stores a scrypt/PBKDF2 *hash* of the Garmin password (so it cannot be recovered), therefore a live sync needs the plaintext garminEmail/garminPassword supplied in the request body. On the library's API — these were verified against garminconnect 0.2.8: * get_user_summary(cdate) -> one day of daily totals * get_sleep_data(cdate) -> sleep, NOT part of the summary * get_hrv_data(cdate) -> HRV, also separate * get_activities_by_date(start, end) -> activities in a date range * get_activities(start, limit) -> PAGINATION, not dates The last two are easy to confuse: `get_activities` takes an offset and a count, so passing it a date silently asks for activity number "2026-08-23". """ import datetime import json import logging import os import threading import uuid from db import execute, query_one, query_all from config import DB_TYPE from services import health from services import garmin_extras as extras from services import garmin_throttle as throttle # How many days back a sync reaches. DEFAULT_SYNC_DAYS = int(os.environ.get("GARMIN_SYNC_DAYS") or 7) def _set_sync_status(user_id, status, now, last_sync_time=None, **fields): """Write the status row; `now` becomes the new last_sync_time by default. Callers that merely *report* being blocked (a rate-limit refusal, a refused start) pass the previous last_sync_time explicitly: a sync that never happened must not move "上次同步" forward, or the UI reads "刚刚" while the data actually stopped syncing hours ago. """ last = last_sync_time if last_sync_time is not None else now cols = ["user_id", "status", "last_sync_time"] + list(fields.keys()) placeholders = ", ".join(["?"] * len(cols)) if DB_TYPE == "mariadb": updates = ", ".join(f"{c}=VALUES({c})" for c in cols if c != "user_id") sql = ( f"INSERT INTO sync_status ({', '.join(cols)}) VALUES ({placeholders}) " f"ON DUPLICATE KEY UPDATE {updates}" ) else: updates = ", ".join(f"{c}=excluded.{c}" for c in cols if c != "user_id") sql = ( f"INSERT INTO sync_status ({', '.join(cols)}) VALUES ({placeholders}) " f"ON CONFLICT(user_id) DO UPDATE SET {updates}" ) execute(sql, [user_id, status, last] + list(fields.values())) def _log_sync_history(user_id, trigger, days, started_at, result): """Append one immutable row per sync attempt. `trigger` names what started it — "auto" (scheduler), "manual" (同步页 开始同步) or "quick" (设置页立即同步) — so the 同步记录 screen can show the whole picture. A failed or refused sync is as much a record as a successful one; the user is usually looking at this list because one of those did not behave. A write failure must never break the sync it is reporting, so it is logged and swallowed. """ try: finished = datetime.datetime.utcnow().isoformat(timespec="seconds") r = result or {} duration = 0 try: s = datetime.datetime.fromisoformat(str(started_at).replace(" ", "T")) f = datetime.datetime.fromisoformat(finished) duration = max(0, int((f - s).total_seconds())) except (TypeError, ValueError): pass execute( "INSERT INTO sync_history (id, user_id, trigger_kind, status, days, " "records_synced, activities_synced, badges_synced, " "personal_records_synced, started_at, finished_at, " "duration_seconds, message) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", [ str(uuid.uuid4()), user_id, trigger, r.get("status") or "error", days if days is not None else 0, r.get("recordsSynced") or 0, r.get("activitiesSynced") or 0, r.get("badgesSynced") or 0, r.get("personalRecordsSynced") or 0, started_at, finished, duration, (r.get("message") or "")[:500], ], ) except Exception as e: # noqa: BLE001 - history must never break a sync logging.getLogger(__name__).warning( "sync history write failed for %s: %s", user_id, e ) def get_sync_history(user_id, limit=50): """Most recent sync attempts for the user, newest first.""" rows = query_all( "SELECT trigger_kind, status, days, records_synced, activities_synced, " "badges_synced, personal_records_synced, started_at, finished_at, " "duration_seconds, message " "FROM sync_history WHERE user_id = ? " "ORDER BY started_at DESC LIMIT ?", [user_id, limit], ) return [ { "triggerKind": r["trigger_kind"], "status": r["status"], "days": r["days"], "recordsSynced": r["records_synced"], "activitiesSynced": r["activities_synced"], "badgesSynced": r["badges_synced"], "personalRecordsSynced": r["personal_records_synced"], "startedAt": r["started_at"], "finishedAt": r["finished_at"], "durationSeconds": r["duration_seconds"], "message": r["message"], } for r in rows ] def get_sync_status(user_id): row = query_one("SELECT * FROM sync_status WHERE user_id = ?", [user_id]) # Count distinct days actually stored in the database for this user. count = query_one( "SELECT COUNT(DISTINCT date) AS cnt FROM health_data WHERE user_id = ?", [user_id] ) total_days = count["cnt"] if count else 0 if not row: return { "status": "idle", "lastSyncTime": None, "recordsSynced": 0, "totalDays": total_days, "lastError": None, } return { "status": row["status"], "lastSyncTime": row["last_sync_time"], "recordsSynced": row["records_synced"], "totalDays": total_days, "lastError": row["last_error"], "progressCurrent": row.get("progress_current"), "progressTotal": row.get("progress_total"), "startedAt": row.get("started_at"), "stage": row.get("stage"), } def reset_stale_syncs(): """Clear a "syncing" status left behind by a process that went away. The status lives in the database but the work lives in a thread. A restart (or a crash) takes the thread and leaves the row, so the UI shows a progress bar that will never move and refuses to start a new sync. """ execute( "UPDATE sync_status SET status = 'idle', stage = NULL " "WHERE status = 'syncing'" ) class RateLimited(RuntimeError): """Garmin answered 429. It reaches us disguised: the body is the plain text "Rate limited", and garth feeds that to json.loads, so the exception surfaced as `JSONDecodeError: Expecting value: line 1 column 1` — which reads like a parsing bug rather than "stop asking". Naming it means the sync status says what is actually wrong. """ # Garmin answering 429 is what deepens the limit: the account once stayed # stuck for days because every retry re-hit the throttle before its own # (multi-hour) window closed. We therefore record a 24h cooldown when a 429 # actually arrives, so the UI and the sync history can say when to expect # recovery. The cooldown is *information*, not a gate: since 2026-09-02 the # local estimate is no longer used to refuse a sync before it starts — every # entry point (manual, 立即同步 and the automatic scheduler alike) issues a # real request and trusts Garmin's live answer. Only a real 429 writes a new # cooldown; a stale local estimate must never keep a healthy account idle. RATE_LIMIT_BACKOFF = datetime.timedelta(hours=24) # In-process cache of the cooldown, kept in sync with the DB copy below and # still the lever the tests reach for via _rate_limited_until.clear(). _rate_limited_until = {} def _parse_dt(v): """Coerce a stored rate-limit time into a naive UTC datetime, or None.""" if v is None: return None if isinstance(v, datetime.datetime): return v if isinstance(v, str): for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S"): try: return datetime.datetime.strptime(v, fmt) except ValueError: continue try: return datetime.datetime.fromisoformat(v) except ValueError: return None return None def rate_limited_until(user_id): """When the account is still cooling down, as a UTC datetime (or None). The cooldown lives in the database and is the single source of truth, so every gunicorn worker and a restart agree on it. A process-local dict caused a nasty "stuck forever" bug: a worker would remember a future deadline whose DB write had silently failed to persist, and go on blocking even after the real deadline had passed. We therefore trust the DB row, falling back to the in-memory cache only when no row exists yet. """ try: row = query_one( "SELECT rate_limited_until FROM sync_status WHERE user_id = ?", [user_id] ) except Exception: row = None db_val = _parse_dt(row["rate_limited_until"] if row else None) if db_val is not None: return db_val return _rate_limited_until.get(user_id) def _note_rate_limit(user_id): """Record a rate-limit cooldown. A 429 means Garmin is throttling this account/IP, and every further request just extends the window — so we stand down for a long, fixed stretch (RATE_LIMIT_BACKOFF) rather than a short one that expires before Garmin's own throttle clears. The cooldown is persisted so every gunicorn worker and a restart agree on it. """ now = datetime.datetime.utcnow() until = now + RATE_LIMIT_BACKOFF _rate_limited_until[user_id] = until try: # Persist alongside the current status so the cooldown survives across # workers and restarts. cur_status = (get_sync_status(user_id) or {}).get("status") or "idle" _set_sync_status( user_id, cur_status, now.isoformat(timespec="seconds"), rate_limited_until=until.isoformat(timespec="seconds"), ) except Exception as e: # no cover - surfaced so a persist failure is visible logging.getLogger(__name__).warning( "rate-limit cooldown failed to persist for %s: %s", user_id, e ) def _clear_rate_limit(user_id): """Drop a recorded cooldown after a sync that actually succeeded. A live success is proof Garmin stopped throttling, so the estimate has served its purpose; leaving a future deadline behind would just mislead the next diagnosis. """ _rate_limited_until.pop(user_id, None) try: execute( "UPDATE sync_status SET rate_limited_until = NULL " "WHERE user_id = ?", [user_id], ) except Exception: # noqa: BLE001 - clearing is best-effort pass def _rate_limit_block(user_id): """Human-readable recovery estimate from the *recorded* cooldown. Only meaningful right after a real 429 wrote a fresh cooldown (mid-run stand-down). It is not a pre-request gate — nothing consults the cooldown before calling Garmin any more. """ until = rate_limited_until(user_id) if not until or datetime.datetime.utcnow() >= until: return None, None mins = max(1, int((until - datetime.datetime.utcnow()).total_seconds() // 60)) return until, ( f"Garmin 仍在限制请求频率，预计约 {mins} 分钟后自动恢复。" "已自动退避，请耐心等待——反复点击正是把限流撞得更深的原因，令牌本身没有失效。" ) def _is_rate_limited(e): """429 from Garmin, however it happens to be dressed. Usually it arrives as a JSONDecodeError, because garth calls .json() on a 429 whose body is the plain text "Rate limited" — the status code is gone by the time the exception reaches us, and the message is the useless "Expecting value: line 1 column 1". What survives is the body itself, on the exception's `doc` attribute, so that is where to look. """ response = getattr(e, "response", None) if response is not None and getattr(response, "status_code", None) == 429: return True # The 429 can be buried several layers down: garth/urllib3 fold the final # 429 into a RetryError whose message is 'too many 429 error responses' — # no status code survives and the words "rate limit" never appear. Walk the # cause chain and also accept the "429" marker itself. seen = set() cur = e while cur is not None and id(cur) not in seen: seen.add(id(cur)) response = getattr(cur, "response", None) if response is not None and getattr(response, "status_code", None) == 429: return True text = f"{cur} {getattr(cur, 'doc', '') or ''}" if "429" in text or "rate limit" in text.lower(): return True cur = getattr(cur, "__cause__", None) or getattr(cur, "__context__", None) return False class MFARequired(RuntimeError): """Raised when a password login needs a code this process cannot obtain.""" # garth sends a browser User-Agent, which its SSO flow needs. The data API # treats that same UA as a browser hitting it directly and answers every # request with HTTP 200 and an empty array — no error, just no data. The # official app's UA (and in fact any non-browser one) returns real data, so # the header is swapped after login, before any API call. API_USER_AGENT = "com.garmin.android.apps.connectmobile" def _use_api_user_agent(client): try: client.garth.sess.headers["User-Agent"] = API_USER_AGENT except AttributeError: pass # a stubbed client in tests has no session def _is_cn(): # Selects Garmin's China service, a separate backend with separate # accounts. This project tracks an international account. return (os.environ.get("GARMIN_IS_CN") or "").lower() in ("1", "true", "yes") def _import_garmin(): try: from garminconnect import Garmin except ImportError: raise RuntimeError( "GARMIN_LIB_MISSING: 请先运行 `pip install garminconnect` 以启用同步" ) return Garmin def load_token(user_id): row = query_one("SELECT token FROM garmin_tokens WHERE user_id = ?", [user_id]) return row["token"] if row else None def _persist_token(user_id, token, garmin_email=None): """Write the token row. Does **not** drop the cached session. Separate from `save_token` because the two callers want opposite things: a re-bind must invalidate the live session, while persisting a refreshed OAuth2 token must keep the very session that just produced it. """ cols = ["user_id", "token", "garmin_email", "updated_at"] placeholders = ", ".join(["?"] * len(cols)) if DB_TYPE == "mariadb": updates = ", ".join(f"{c}=VALUES({c})" for c in cols if c != "user_id") sql = (f"INSERT INTO garmin_tokens ({', '.join(cols)}) VALUES ({placeholders}) " f"ON DUPLICATE KEY UPDATE {updates}") else: updates = ", ".join(f"{c}=excluded.{c}" for c in cols if c != "user_id") sql = (f"INSERT INTO garmin_tokens ({', '.join(cols)}) VALUES ({placeholders}) " f"ON CONFLICT(user_id) DO UPDATE SET {updates}") params = [user_id, token, garmin_email, datetime.datetime.utcnow().isoformat(timespec="seconds")] if garmin_email is None: # A refresh must not blank the remembered email: the column is part of # the upsert, so passing None would overwrite it with NULL and the # 数据同步 page would forget which account is bound. cols_kept = [c for c in cols if c != "garmin_email"] placeholders_kept = ", ".join(["?"] * len(cols_kept)) if DB_TYPE == "mariadb": updates_kept = ", ".join( f"{c}=VALUES({c})" for c in cols_kept if c != "user_id") sql = (f"INSERT INTO garmin_tokens ({', '.join(cols_kept)}) " f"VALUES ({placeholders_kept}) " f"ON DUPLICATE KEY UPDATE {updates_kept}") else: updates_kept = ", ".join( f"{c}=excluded.{c}" for c in cols_kept if c != "user_id") sql = (f"INSERT INTO garmin_tokens ({', '.join(cols_kept)}) " f"VALUES ({placeholders_kept}) " f"ON CONFLICT(user_id) DO UPDATE SET {updates_kept}") params = [user_id, token, params[-1]] execute(sql, params) def save_token(user_id, token, garmin_email=None): """Store a token from a fresh login, and stand the old session down.""" _persist_token(user_id, token, garmin_email) # A re-bind means the old session is stale; the next call must build a # fresh one rather than keep using the session the old token minted. forget_client(user_id) def has_token(user_id): return load_token(user_id) is not None def get_remembered_email(user_id): """The Garmin email last used to sign in this account, if any. `garmin_tokens` is the live Garmin binding, so it is checked first. The fallback to `users.garmin_email` exists only for accounts that bound Garmin before this table did the remembering — that column has been unused by every path that binds a *new* account since auth-hub replaced local login, but dropping it would silently make those older accounts retype their Garmin email on every sync. """ row = query_one("SELECT garmin_email FROM garmin_tokens WHERE user_id = ?", [user_id]) if row and row.get("garmin_email"): return row["garmin_email"] row = query_one("SELECT garmin_email FROM users WHERE id = ?", [user_id]) return (row or {}).get("garmin_email") or "" def delete_token(user_id): """Forget the stored Garmin OAuth token. The next sync or login will have to re-authenticate and mint a fresh token. garmin_email on the user record is left in place so re-login only needs the password. The cached client — built from the old token's session — is dropped in the same step so a stale session can't keep being reused. """ execute("DELETE FROM garmin_tokens WHERE user_id = ?", [user_id]) forget_client(user_id) # An authenticated client, reused across requests in this process. # # Building one costs ~11s against Garmin — loading the token, refreshing the # OAuth2 grant and fetching the profile — which dwarfed the ~4s of actual data # fetching behind an activity-detail request. The session is a requests.Session # underneath, so it is reusable; it is dropped after CLIENT_TTL so a refreshed # or revoked token is picked up rather than being cached indefinitely. CLIENT_TTL_SECONDS = 900 _clients = {} _clients_lock = threading.Lock() # Held only around an OAuth2 refresh — see _connect. _refresh_lock = threading.Lock() def _cached_client(user_id): entry = _clients.get(user_id) if entry and (datetime.datetime.utcnow() - entry[1]).total_seconds() < CLIENT_TTL_SECONDS: return entry[0] return None def _cache_client(user_id, client): if user_id: _clients[user_id] = (client, datetime.datetime.utcnow()) def forget_client(user_id): """Drop the cached session — call after re-binding an account.""" _clients.pop(user_id, None) def _connect(creds, user_id=None): """Obtain a logged-in Garmin client. Prefers stored OAuth tokens: an account with two-factor auth cannot be logged into from a web worker, because the library asks for the code on stdin and there is none (the failure surfaces as "EOFError: EOF when reading a line"). Tokens are minted once by `garmin_login.py`, which runs in a terminal where a code can be typed. """ if user_id: with _clients_lock: cached = _cached_client(user_id) if cached is not None: return cached Garmin = _import_garmin() client = Garmin(is_cn=_is_cn()) # Garmin SSO can be slow from some networks; the default 10s timeout in # the underlying garth library is too tight for the initial login. client.garth.configure(timeout=30) token = load_token(user_id) if user_id else None if token: client.garth.loads(token) _use_api_user_agent(client) # Only when it has actually expired. Refreshing on every connect spends # quota for nothing, and that is what walked the account into a 429. oauth2 = getattr(client.garth, "oauth2_token", None) if oauth2 is None or getattr(oauth2, "expired", True): # One refresh at a time per process. Two threads reaching a cold # client cache together would otherwise each mint a token against # the endpoint that rate-limits hardest. with _refresh_lock: oauth2 = getattr(client.garth, "oauth2_token", None) stored = load_token(user_id) if user_id else None if stored and stored != token: # Another thread refreshed while we waited: adopt its # token instead of spending an SSO call of our own. client.garth.loads(stored) oauth2 = getattr(client.garth, "oauth2_token", None) if oauth2 is None or getattr(oauth2, "expired", True): try: client.garth.refresh_oauth2() except Exception as e: # noqa: BLE001 - re-raised, named better if _is_rate_limited(e): _note_rate_limit(user_id) raise RateLimited( "Garmin 暂时限制了请求频率。这通常是短时间内连接过于频繁，" "等待约半小时后会自动恢复，令牌本身没有失效。" ) from e raise # Keep it. This line is the whole reason the account kept # getting throttled: the refreshed OAuth2 token used to # live only in this process's memory, so every cold client # cache — a 15-minute timeout, a second gunicorn worker, a # deploy restart — re-read the same expired token from the # database and minted another one against Garmin's SSO # endpoint. That endpoint limits per account and blocks for # hours. Persisting it turns "an SSO call per connect" into # "one per token lifetime" (about an hour). if user_id: try: _persist_token(user_id, client.garth.dumps()) except Exception as e: # noqa: BLE001 - never fail a sync logging.getLogger(__name__).warning( "refreshed Garmin token failed to persist " "for %s: %s", user_id, e ) # garminconnect builds most of its URLs from display_name, so leaving # it unset sends every request to ".../None". client.display_name = client.garth.profile["displayName"] with _clients_lock: _cache_client(user_id, client) return client if not creds.get("garminPassword"): raise RuntimeError("缺少 Garmin 密码，且未找到已保存的登录令牌") client.username = creds["garminEmail"] client.password = creds["garminPassword"] try: client.login() except EOFError as e: # garth's default MFA prompt calls input(); under gunicorn stdin is # closed, so it raises EOFError rather than anything descriptive. raise MFARequired( "该 Garmin 账号开启了两步验证。请在「数据同步」页面用密码重新绑定，" "系统会提示你输入验证码。" ) from e _use_api_user_agent(client) with _clients_lock: _cache_client(user_id, client) return client def describe(e): """A message that is never empty. Some exceptions carry no text at all — a bare `assert` raises AssertionError with str(e) == "" — and storing that produced a failed sync whose recorded reason was blank, which is undiagnosable. """ text = str(e).strip() return f"{type(e).__name__}: {text}" if text else type(e).__name__ def _num(*values): """First value that is a usable number.""" for v in values: if isinstance(v, (int, float)) and not isinstance(v, bool): return v return None def _safe(fn, default=None): """Call an optional endpoint; a metric the device does not record must not abort the whole day.""" try: return fn() except Exception: return default def _first(seq): return seq[0] if isinstance(seq, list) and seq else {} def _to_datetime(*values): """Normalise Garmin's several timestamp shapes into an ISO string. The same payload mixes ISO strings ("2019-10-13T10:10:12.0") with epoch milliseconds (1570961412000); handing the latter to a DATETIME column is rejected outright, so a personal record whose only timestamp was numeric failed the whole batch. """ for v in values: if v is None or v == "": continue if isinstance(v, str): return v[:26] if isinstance(v, (int, float)) and not isinstance(v, bool): # Values past ~1e11 are milliseconds, below that seconds. seconds = v / 1000 if v > 1e11 else v try: return datetime.datetime.utcfromtimestamp(seconds).isoformat( timespec="seconds" ) except (ValueError, OverflowError, OSError): continue return None def _extract_daily(client, date_str): """Everything Garmin exposes for one day. The daily summary is the bulk of it, but sleep, HRV, training readiness and endurance each live behind their own endpoint — none of them appear in get_user_summary. Each is fetched defensively so a metric this device does not record leaves a NULL instead of failing the day. """ s = client.get_user_summary(date_str) or {} sleep_dto = (_safe(lambda: client.get_sleep_data(date_str)) or {}).get( "dailySleepDTO" ) or {} scores = sleep_dto.get("sleepScores") if isinstance( sleep_dto.get("sleepScores"), dict ) else {} sleep_seconds = _num(sleep_dto.get("sleepTimeSeconds")) hrv_summary = (_safe(lambda: client.get_hrv_data(date_str)) or {}).get( "hrvSummary" ) or {} readiness = _first(_safe(lambda: client.get_training_readiness(date_str), [])) training = _safe(lambda: client.get_training_status(date_str), {}) or {} vo2 = (training.get("mostRecentVO2Max") or {}).get("generic") or {} endurance = _safe(lambda: client.get_endurance_score(date_str), {}) or {} def secs(key): return _num(sleep_dto.get(key)) return { "date": date_str, # --- activity / energy --- "steps": _num(s.get("totalSteps")), "stepGoal": _num(s.get("dailyStepGoal")), "distanceMeters": _num(s.get("totalDistanceMeters")), "caloriesBurned": _num(s.get("totalKilocalories")), "activeCalories": _num(s.get("activeKilocalories")), "bmrCalories": _num(s.get("bmrKilocalories")), "floorsAscended": _num(s.get("floorsAscended")), "floorsDescended": _num(s.get("floorsDescended")), "intensityMinutes": ( (_num(s.get("moderateIntensityMinutes")) or 0) + (_num(s.get("vigorousIntensityMinutes")) or 0) ) or None, "sedentarySeconds": _num(s.get("sedentarySeconds")), "activeSeconds": _num(s.get("activeSeconds")), # --- heart / stress --- "heartRate": _num(s.get("restingHeartRate"), s.get("averageHeartRate")), "heartRateMax": _num(s.get("maxHeartRate")), "heartRateMin": _num(s.get("minHeartRate")), "heartRateVariability": _num( hrv_summary.get("lastNightAvg"), hrv_summary.get("weeklyAvg") ), "stress": _num(s.get("averageStressLevel")), "stressMax": _num(s.get("maxStressLevel")), # --- body battery --- "bodyBatteryHigh": _num(s.get("bodyBatteryHighestValue")), "bodyBatteryLow": _num(s.get("bodyBatteryLowestValue")), "bodyBatteryCharged": _num(s.get("bodyBatteryChargedValue")), "bodyBatteryDrained": _num(s.get("bodyBatteryDrainedValue")), # --- breathing / blood oxygen --- "spo2Avg": _num(s.get("averageSpo2")), "spo2Min": _num(s.get("lowestSpo2")), "respirationAvg": _num( s.get("avgWakingRespirationValue"), s.get("latestRespirationValue") ), "respirationMin": _num(s.get("lowestRespirationValue")), "respirationMax": _num(s.get("highestRespirationValue")), # --- sleep --- "sleepDuration": round(sleep_seconds / 3600, 1) if sleep_seconds else None, "sleepQuality": _num((scores.get("overall") or {}).get("value")), "sleepDeepSeconds": secs("deepSleepSeconds"), "sleepLightSeconds": secs("lightSleepSeconds"), "sleepRemSeconds": secs("remSleepSeconds"), "sleepAwakeSeconds": secs("awakeSleepSeconds"), "sleepSpo2Avg": secs("averageSpO2Value"), "sleepRespirationAvg": secs("averageRespirationValue"), "sleepStressAvg": secs("avgSleepStress"), # --- training --- "trainingReadiness": _num(readiness.get("score")), "vo2max": _num(vo2.get("vo2MaxValue")), "enduranceScore": _num(endurance.get("overallScore")), } def sync_badges(client, user_id): """Earned badges. Keyed by Garmin's badge id, so re-syncing updates.""" badges = _safe(lambda: client.get_earned_badges(), []) or [] stored = 0 for b in badges: bid = b.get("badgeId") if bid is None: continue health.upsert_badge(user_id, { "id": str(bid), "badgeKey": b.get("badgeKey"), "name": b.get("badgeName"), "categoryId": _num(b.get("badgeCategoryId")), "difficultyId": _num(b.get("badgeDifficultyId")), "earnedDate": _to_datetime(b.get("badgeEarnedDate")), "earnedCount": _num(b.get("badgeEarnedNumber")), "points": _num(b.get("badgePoints")), }) stored += 1 return stored def sync_personal_records(client, user_id): records = _safe(lambda: client.get_personal_record(), []) or [] stored = 0 for r in records: rid = r.get("id") if rid is None: continue health.upsert_personal_record(user_id, { "id": str(rid), "typeId": _num(r.get("typeId")), "activityId": r.get("activityId"), "activityName": r.get("activityName"), "activityType": r.get("activityType"), "value": _num(r.get("value")), # Prefer the pre-formatted strings; the bare fields are epoch ms. "achievedAt": _to_datetime( r.get("prStartTimeLocalFormatted"), r.get("prStartTimeGmtFormatted"), r.get("activityStartDateTimeLocalFormatted"), r.get("prStartTimeLocal"), r.get("prStartTimeGmt"), ), }) stored += 1 return stored def _activity_end(start, duration_seconds): if not start or not duration_seconds: return start for fmt in ("%Y-%m-%dT%H:%M:%S", "%Y-%m-%d %H:%M:%S", "%Y-%m-%dT%H:%M:%S.%f"): try: dt = datetime.datetime.strptime(start[:26], fmt) return (dt + datetime.timedelta(seconds=duration_seconds)).isoformat() except ValueError: continue return start def _sync_activities(client, user_id, start_date, end_date): """Fetch the window's activities in one call and store the new ones.""" activities = client.get_activities_by_date(start_date, end_date) or [] stored = 0 for a in activities: start = a.get("startTimeLocal") or a.get("startTime") activity_type = ( (a.get("activityType") or {}).get("typeKey") if isinstance(a.get("activityType"), dict) else a.get("activityType") ) or "unknown" duration = _num(a.get("duration")) # Garmin activity ids are stable, so re-syncing a window must not # duplicate what is already stored. garmin_id = a.get("activityId") if garmin_id is not None: existing = query_one( "SELECT id FROM activities WHERE user_id = ? AND id = ?", [user_id, str(garmin_id)], ) if existing: continue health.insert_activity( user_id, { "id": str(garmin_id) if garmin_id is not None else None, "activityType": activity_type, "startTime": start, "endTime": _activity_end(start, duration), "duration": duration, "distance": _num(a.get("distance")), "calories": _num(a.get("calories")), "heartRateAverage": _num(a.get("averageHR")), "heartRateMax": _num(a.get("maxHR")), }, ) stored += 1 return stored # --- one activity, in full --------------------------------------------------- # Garmin will return thousands of samples per activity. A phone chart cannot # draw more than a few hundred usefully, and the payload is stored as a row, so # the series are thinned on the way in rather than on every read. DETAIL_MAX_POINTS = 300 # Descriptor key -> the name the UI charts by. Anything not listed is dropped: # the full descriptor set runs to dozens of fields, most of them empty. SERIES_KEYS = { "directTimestamp": "timestamp", "sumElapsedDuration": "elapsed", "sumDuration": "duration", "sumDistance": "distance", "directHeartRate": "heartRate", "directSpeed": "speed", "directElevation": "elevation", "directRunCadence": "cadence", "directBikeCadence": "cadence", "directDoubleCadence": "cadence", "directPower": "power", "directAirTemperature": "temperature", } def _thin(values, limit=DETAIL_MAX_POINTS): """Evenly sample a list down to `limit` points, keeping first and last.""" if len(values) <= limit: return values step = (len(values) - 1) / (limit - 1) return [values[int(round(i * step))] for i in range(limit)] def _series_from_details(details): """Turn Garmin's column-store detail payload into per-metric arrays. The response is a descriptor list plus rows of parallel values, so every metric has to be read out by the index its descriptor names. """ descriptors = details.get("metricDescriptors") or [] rows = details.get("activityDetailMetrics") or [] if not descriptors or not rows: return {} index = {} for d in descriptors: name = SERIES_KEYS.get(d.get("key")) if name and name not in index: index[name] = d.get("metricsIndex") rows = _thin(rows) out = {} for name, position in index.items(): if position is None: continue column = [] for row in rows: metrics = row.get("metrics") or [] column.append(metrics[position] if position < len(metrics) else None) # A column of nothing but nulls is a sensor the watch does not have. if any(v is not None for v in column): out[name] = column return out def _lap_rows(splits): laps = [] for i, lap in enumerate((splits or {}).get("lapDTOs") or [], start=1): laps.append({ "index": lap.get("lapIndex") or i, "duration": _num(lap.get("duration")), "movingDuration": _num(lap.get("movingDuration")), "distance": _num(lap.get("distance")), "averageSpeed": _num(lap.get("averageSpeed")), "maxSpeed": _num(lap.get("maxSpeed")), "calories": _num(lap.get("calories")), "averageHR": _num(lap.get("averageHR")), "maxHR": _num(lap.get("maxHR")), "elevationGain": _num(lap.get("elevationGain")), "elevationLoss": _num(lap.get("elevationLoss")), }) return laps def _hr_zones(zones): out = [] for z in zones or []: out.append({ "zone": z.get("zoneNumber"), "seconds": _num(z.get("secsInZone")) or 0, "lowBoundary": _num(z.get("zoneLowBoundary")), }) return sorted(out, key=lambda z: z.get("zone") or 0) def _build_detail(client, activity_id): """Assemble everything Garmin knows about one activity. Each call is wrapped: a watch without a barometer has no weather, a treadmill run has no gear, and a missing optional endpoint must leave the rest of the page intact rather than fail the request. """ summary = _safe(lambda: client.get_activity_evaluation(activity_id), {}) or {} # 500 is already more samples than the 300 we keep, and asking for 2000 # triples the payload for points that get thinned away anyway. details = _safe( lambda: client.get_activity_details(activity_id, maxchart=500, maxpoly=0), {} ) or {} return { "activityId": str(activity_id), "summary": summary.get("summaryDTO") or {}, "activityName": summary.get("activityName"), "activityType": (summary.get("activityTypeDTO") or {}).get("typeKey"), "eventType": (summary.get("eventTypeDTO") or {}).get("typeKey"), "laps": _lap_rows(_safe(lambda: client.get_activity_splits(activity_id), {})), "hrZones": _hr_zones( _safe(lambda: client.get_activity_hr_in_timezones(activity_id), []) ), "weather": _safe(lambda: client.get_activity_weather(activity_id), {}) or {}, "gear": _safe(lambda: client.get_activity_gear(activity_id), []) or [], "exerciseSets": ( _safe(lambda: client.get_activity_exercise_sets(activity_id), {}) or {} ).get("exerciseSets") or [], "series": _series_from_details(details), } def _store_detail(user_id, activity_id, detail): cols = ["activity_id", "user_id", "payload", "fetched_at"] values = [activity_id, user_id, json.dumps(detail, default=str), datetime.datetime.utcnow().isoformat(timespec="seconds")] placeholders = ", ".join(["?"] * len(cols)) if DB_TYPE == "mariadb": updates = ", ".join(f"{c}=VALUES({c})" for c in cols if c != "activity_id") sql = (f"INSERT INTO activity_details ({', '.join(cols)}) " f"VALUES ({placeholders}) ON DUPLICATE KEY UPDATE {updates}") else: updates = ", ".join(f"{c}=excluded.{c}" for c in cols if c != "activity_id") sql = (f"INSERT INTO activity_details ({', '.join(cols)}) VALUES " f"({placeholders}) ON CONFLICT(activity_id) DO UPDATE SET {updates}") execute(sql, values) def read_activity_detail(user_id, activity_id): """The stored detail for one activity, or None if it was never synced.""" row = query_one( "SELECT payload FROM activity_details WHERE user_id = ? AND activity_id = ?", [user_id, str(activity_id)], ) if not row or not row.get("payload"): return None try: return json.loads(row["payload"]) except ValueError: # A truncated row is worth re-syncing, not worth crashing on. return None def sync_activity_details(client, user_id, limit=None, on_progress=None): """Fetch and store the full detail for activities that lack one. Detail used to be fetched when the user opened an activity, which meant seven Garmin calls on the critical path of a tap: slow at best, and a timeout whenever the link was poor. It belongs in the sync, so the screen only ever reads the local database. """ rows = query_all( "SELECT a.id FROM activities a " "LEFT JOIN activity_details d ON d.activity_id = a.id " "WHERE a.user_id = ? AND d.activity_id IS NULL " "ORDER BY a.start_time DESC", [user_id], ) if limit: rows = rows[:limit] stored = 0 for i, row in enumerate(rows): try: _store_detail(user_id, row["id"], _build_detail(client, row["id"])) stored += 1 except Exception: # noqa: BLE001 - one bad activity must not stop the rest continue if on_progress: on_progress(i + 1, len(rows)) return stored _backfill_progress = {} def backfill_status(user_id): """Progress of the historical backfill for this account.""" return _backfill_progress.get(user_id) or { "running": False, "stage": None, "done": 0, "total": 0, "error": None, } def _set_backfill(user_id, **fields): state = dict(_backfill_progress.get(user_id) or {}) state.update(fields) _backfill_progress[user_id] = state def days_missing_series(user_id, limit=None): """Days that have a health row but no within-day curves stored.""" rows = query_all( "SELECT h.date FROM health_data h " "LEFT JOIN daily_series s ON s.user_id = h.user_id AND s.date = h.date " "WHERE h.user_id = ? AND s.date IS NULL " "GROUP BY h.date ORDER BY h.date DESC", [user_id], ) dates = [str(r["date"])[:10] for r in rows] return dates[:limit] if limit else dates def start_backfill(user_id, limit=None): """Fill in everything the per-day sync leaves out, in the background. Two long jobs share one runner because they share a cause — an account whose history predates these features — and because the user should press one button, not two. Each activity costs several Garmin calls and each day of curves costs five, so this runs for minutes; the UI polls. """ state = _backfill_progress.get(user_id) if state and state.get("running"): return state _set_backfill(user_id, running=True, stage="启动中", done=0, total=0, error=None) def run(): try: client = _connect({}, user_id=user_id) _set_backfill(user_id, stage="运动详情", done=0, total=0) sync_activity_details( client, user_id, limit, on_progress=lambda d, n: _set_backfill( user_id, stage="运动详情", done=d, total=n), ) dates = days_missing_series(user_id, limit) _set_backfill(user_id, stage="每日曲线", done=0, total=len(dates)) for i, date in enumerate(dates): try: extras.sync_daily_series(client, user_id, date) except Exception: # noqa: BLE001 - one day must not stop the rest pass _set_backfill(user_id, stage="每日曲线", done=i + 1, total=len(dates)) _set_backfill(user_id, running=False, stage="完成", error=None) except Exception as e: # noqa: BLE001 - reported through the status endpoint _set_backfill(user_id, running=False, stage=None, error=describe(e)) threading.Thread(target=run, daemon=True, name=f"backfill-{user_id}").start() return backfill_status(user_id) # What 全部历史 actually means. 730 was a made-up ceiling that silently # truncated anyone with more than two years of Garmin history to two years. # The real end of the data is found by walking back until the days stop # containing anything (EMPTY_RUN_STOP below), so this is only a backstop. MAX_HISTORY_DAYS = int(os.environ.get("GARMIN_MAX_HISTORY_DAYS") or 3650) # A long backfill stops once this many consecutive fetched days come back with # nothing at all: that is what reaching the start of the account looks like, # and without it 全部历史 would spend thousands of requests on years that # predate the watch. Long enough to ride out a season of not wearing it. EMPTY_RUN_STOP = 120 # The most recent days are still being written to, so a backfill refetches # them even when they are already stored. Everything older is skipped if it is # already in the database — that is what makes a multi-year sync resumable # after a rate limit instead of restarting from today every time. ALWAYS_REFETCH_DAYS = 3 # Up to this many days, a sync also pulls each day's within-day curves inline. # Beyond it the curves are left to the background backfill: five extra calls # per day would turn a year's sync into an hour. SERIES_INLINE_DAYS = 14 # Activity details fetched per sync run. One request each, and a fresh account # has hundreds — fetching them all in one run is the single biggest burst a # sync can produce. The rest are picked up by the next run and by # start_backfill, so nothing is lost by spreading them out. DETAILS_PER_SYNC = int(os.environ.get("GARMIN_DETAILS_PER_SYNC") or 40) # Above this many days a sync is long enough that the caller must not block # on it — a year takes roughly 20 minutes at ~3s per day. BACKGROUND_THRESHOLD_DAYS = 14 def start_sync(user_id, creds, days=None): """Run a sync in the background and return immediately. Progress lands in sync_status, which the UI polls; a full backfill runs far longer than any sensible HTTP timeout. No local rate-limit estimate is consulted here: every manual start issues a real request and trusts Garmin's live answer. A real 429 writes the cooldown and the run stands down inside sync_data. """ days = DEFAULT_SYNC_DAYS if days is None else days now = datetime.datetime.utcnow().isoformat(timespec="seconds") _set_sync_status( user_id, "syncing", now, records_synced=0, progress_current=0, progress_total=days, started_at=now, last_error=None, ) thread = threading.Thread( target=sync_data, args=(user_id, creds, days), daemon=True ) thread.start() return {"status": "syncing", "days": days} def sync_data(user_id, creds, days=None, client=None, trigger="manual"): """Pull the last `days` days from Garmin Connect into the local database. `client` exists so tests can inject a stub instead of reaching Garmin. `trigger` ("auto" / "manual" / "quick") names what started this run and is recorded on the sync_history row every exit path appends. """ days = DEFAULT_SYNC_DAYS if days is None else days if days == 0: days = MAX_HISTORY_DAYS # 全部历史 → 走到数据尽头为止 now = datetime.datetime.utcnow().isoformat(timespec="seconds") started = now def finish(result): """Record the attempt, then hand the result to the caller.""" _log_sync_history(user_id, trigger, days, started, result) return result # No local cooldown gate here (2026-09-02): the recorded rate_limited_until # is an estimate, and refusing on it kept accounts idle after Garmin had # already recovered. Every run issues a real request; only a real 429 # stands the run down (mid-run, below) and writes a fresh cooldown. _set_sync_status( user_id, "syncing", now, records_synced=0, progress_current=0, progress_total=days, stage="连接 Garmin", ) try: client = client or _connect(creds, user_id) except RateLimited as e: # A genuine 429 at connect time: _connect already wrote the fresh # cooldown, so surface this as rate_limited (the UI draws a distinct # stand-down state for it) rather than a generic error. message = str(e) _set_sync_status(user_id, "rate_limited", now, records_synced=0, last_error=message) return finish({ "status": "rate_limited", "recordsSynced": 0, "message": message, "mfaRequired": False, "lastSyncTime": now, }) except Exception as e: message = describe(e) _set_sync_status(user_id, "error", now, records_synced=0, last_error=message) return finish({ "status": "error", "recordsSynced": 0, "message": message, "mfaRequired": isinstance(e, MFARequired), "lastSyncTime": now, }) # Every Garmin call from here on is spaced and counted. Applied here # rather than in `_connect` so an injected client is paced identically — # the pacing is a property of a sync run, not of how the session was # obtained. See services/garmin_throttle.py for why this is necessary at # all: a day of history costs nine requests, and a full backfill used to # fire six thousand of them flat out. client = throttle.pace(client, user_id) today = datetime.date.today() # -1 means "incremental sync": pick up from the latest date already in the # local database rather than pulling a fixed window. if days == -1: # The daily rows live in `health_data`; `health_daily` is the name of # the *writer* (health.upsert_health_daily), not of any table. Reading # it raised inside the background thread, so an incremental sync died # silently and left the status stuck on "syncing". row = query_one( "SELECT MAX(date) AS latest FROM health_data WHERE user_id = ?", (user_id,) ) latest = row["latest"] if row and row.get("latest") else None if latest is None: days = DEFAULT_SYNC_DAYS # first sync → fall back to default window else: days = max(1, (today - datetime.date.fromisoformat(latest)).days) # Rewrite the progress total now that we know the real day count. _set_sync_status( user_id, "syncing", now, records_synced=0, progress_current=0, progress_total=days, stage="连接 Garmin", ) start_date = (today - datetime.timedelta(days=days - 1)).isoformat() # Days already stored, so a resumed backfill does not spend its whole # rate-limit budget re-fetching what it already has. stored = { r["date"] for r in query_all( "SELECT date FROM health_data WHERE user_id = ? AND date >= ?", (user_id, start_date), ) } days_synced = 0 days_attempted = 0 empty_run = 0 reached_the_start = False day_errors = [] rate_limited_mid_run = None budget_spent = False for i in range(days): date_str = (today - datetime.timedelta(days=i)).isoformat() if i >= ALWAYS_REFETCH_DAYS and date_str in stored: continue days_attempted += 1 try: record = _extract_daily(client, date_str) record.update(extras.daily_extras(client, date_str)) except throttle.BudgetExhausted: # Not a failure: this run has issued as many requests as it is # allowed. Keep what is stored and stop — the next run skips the # days already written and carries on from here. budget_spent = True break except Exception as e: # A 429 inside the day loop used to be filed as one more skipped # day, so a 730-day backfill kept hammering Garmin for another # 700 days and drove the throttle deeper. Stand down at the first # one and keep whatever was already stored. if isinstance(e, RateLimited) or _is_rate_limited(e): _note_rate_limit(user_id) rate_limited_mid_run = describe(e) break day_errors.append(f"{date_str}: {describe(e)}") continue # A day Garmin has no data for comes back all-None; storing it would # create an empty row that the metric endpoints then have to filter. if any(record[k] is not None for k in record if k != "date"): health.upsert_health_daily(user_id, record) days_synced += 1 empty_run = 0 # Within-day curves for short syncs only. A year-long backfill # would add five calls per day on top of everything else; those # days are filled by start_backfill instead. if days <= SERIES_INLINE_DAYS: try: extras.sync_daily_series(client, user_id, date_str) except Exception as e: # noqa: BLE001 day_errors.append(f"{date_str} series: {describe(e)}") else: # Walked back past the start of the account. Without this, 全部历史 # would keep asking Garmin about years before the watch existed. empty_run += 1 if days > 90 and empty_run >= EMPTY_RUN_STOP: reached_the_start = True break # A long backfill reports every fifth day — the write is cheap but not # free. A short one reports every day: at 7 days a "every 5th" cadence # meant the bar sat at 0 for most of the run and then vanished. if days <= 30 or (i + 1) % 5 == 0 or i + 1 == days: _set_sync_status( user_id, "syncing", now, records_synced=days_synced, progress_current=i + 1, progress_total=days, stage=f"每日数据 {date_str}", ) if rate_limited_mid_run: until, msg = _rate_limit_block(user_id) message = msg or "Garmin 限制了请求频率，已自动退避。" _set_sync_status( user_id, "rate_limited", now, records_synced=days_synced, progress_current=days_synced, progress_total=days, stage=None, last_error=message, ) return finish({ "status": "rate_limited", "recordsSynced": days_synced, "message": f"已同步 {days_synced} 天后被 Garmin 限流：{message}", "retryAfterSeconds": ( int((until - datetime.datetime.utcnow()).total_seconds()) if until else None ), "lastSyncTime": now, }) activities_synced = 0 _set_sync_status(user_id, "syncing", now, records_synced=days_synced, progress_current=days, progress_total=days, stage="运动记录") try: if not budget_spent: activities_synced = _sync_activities( client, user_id, start_date, today.isoformat() ) except throttle.BudgetExhausted: budget_spent = True except Exception as e: day_errors.append(f"activities: {describe(e)}") # Full detail for any activity that does not have it stored yet, so that # opening one later is a local read. details_synced = 0 try: if not budget_spent: # Capped per run. On a fresh account this is one request for each # of a couple of hundred activities; unbounded, it spends the whole # budget (and used to spend the account's goodwill) in one burst. details_synced = sync_activity_details( client, user_id, limit=DETAILS_PER_SYNC, on_progress=lambda d, n: _set_sync_status( user_id, "syncing", now, records_synced=days_synced, progress_current=days, progress_total=days, stage=f"运动详情 {d}/{n}"), ) except throttle.BudgetExhausted: budget_spent = True except Exception as e: day_errors.append(f"activity_details: {describe(e)}") # Everything else Garmin holds: body composition, blood pressure, race # predictions, challenges and devices. Account-wide, so once per sync. extra_counts = {} stage_names = { "bodyComposition": "身体成分", "bloodPressure": "血压", "racePredictions": "成绩预测", "challenges": "挑战赛", "devices": "设备", } for name, call in ( ("bodyComposition", lambda: extras.sync_body_composition(client, user_id, start_date, today.isoformat())), ("bloodPressure", lambda: extras.sync_blood_pressure(client, user_id, start_date, today.isoformat())), ("racePredictions", lambda: extras.sync_race_predictions(client, user_id)), ("challenges", lambda: extras.sync_challenges(client, user_id)), ("devices", lambda: extras.sync_devices(client, user_id)), ): _set_sync_status(user_id, "syncing", now, records_synced=days_synced, progress_current=days, progress_total=days, stage=stage_names.get(name, name)) try: extra_counts[name] = call() except Exception as e: # noqa: BLE001 - one section must not fail the sync day_errors.append(f"{name}: {describe(e)}") # Badges and personal records are account-wide rather than per-day, so # they are fetched once per sync rather than inside the day loop. badges_synced = 0 records_synced_pr = 0 try: badges_synced = sync_badges(client, user_id) except Exception as e: day_errors.append(f"badges: {describe(e)}") try: records_synced_pr = sync_personal_records(client, user_id) except Exception as e: day_errors.append(f"personal_records: {describe(e)}") # Every single day failing means something systemic (expired session, # API change) — reporting that as a clean success would hide it. if days_synced == 0 and days_attempted > 0 and len(day_errors) >= days_attempted: message = "; ".join(day_errors[:3]) _set_sync_status(user_id, "error", now, records_synced=0, last_error=message) return finish({"status": "error", "recordsSynced": 0, "message": f"同步失败：{message}", "lastSyncTime": now}) _set_sync_status( user_id, "idle", now, records_synced=days_synced, progress_current=days, progress_total=days, stage=None, last_error="; ".join(day_errors[:3]) if day_errors else None, ) # A live success proves Garmin stopped throttling — retire any recorded # cooldown so it cannot mislead a later diagnosis. _clear_rate_limit(user_id) message = ( f"同步完成，更新 {days_synced} 天数据、{activities_synced} 条运动记录" f"（含 {details_synced} 条详情）、" f"{badges_synced} 个奖励、{records_synced_pr} 项个人纪录" ) if day_errors: message += f"（{len(day_errors)} 项跳过）" if budget_spent: # Said plainly: an unexplained "only 120 of 730 days" reads as a bug, # and the next run picking up where this one stopped is the design. message += "；本次已达请求预算上限，剩余部分下次同步继续" requests_used = getattr(getattr(client, "limiter", None), "used", None) return finish({ "status": "success", "recordsSynced": days_synced, "activitiesSynced": activities_synced, "badgesSynced": badges_synced, "personalRecordsSynced": records_synced_pr, "requestsUsed": requests_used, "partial": budget_spent, "message": message, "lastSyncTime": now, })