Download your Personal Strava Activities

download_strava_activities.py

from pathlib import Path
from typing import Any

import click
import pandas as pd
import requests
from requests import Session

# Strava API endpoints
OAUTH_BASE_URL = "https://www.strava.com/oauth"
TOKEN_URL = f"{OAUTH_BASE_URL}/token"
AUTH_URL = f"{OAUTH_BASE_URL}/authorize"
BASE_URL = "https://www.strava.com/api/v3"
CALLBACK_DOMAIN = "http://localhost"


def create_session(client_id: str, client_secret: str, refresh_token: str) -> Session:
    """Create a session with the Strava API."""
    response = requests.post(
        TOKEN_URL,
        data={
            "client_id": client_id,
            "client_secret": client_secret,
            "refresh_token": refresh_token,
            "grant_type": "refresh_token",
        },
    )

    try:
        access_token = response.json()["access_token"]
        session = Session()
        session.headers.update({"Authorization": f"Bearer {access_token}"})
        return session
    except KeyError as e:
        raise RuntimeError(f"Error refreshing token: {response.text}") from e


def create_scoped_session(client_id: str, client_secret: str) -> Session:
    """Create a scoped session with the Strava API."""
    res = requests.get(
        AUTH_URL,
        params={
            "client_id": client_id,
            "redirect_uri": CALLBACK_DOMAIN,
            "scope": ["activity:read_all"],
            "response_type": "code",
        },
    )
    url = res.history[0].url if res.history else res.url
    code = input(
        f"VISIT THIS URL, LOG IN, AND GRANT ACCESS: {url}\nEnter code from redirected URL: "
    )

    response = requests.post(
        TOKEN_URL,
        data={
            "client_id": client_id,
            "client_secret": client_secret,
            "code": code,
            "grant_type": "authorization_code",
        },
    )
    access_token = response.json()["access_token"]
    session = Session()
    session.headers.update({"Authorization": f"Bearer {access_token}"})
    return session


def get_all_activities(scoped_session: Session) -> list[dict[str, Any]]:
    """Fetch all activities from Strava API."""
    activities = []
    page = 1
    print("Fetching activities...")
    while True:
        response = scoped_session.get(
            f"{BASE_URL}/athlete/activities", params={"page": page, "per_page": 50}
        )
        if response.status_code != 200:
            print(f"Error fetching activities: {response.text}")
            break
        data = response.json()
        if not data:
            break
        activities.extend(data)
        print(f"Fetched {len(data)} activities. Total: {len(activities)}", end="\r")
        page += 1
    print(f"Fetched all activities. Total: {len(activities)}.")
    return activities


def get_athelete(session: Session) -> dict[str, Any]:
    """Get athlete from Strava API."""
    athlete_res = session.get(f"{BASE_URL}/athlete")
    if athlete_res.status_code != 200:
        raise ValueError(f"Error fetching athlete: {athlete_res.text}")
    return athlete_res.json()


@click.command()
@click.argument("client-id", type=str)
@click.argument("client-secret", type=str)
@click.option("--output-file", type=str, default="activities.csv")
def get_activities(client_id: str, client_secret: str, output_file: str) -> None:
    """Get activities from Strava API."""
    scoped_session = create_scoped_session(client_id, client_secret)
    activities = get_all_activities(scoped_session)

    output_path = Path(output_file).with_suffix(".csv")

    if not output_path.is_relative_to("."):
        output_path.parent.mkdir(parents=True, exist_ok=True)
    df = pd.DataFrame(activities)
    # Pop unnecessary columns (e.g. structured data or columns with only null values)
    df.drop(
        columns=[
            "resource_state",
            "athlete",
            "location_city",
            "map",
            "location_city",
            "location_state",
            "location_country",
            "start_latlng",
            "end_latlng",
            "heartrate_opt_out",
            "display_hide_heartrate_option",
            "upload_id",
            "upload_id_str",
            "total_photo_count",
        ],
        inplace=True,
    )

    df.to_csv(output_path, index=False)
    print("Saved activities to:", str(output_path))


if __name__ == "__main__":
    get_activities()