march_running_from_strava.py

"""
A script to fetch and analyze running activities from Strava for the current year's March.
This script uses the Strava API to:
1. Authenticate using OAuth 2.0
2. Retrieve all activities for March of the current year
3. Filter for running activities only
4. Calculate daily running distances in miles
5. Generate a table showing miles run each day in March
6. Export the results to 'march_running_activities.csv'

Then you can just copy-paste from the CSV to the google sheet tally!

Requirements:
- Strava API credentials (CLIENT_ID and CLIENT_SECRET)
- Python packages: `pip install requests pandas datetime dateutil pytz tabulate`

Usage:
1. Create a Strava API application at https://www.strava.com/settings/api
    - Category: Visualization or anything
    - Any image for the icon
    - Website: http://localhost
    - Authorization Callback Domain: localhost
2. Specify your CLIENT_ID and CLIENT_SECRET as environment variables or as 'CLIENT_ID'
    E.g. in your terminal: `export CLIENT_ID='your_client_id'`
3. Run the script:
```
python march_running_from_strava.py
```

The script will open a browser window for Strava authentication.
After authorizing the application, copy the authorization code from the URL and paste it the terminal.
The script will then fetch and process running data and save the results to 'march_running_activities.csv'.

Acknowledgements: Claude 3.7 Sonnet
"""
# %%
import os
import requests
import pandas as pd
import datetime
from dateutil import parser
import pytz
from tabulate import tabulate
import webbrowser

# OAuth 2.0 Authentication Configuration
CLIENT_ID = os.getenv('CLIENT_ID')      # Replace with your client ID
CLIENT_SECRET = os.getenv('CLIENT_SECRET')  # Replace with your client secret
# We'll generate a new refresh token with the correct permissions

# %%
def get_authorization_code(client_id: str) -> str:
    """
    Open browser to get authorization code with activity:read_all scope.
    This is a manual step that requires user interaction.
    """
    # Important: scope should be set to activity:read
    auth_url = (
        f"http://www.strava.com/oauth/authorize"
        f"?client_id={client_id}"
        f"&response_type=code"
        f"&redirect_uri=http://localhost/exchange_token"
        f"&approval_prompt=force"
        f"&scope=activity:read"
    )
    
    print(f"Opening browser to authorize application with the correct scope...")
    print(f"Please authorize the app and copy the 'code' parameter from the URL you are redirected to.")
    print(f"The URL will look like: http://localhost/exchange_token?state=&code=AUTHORIZATION_CODE&scope=...")
    
    # Open the browser to the authorization URL
    webbrowser.open(auth_url)
    
    # User will need to manually copy the code from the redirect URL
    auth_code = input("Enter the authorization code from the URL: ")
    return auth_code

def exchange_code_for_tokens(client_id: str, client_secret: str, auth_code: str) -> tuple[str, str]:
    """
    Exchange authorization code for tokens.
    
    Returns a tuple of access token and refresh token.
    """
    token_url = "https://www.strava.com/oauth/token"
    payload = {
        'client_id': client_id,
        'client_secret': client_secret,
        'code': auth_code,
        'grant_type': 'authorization_code'
    }
    
    response = requests.post(token_url, data=payload)
    tokens = response.json()
    
    if 'access_token' not in tokens:
        print("Error getting tokens:", tokens)
        return None, None
    
    print("Successfully obtained access token and refresh token!")
    return tokens['access_token'], tokens['refresh_token']

def get_access_token(client_id, client_secret, refresh_token=None):
    """Get a fresh access token, either using refresh token or new authorization."""
    if refresh_token:
        # Try using the refresh token if provided
        auth_url = "https://www.strava.com/oauth/token"
        payload = {
            'client_id': client_id,
            'client_secret': client_secret,
            'refresh_token': refresh_token,
            'grant_type': 'refresh_token'
        }
        
        response = requests.post(auth_url, data=payload)
        response_json = response.json()
        
        if 'access_token' in response_json:
            print("Successfully refreshed access token!")
            return response_json.get('access_token')
        else:
            print("Error refreshing token, need to re-authorize:", response_json)
    
    # If no refresh token or refresh failed, get a new authorization
    print("Initiating new authorization flow...")
    auth_code = get_authorization_code(client_id)
    access_token, new_refresh_token = exchange_code_for_tokens(client_id, client_secret, auth_code)
    
    # Save the refresh token for future use
    if new_refresh_token:
        print(f"New refresh token: {new_refresh_token}")
        print("Save this refresh token for future use!")
    
    return access_token

def get_activities(access_token, after_date, before_date):
    """Fetch activities from Strava within a date range."""
    activities_url = "https://www.strava.com/api/v3/athlete/activities"
    
    # Convert datetime objects to Unix timestamps (seconds since epoch)
    after_timestamp = int(after_date.timestamp())
    before_timestamp = int(before_date.timestamp())
    
    # Parameters for the request
    params = {
        'after': after_timestamp,
        'before': before_timestamp,
        'per_page': 200  # Maximum allowed by Strava API
    }
    
    headers = {'Authorization': f'Bearer {access_token}'}
    
    # Make the request
    response = requests.get(activities_url, headers=headers, params=params)
    activities = response.json()
    
    return activities

def filter_running_activities(activities: list[dict]) -> list[dict]:
    """Filter for running activities only."""
    running_activities = [activity for activity in activities if activity['type'] == 'Run']
    return running_activities

def process_activities(activities: list[dict], timezone: str = 'US/Eastern') -> dict:
    """
    Process activities to extract date and distance information.
    
    Returns a dictionary with daily miles run.
    """
    tz = pytz.timezone(timezone)
    
    # Initialize a dictionary to store daily miles
    daily_miles = {}
    
    for activity in activities:
        # Parse the start date
        start_date_str = activity['start_date']
        start_date_utc = parser.parse(start_date_str)
        
        # Convert to Eastern Time
        start_date_eastern = start_date_utc.astimezone(tz)
        
        # Get just the date part (year, month, day)
        activity_date = start_date_eastern.date()
        
        # Skip if not in March
        if activity_date.month != 3:
            continue
        
        # Convert distance from meters to miles
        distance_miles = activity['distance'] * 0.000621371
        
        # Add to daily totals
        if activity_date in daily_miles:
            daily_miles[activity_date] += distance_miles
        else:
            daily_miles[activity_date] = distance_miles
    
    return daily_miles

def create_march_table(daily_miles: dict) -> pd.DataFrame:
    """Create a table showing miles run each day in March."""
    # Get the current year
    current_year = datetime.datetime.now().year
    
    # Create a DataFrame for all days in March
    all_days = pd.date_range(start=f'{current_year}-03-01', end=f'{current_year}-03-31', freq='D')
    
    # Convert daily_miles dictionary to a DataFrame
    miles_df = pd.DataFrame(index=all_days)
    miles_df.index = miles_df.index.date  # Convert to date objects for matching
    
    # Create a column for miles run
    miles_df['Miles Run'] = 0.0
    
    # Fill in the miles for days with activities
    for date, miles in daily_miles.items():
        if date in miles_df.index:
            miles_df.loc[date, 'Miles Run'] = round(miles, 2)
    
    # Format the index for display
    miles_df.index = miles_df.index.map(lambda x: x.strftime('%B %d'))

    # Add name to the index
    miles_df.index.name = 'Date'
    
    return miles_df

# %%
# Get access token - we're passing None to force a new authentication with proper scopes
# After the first run, you can save the refresh token and use it in future runs
access_token = get_access_token(CLIENT_ID, CLIENT_SECRET, None)

if not access_token:
    print("Failed to get access token. Please try again.")

# Test the token to make sure we have the correct permissions
test_url = "https://www.strava.com/api/v3/athlete/activities?per_page=1"
headers = {'Authorization': f'Bearer {access_token}'}
test_response = requests.get(test_url, headers=headers)

if test_response.status_code != 200:
    print(f"Error testing token: {test_response.status_code}")
    print(test_response.json())

print("Successfully authenticated with proper permissions!")
    
# %%
# Set date range for March (current year)
current_year = datetime.datetime.now().year
after_date = datetime.datetime(current_year, 3, 1, 0, 0, 0, tzinfo=datetime.timezone.utc)
before_date = datetime.datetime(current_year, 4, 1, 0, 0, 0, tzinfo=datetime.timezone.utc)

print(f"Fetching running activities for March {current_year}...")

# Get activities
activities = get_activities(access_token, after_date, before_date)

if not activities:
    print("No activities found for the specified date range.")

print(f"Found {len(activities)} activities. Filtering for runs...")
    
# %%
# Filter running activities
running_activities = filter_running_activities(activities)

print(f"Found {len(running_activities)} running activities.")
# %%
    
# Process activities to get daily miles
daily_miles = process_activities(running_activities)
    
# %%
# Create table
march_df = create_march_table(daily_miles)
    
# %%
# Display the table
print("\nMarch Running Activities\n")
print(tabulate(march_df, headers=['Date', 'Miles Run'], tablefmt='grid'))

print(f"Saving march running stats to 'march_running_activities.csv'...")
march_df.to_csv('march_running_activities.csv')

# %%