#!/usr/bin/env python3 """ WHOOP Activity Sync - Fetches workouts from WHOOP API. Automatically syncs WHOOP workouts (surfing, mountain biking, training, etc.) to the activity tracking system. Run daily via cron or manually. Usage: python sync-whoop-activities.py # Sync last 7 days python sync-whoop-activities.py --days 14 # Sync last 14 days python sync-whoop-activities.py --test # Test without saving python sync-whoop-activities.py --debug # Show auth diagnostics Configuration: WHOOP mappings in .claude/context/activity-config.json: { "whoop_mappings": { "Surfing": "surfing", "Mountain Biking": "mtb", ... } } """ import json import sys from datetime import datetime from pathlib import Path # Add paths for imports sys.path.insert(0, str(Path(__file__).parent / '.claude' / 'skills' / 'supabase-connector')) sys.path.insert(0, str(Path(__file__).parent / 'mcp-servers' / 'whoop' / 'src')) CONFIG_PATH = Path(__file__).parent / '.claude' / 'context' / 'activity-config.json' def log(message): """Print timestamped log message.""" timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S") print(f"[{timestamp}] {message}") def show_auth_diagnostics(): """Show authentication storage diagnostics.""" log("Authentication Diagnostics") print("-" * 40) try: from whoop_mcp.auth import HybridTokenStorage status = HybridTokenStorage.get_storage_status() print("Keychain:") if status["keychain"]["available"]: expires = datetime.fromtimestamp(status["keychain"]["expires_at"]) print(f" Status: Available") print(f" Expires: {expires.strftime('%Y-%m-%d %H:%M:%S')}") else: print(f" Status: NOT AVAILABLE (expected in cron context)") print() print("File:") print(f" Path: {status['file']['path']}") if status["file"]["available"]: expires = datetime.fromtimestamp(status["file"]["expires_at"]) print(f" Status: Available") print(f" Expires: {expires.strftime('%Y-%m-%d %H:%M:%S')}") else: print(f" Status: NOT AVAILABLE") print() print("FIX: Run 'python mcp-servers/whoop/scripts/setup_auth.py --sync'") print(" to copy keychain tokens to file for cron access.") print("-" * 40) return status["keychain"]["available"] or status["file"]["available"] except Exception as e: print(f"Error checking auth status: {e}") return False def load_config(): """Load activity configuration""" try: with open(CONFIG_PATH) as f: return json.load(f) except Exception as e: log(f"Error loading config: {e}") return None def fetch_whoop_workouts(days=7): """Fetch workouts from WHOOP API. Returns list of workouts or None on error.""" try: from whoop_mcp.client import WhoopClient client = WhoopClient() workouts = client.get_workout_history(days=days) return [w.model_dump() for w in workouts] except Exception as e: log(f"ERROR fetching WHOOP workouts: {e}") # Show diagnostics on auth failure if "Not authorized" in str(e) or "401" in str(e): print() show_auth_diagnostics() return None def sync_workouts(workouts, whoop_mappings, test_mode=False): """Sync workouts to Supabase""" if not workouts: log("No workouts to sync") return 0 if test_mode: log(f"TEST MODE: Would sync {len(workouts)} workouts") for w in workouts[:5]: activity = whoop_mappings.get(w['sport_name'], 'other') print(f" {w['date']} {w['time']} - {w['sport_name']} -> {activity} ({w['duration_minutes']:.0f}min, strain: {w['strain']})") if len(workouts) > 5: print(f" ... and {len(workouts) - 5} more") return len(workouts) try: from client import get_supabase_client supabase = get_supabase_client() except Exception as e: log(f"Error connecting to Supabase: {e}") return 0 synced = 0 skipped = 0 for w in workouts: # Map WHOOP sport name to activity key activity = whoop_mappings.get(w['sport_name']) if not activity: log(f"Skipping unmapped activity: {w['sport_name']}") skipped += 1 continue # Use WHOOP workout ID for deduplication (first 8 chars) workout_id = w['id'][:8] if len(w['id']) >= 8 else w['id'] entry = { "id": workout_id, "date": w['date'], "time": w['time'] + ":00" if len(w['time']) == 5 else w['time'], "activity": activity, "source": "whoop", "duration_minutes": int(w['duration_minutes']) if w['duration_minutes'] else None, "metadata": { "sport_name": w['sport_name'], "sport_id": w['sport_id'], "strain": w['strain'], "avg_hr": w['avg_hr'], "max_hr": w['max_hr'], "calories_kj": w['calories'] } } try: supabase.table('chl_activity_logs').upsert(entry).execute() synced += 1 except Exception as e: log(f"Error syncing workout {workout_id}: {e}") log(f"Synced {synced} workouts, skipped {skipped} unmapped") return synced def main(): import argparse parser = argparse.ArgumentParser(description='Sync WHOOP workouts to activity tracker') parser.add_argument('--days', type=int, default=7, help='Number of days to sync (default: 7)') parser.add_argument('--test', action='store_true', help='Test mode - no database writes') parser.add_argument('--debug', action='store_true', help='Show auth diagnostics') args = parser.parse_args() if args.debug: show_auth_diagnostics() return 0 config = load_config() if not config: log("Failed to load configuration") return 1 # Check if WHOOP sync is enabled whoop_config = config.get('api_sources', {}).get('whoop', {}) if not whoop_config.get('enabled', True): log("WHOOP sync is disabled in config") return 0 whoop_mappings = config.get('whoop_mappings', {}) if not whoop_mappings: log("No WHOOP mappings configured in activity-config.json") return 1 log("WHOOP Activity Sync") log(f"Period: Last {args.days} days") if args.test: log("MODE: Test (no database writes)") log("Fetching WHOOP workouts...") workouts = fetch_whoop_workouts(args.days) if workouts is None: log("FAILED: Could not fetch workouts from WHOOP API") return 1 log(f"Found {len(workouts)} workouts") if workouts: log("Syncing to database...") synced = sync_workouts(workouts, whoop_mappings, args.test) log(f"Complete: {synced} workouts synced") else: log("No workouts found in the specified period") return 0 if __name__ == "__main__": sys.exit(main())