strava.py

#!/usr/bin/env python3
"""Shared Strava plumbing: .env loading, auth, and API access.

Credentials come from .env (see get_token.py) or the environment:
  STRAVA_ACCESS_TOKEN                                       # short-lived, ~6h
  STRAVA_CLIENT_ID + STRAVA_CLIENT_SECRET + STRAVA_REFRESH_TOKEN
"""

from __future__ import annotations

import json
import os
import sys
import time
import urllib.error
import urllib.parse
import urllib.request
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any

API = "https://www.strava.com/api/v3"
TOKEN_URL = "https://www.strava.com/oauth/token"
ENV_FILE = Path(__file__).resolve().parent / ".env"
RATE_RETRIES = 10

Activity = dict[str, Any]


def load_env(path: Path = ENV_FILE) -> None:
    if not path.exists():
        return
    for line in path.read_text().splitlines():
        line = line.strip().removeprefix("export ")
        if not line or line.startswith("#") or "=" not in line:
            continue
        key, _, value = line.partition("=")
        os.environ.setdefault(key.strip(), value.strip().strip("'\""))


def rate_limit_reset(headers: Any, now: datetime | None = None) -> tuple[datetime, bool]:
    """When the exceeded rate-limit window resets, and whether it is the daily one.

    Strava sends `X-RateLimit-Limit: short,daily` and `X-RateLimit-Usage: short,daily`.
    Short windows are wall-clock aligned (:00 :15 :30 :45), the daily one is UTC midnight.
    """
    now = now or datetime.now(timezone.utc)
    try:
        limits = [int(x) for x in str(headers.get("X-RateLimit-Limit", "")).split(",")]
        usage = [int(x) for x in str(headers.get("X-RateLimit-Usage", "")).split(",")]
        daily_exhausted = len(limits) > 1 and len(usage) > 1 and usage[1] >= limits[1]
    except ValueError:  # headers missing or malformed: assume the short window
        daily_exhausted = False
    if daily_exhausted:
        midnight = now.replace(hour=0, minute=0, second=0, microsecond=0)
        return midnight + timedelta(days=1), True
    quarter = now.replace(minute=0, second=0, microsecond=0)
    return quarter + timedelta(minutes=15 * (now.minute // 15 + 1)), False


def request_json(
    url: str,
    token: str | None = None,
    data: bytes | None = None,
    missing_ok: bool = False,
    content_type: str | None = None,
    errors_ok: bool = False,
    method: str | None = None,
) -> Any:
    """GET/POST JSON, sleeping through 429s until the 15-minute window resets.

    `errors_ok` returns the error body (with `_status`) instead of exiting, for callers that
    process many items and must not die on one bad one. A hit daily limit always exits.
    """
    for attempt in range(RATE_RETRIES + 1):
        req = urllib.request.Request(url, data=data, method=method)
        req.add_header("Cache-Control", "no-cache")
        if token:
            req.add_header("Authorization", f"Bearer {token}")
        if content_type:
            req.add_header("Content-Type", content_type)
        try:
            with urllib.request.urlopen(req) as resp:
                return json.load(resp)
        except urllib.error.HTTPError as e:
            if missing_ok and e.code in (401, 404):
                return None
            body = e.read().decode(errors="replace")
            if e.code == 429:
                reset, daily = rate_limit_reset(e.headers)
                if daily:
                    sys.exit(
                        "Strava daily rate limit reached. Retry after "
                        f"{reset.astimezone():%Y-%m-%d %H:%M %Z} (already-imported files are "
                        "skipped on the next run)."
                    )
                if attempt == RATE_RETRIES:
                    sys.exit(f"Still rate limited after {RATE_RETRIES} waits. Retry later.")
                wait = max(5.0, (reset - datetime.now(timezone.utc)).total_seconds() + 2)
                print(
                    f"rate limited, sleeping {wait / 60:.1f} min until "
                    f"{reset.astimezone():%H:%M %Z}",
                    file=sys.stderr,
                )
                time.sleep(wait)
                continue
            if errors_ok:
                try:
                    parsed = json.loads(body)
                except ValueError:
                    parsed = {"message": body[:400]}
                return {**parsed, "_status": e.code} if isinstance(parsed, dict) else parsed
            sys.exit(f"Strava API error {e.code}: {body[:400]}")


def exchange_tokens(client_id: str, client_secret: str, **grant: str) -> dict[str, Any]:
    """POST to the token endpoint. `grant` carries grant_type plus its code/refresh_token."""
    body = urllib.parse.urlencode(
        {"client_id": client_id, "client_secret": client_secret, **grant}
    ).encode()
    return request_json(TOKEN_URL, data=body)


def access_token() -> str:
    load_env()
    if token := os.environ.get("STRAVA_ACCESS_TOKEN"):
        return token
    client_id = os.environ.get("STRAVA_CLIENT_ID")
    client_secret = os.environ.get("STRAVA_CLIENT_SECRET")
    refresh = os.environ.get("STRAVA_REFRESH_TOKEN")
    if not (client_id and client_secret and refresh):
        sys.exit("No credentials. Run: python3 get_token.py")
    tokens = exchange_tokens(
        client_id, client_secret, grant_type="refresh_token", refresh_token=refresh
    )
    return tokens["access_token"]


def fetch_activities(token: str, after: int | None = None) -> list[Activity]:
    activities: list[Activity] = []
    page = 1
    while True:
        params: dict[str, Any] = {"per_page": 200, "page": page}
        if after:
            params["after"] = after
        batch = request_json(f"{API}/athlete/activities?" + urllib.parse.urlencode(params), token)
        if not batch:
            return activities
        activities.extend(batch)
        print(f"fetched {len(activities)} activities...", file=sys.stderr)
        page += 1