""" Strava API sync helpers. OAuth: exchange_code() / refresh_tokens() talk to Strava's token endpoint. get_valid_access_token() transparently refreshes an expired access token and persists the rotated refresh token. sync_strava_activities() lists the athlete's activities, fetches per-activity streams, builds a fit_parser-shaped dict and hands it to persist_activity() (shared with the file-upload path) so Strava activities get identical dedup, PR, route and segment handling. Tokens are Fernet-encrypted with the same SECRET_KEY scheme as Garmin creds. """ import logging from datetime import date, datetime, timedelta, timezone from typing import Optional import httpx import polyline as polyline_lib from app.services.fit_parser import _bounding_box, _active_spans from app.services.garmin_connect_sync import _fernet # reuse the SECRET_KEY-derived Fernet logger = logging.getLogger(__name__) STRAVA_AUTHORIZE_URL = "https://www.strava.com/oauth/authorize" STRAVA_TOKEN_URL = "https://www.strava.com/oauth/token" STRAVA_API_BASE = "https://www.strava.com/api/v3" # Scope needed to read all activities (including those marked "Only You"). STRAVA_SCOPE = "read,activity:read_all" # Like Garmin: incremental syncs only re-scan the last day or two for late edits. INCREMENTAL_BUFFER_DAYS = 1 # Strava activity type / sport_type → MileVault internal sport_type. Gym types map # onto the internal vocabulary so sportColor() paints them red, GPS types green/etc. STRAVA_SPORT_MAP = { "run": "running", "trailrun": "running", "virtualrun": "running", "ride": "cycling", "virtualride": "cycling", "mountainbikeride": "cycling", "gravelride": "cycling", "ebikeride": "cycling", "emountainbikeride": "cycling", "handcycle": "cycling", "velomobile": "cycling", "walk": "walking", "hike": "hiking", "swim": "swimming", "weighttraining": "strength_training", "workout": "training", "crossfit": "hiit", "hiit": "hiit", "elliptical": "fitness_equipment", "stairstepper": "fitness_equipment", "golf": "golf", "yoga": "yoga", } def encrypt_token(token: str) -> str: return _fernet().encrypt(token.encode()).decode() def decrypt_token(enc: str) -> str: return _fernet().decrypt(enc.encode()).decode() # ── OAuth ────────────────────────────────────────────────────────────────────── def _token_request(payload: dict) -> dict: from app.core.config import settings payload = { "client_id": settings.strava_client_id, "client_secret": settings.strava_client_secret, **payload, } with httpx.Client(timeout=30) as client: resp = client.post(STRAVA_TOKEN_URL, data=payload) resp.raise_for_status() return resp.json() def exchange_code(code: str) -> dict: """Exchange an authorization code for tokens. Returns the raw token dict (access_token, refresh_token, expires_at, athlete{...}).""" return _token_request({"code": code, "grant_type": "authorization_code"}) def refresh_tokens(refresh_token: str) -> dict: """Get a fresh access token (and possibly rotated refresh token).""" return _token_request({"refresh_token": refresh_token, "grant_type": "refresh_token"}) def get_valid_access_token(cfg, db) -> str: """Return a usable access token for `cfg`, refreshing and persisting if the current one is expired or about to expire. `cfg` is a StravaConfig row.""" now = datetime.now(timezone.utc) expires_at = cfg.expires_at if expires_at is not None and expires_at.tzinfo is None: expires_at = expires_at.replace(tzinfo=timezone.utc) if expires_at is None or expires_at <= now + timedelta(seconds=60): tok = refresh_tokens(decrypt_token(cfg.refresh_token_enc)) cfg.access_token_enc = encrypt_token(tok["access_token"]) cfg.refresh_token_enc = encrypt_token(tok["refresh_token"]) cfg.expires_at = datetime.fromtimestamp(tok["expires_at"], tz=timezone.utc) db.commit() return tok["access_token"] return decrypt_token(cfg.access_token_enc) # ── Activity sync ────────────────────────────────────────────────────────────── def _api_get(client: httpx.Client, token: str, path: str, **params): resp = client.get( f"{STRAVA_API_BASE}{path}", headers={"Authorization": f"Bearer {token}"}, params=params, ) if resp.status_code == 429: raise StravaRateLimited() resp.raise_for_status() return resp.json() class StravaRateLimited(Exception): """Raised when Strava returns HTTP 429 (rate limit exceeded).""" def _streams_to_points(streams: dict, start_dt: datetime) -> list: """Convert Strava's key_by_type streams into fit_parser-shaped data points.""" time_s = (streams.get("time") or {}).get("data") or [] if not time_s: return [] latlng = (streams.get("latlng") or {}).get("data") or [] altitude = (streams.get("altitude") or {}).get("data") or [] heartrate = (streams.get("heartrate") or {}).get("data") or [] cadence = (streams.get("cadence") or {}).get("data") or [] watts = (streams.get("watts") or {}).get("data") or [] velocity = (streams.get("velocity_smooth") or {}).get("data") or [] temp = (streams.get("temp") or {}).get("data") or [] distance = (streams.get("distance") or {}).get("data") or [] def at(arr, i): return arr[i] if i < len(arr) else None points = [] for i, off in enumerate(time_s): ll = at(latlng, i) lat = ll[0] if ll else None lng = ll[1] if ll else None points.append({ "timestamp": (start_dt + timedelta(seconds=off)).isoformat(), "latitude": lat, "longitude": lng, "altitude_m": at(altitude, i), "heart_rate": at(heartrate, i), "cadence": at(cadence, i), "speed_ms": at(velocity, i), "power": at(watts, i), "temperature_c": at(temp, i), "distance_m": at(distance, i), }) return points def _build_parsed(summary: dict, points: list) -> dict: """Assemble a fit_parser-shaped dict from a Strava summary + stream points.""" raw_type = (summary.get("sport_type") or summary.get("type") or "workout") sport = STRAVA_SPORT_MAP.get(str(raw_type).lower(), str(raw_type).lower()) start_str = summary.get("start_date") # UTC ISO, e.g. 2020-01-01T08:00:00Z start_dt = datetime.fromisoformat(start_str.replace("Z", "+00:00")) if start_str else None coords = [(p["latitude"], p["longitude"]) for p in points if p["latitude"] and p["longitude"]] encoded = polyline_lib.encode(coords) if coords else (summary.get("map") or {}).get("summary_polyline") # Elevation loss from the altitude stream (summary only gives gain). alts = [p["altitude_m"] for p in points if p["altitude_m"] is not None] downhill = sum(max(0.0, alts[i-1] - alts[i]) for i in range(1, len(alts))) if alts else None return { "name": summary.get("name") or f"{sport.title()} {start_dt.date() if start_dt else ''}".strip(), "sport_type": sport, "start_time": start_dt.isoformat() if start_dt else None, "distance_m": summary.get("distance"), "duration_s": summary.get("elapsed_time"), "moving_time_s": summary.get("moving_time"), "elevation_gain_m": summary.get("total_elevation_gain"), "elevation_loss_m": downhill, "avg_heart_rate": summary.get("average_heartrate"), "max_heart_rate": summary.get("max_heartrate"), "avg_cadence": summary.get("average_cadence"), "avg_power": summary.get("average_watts"), "normalized_power": summary.get("weighted_average_watts"), "avg_speed_ms": summary.get("average_speed"), "max_speed_ms": summary.get("max_speed"), "active_spans": _active_spans(points), "avg_temperature_c": summary.get("average_temp"), "calories": summary.get("calories"), "training_stress_score": None, "vo2max_estimate": None, "polyline": encoded, "bounding_box": _bounding_box(coords), "source_type": "strava", "rejected_reason": None, "data_points": points, "laps": [], } def sync_strava_activities(cfg, user_id: int, db, lookback_days: int = 30, status_callback=None) -> int: """List the athlete's activities since the appropriate window, fetch streams for each new one and persist it. Returns the number of activities imported.""" import time as _time from app.workers.tasks import persist_activity since = cfg.last_sync_at if since is not None and since.tzinfo is None: since = since.replace(tzinfo=timezone.utc) if since: after_dt = since - timedelta(days=INCREMENTAL_BUFFER_DAYS) elif lookback_days == -1: after_dt = datetime(2010, 1, 1, tzinfo=timezone.utc) else: after_dt = datetime.now(timezone.utc) - timedelta(days=max(lookback_days, 1)) after_epoch = int(after_dt.timestamp()) token = get_valid_access_token(cfg, db) imported = 0 with httpx.Client(timeout=60) as client: # 1) Page through activity summaries (newest first within each page). summaries = [] page = 1 while True: batch = _api_get(client, token, "/athlete/activities", after=after_epoch, page=page, per_page=100) if not batch: break summaries.extend(batch) if len(batch) < 100: break page += 1 _time.sleep(0.3) total = len(summaries) if status_callback: status_callback(f"Syncing activities: 0/{total} imported") # 2) For each, fetch streams and persist. Oldest-first so PRs accrue in order. for idx, summary in enumerate(sorted(summaries, key=lambda s: s.get("start_date") or "")): sid = str(summary.get("id") or "").strip() if not sid: continue points = [] try: streams = _api_get( client, token, f"/activities/{sid}/streams", keys="time,latlng,altitude,heartrate,cadence,watts,velocity_smooth,temp,distance", key_by_type="true", ) start_str = summary.get("start_date") start_dt = datetime.fromisoformat(start_str.replace("Z", "+00:00")) if start_str else None if start_dt: points = _streams_to_points(streams, start_dt) except StravaRateLimited: # Out of quota — stop here; the next scheduled sync resumes the rest. if status_callback: status_callback(f"Rate-limited by Strava — imported {imported}, will resume next sync") logger.warning("Strava rate limit hit for user %s after %d imports", user_id, imported) break except Exception as exc: logger.warning("Failed to fetch streams for Strava activity %s: %s", sid, exc) # Still persist from the summary (map polyline only, no per-point data). parsed = _build_parsed(summary, points) if not parsed.get("start_time"): continue try: result = persist_activity(db, user_id, parsed, strava_activity_id=sid) if result.get("status") == "ok": imported += 1 except Exception as exc: logger.warning("Failed to persist Strava activity %s: %s", sid, exc) db.rollback() if status_callback and (idx % 5 == 0 or idx == total - 1): status_callback(f"Syncing activities: {imported}/{total} imported") _time.sleep(0.3) # be gentle on the rate limit return imported