#!/usr/bin/env python3
"""
Import Hevy CSV export into SQLite database.
Usage:
python import_csv.py <csv_file> [--db <database_path>]
"""
import argparse
import csv
import sqlite3
import sys
from datetime import datetime
from pathlib import Path
def create_schema(conn: sqlite3.Connection) -> None:
"""Create the workout_sets table with proper schema."""
conn.execute("""
CREATE TABLE IF NOT EXISTS workout_sets (
id INTEGER PRIMARY KEY AUTOINCREMENT,
title TEXT,
start_time TEXT,
end_time TEXT,
description TEXT,
exercise_title TEXT NOT NULL,
superset_id TEXT,
exercise_notes TEXT,
set_index INTEGER,
set_type TEXT,
weight_lbs REAL,
reps INTEGER,
distance_miles REAL,
duration_seconds INTEGER,
rpe REAL,
imported_at TEXT DEFAULT CURRENT_TIMESTAMP
)
""")
# Create indexes for common queries
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_exercise_title ON workout_sets(exercise_title)"
)
conn.execute("CREATE INDEX IF NOT EXISTS idx_start_time ON workout_sets(start_time)")
conn.execute(
"CREATE INDEX IF NOT EXISTS idx_exercise_start ON workout_sets(exercise_title, start_time)"
)
conn.commit()
def parse_value(value: str, target_type: type):
"""Parse CSV value to target type, handling empty strings."""
if not value or value.strip() == "":
return None
try:
if target_type == int:
return int(float(value)) # Handle "1.0" -> 1
elif target_type == float:
return float(value)
else:
return value.strip()
except (ValueError, TypeError):
return None
def import_csv(csv_path: Path, db_path: Path) -> dict:
"""Import CSV data into SQLite database."""
if not csv_path.exists():
raise FileNotFoundError(f"CSV file not found: {csv_path}")
conn = sqlite3.connect(db_path)
create_schema(conn)
# Clear existing data
conn.execute("DELETE FROM workout_sets")
stats = {"rows": 0, "errors": 0}
with open(csv_path, "r", encoding="utf-8") as f:
reader = csv.DictReader(f)
for row_num, row in enumerate(reader, start=2): # Start at 2 (header is row 1)
try:
conn.execute(
"""
INSERT INTO workout_sets (
title, start_time, end_time, description,
exercise_title, superset_id, exercise_notes,
set_index, set_type, weight_lbs, reps,
distance_miles, duration_seconds, rpe
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""",
(
parse_value(row.get("title", ""), str),
parse_value(row.get("start_time", ""), str),
parse_value(row.get("end_time", ""), str),
parse_value(row.get("description", ""), str),
parse_value(row.get("exercise_title", ""), str),
parse_value(row.get("superset_id", ""), str),
parse_value(row.get("exercise_notes", ""), str),
parse_value(row.get("set_index", ""), int),
parse_value(row.get("set_type", ""), str),
parse_value(row.get("weight_lbs", ""), float),
parse_value(row.get("reps", ""), int),
parse_value(row.get("distance_miles", ""), float),
parse_value(row.get("duration_seconds", ""), int),
parse_value(row.get("rpe", ""), float),
),
)
stats["rows"] += 1
except Exception as e:
print(f"Error on row {row_num}: {e}", file=sys.stderr)
print(f"Row data: {row}", file=sys.stderr)
stats["errors"] += 1
conn.commit()
conn.close()
return stats
def main():
parser = argparse.ArgumentParser(description="Import Hevy CSV export to SQLite")
parser.add_argument("csv_file", type=Path, help="Path to Hevy CSV export")
parser.add_argument(
"--db",
type=Path,
default=Path("hevy.db"),
help="Path to SQLite database (default: hevy.db)",
)
args = parser.parse_args()
print(f"Importing {args.csv_file} into {args.db}...")
try:
stats = import_csv(args.csv_file, args.db)
print(f"✓ Import complete!")
print(f" Rows imported: {stats['rows']}")
if stats["errors"] > 0:
print(f" Errors: {stats['errors']}")
print(f" Database: {args.db.absolute()}")
except Exception as e:
print(f"✗ Import failed: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
