app/utils/parser.py

"""Generic gym-app workout parser.

Designed to ingest CSV exports from a variety of tracking apps (Hevy, Strong,
FitNotes, etc.), not just one. Instead of hard-coding column names, we map a set
of *aliases* onto a normalized schema, auto-detect weight units, and group rows
into sessions -> exercises -> sets.

Normalized structures:

    Set    -> one performed set (weight in kg + lbs, reps, rpe, distance, duration)
    Exercise -> named movement with its ordered sets
    Session  -> a single workout (title, start/end, ordered exercises)
"""

from __future__ import annotations

import io
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any, Iterable

import pandas as pd

LB_TO_KG = 0.45359237

# Column aliases: normalized_name -> possible source headers (lower-cased, stripped).
COLUMN_ALIASES: dict[str, tuple[str, ...]] = {
    "session_title": ("title", "workout_name", "workout", "routine", "name"),
    "start_time": ("start_time", "start", "date", "datetime", "workout_date", "time"),
    "end_time": ("end_time", "end", "finish_time"),
    "exercise": ("exercise_title", "exercise_name", "exercise", "movement"),
    "superset_id": ("superset_id", "superset"),
    "notes": ("exercise_notes", "notes", "note", "comment"),
    "set_index": ("set_index", "set_number", "set", "set_order"),
    "set_type": ("set_type", "type"),
    "weight_lbs": ("weight_lbs", "weight_lb", "weight_pounds", "lbs"),
    "weight_kg": ("weight_kg", "weight_kgs", "kg", "kilograms"),
    "weight": ("weight",),  # ambiguous unit; resolved via `weight_unit`
    "weight_unit": ("weight_unit", "unit", "units"),
    "reps": ("reps", "rep", "repetitions"),
    "distance_km": ("distance_km", "distance"),
    "duration_seconds": ("duration_seconds", "duration", "seconds", "time_seconds"),
    "rpe": ("rpe", "rir"),
}

DATE_FORMATS = (
    "%d %b %Y, %H:%M",        # "3 jun 2026, 12:45"
    "%Y-%m-%d %H:%M:%S",
    "%Y-%m-%d %H:%M",
    "%d/%m/%Y %H:%M",
    "%m/%d/%Y %H:%M",
    "%Y-%m-%dT%H:%M:%S",
    "%Y-%m-%d",
)

# Spanish month abbreviations -> English, so strptime can read either language.
_ES_MONTHS = {
    "ene": "Jan", "feb": "Feb", "mar": "Mar", "abr": "Apr", "may": "May",
    "jun": "Jun", "jul": "Jul", "ago": "Aug", "sep": "Sep", "set": "Sep",
    "oct": "Oct", "nov": "Nov", "dic": "Dec",
}


@dataclass
class WorkoutSet:
    set_index: int
    set_type: str = "normal"
    weight_kg: float | None = None
    reps: int | None = None
    rpe: float | None = None
    distance_km: float | None = None
    duration_seconds: float | None = None

    @property
    def weight_lbs(self) -> float | None:
        return None if self.weight_kg is None else round(self.weight_kg / LB_TO_KG, 2)

    @property
    def volume_kg(self) -> float:
        if self.weight_kg is None or self.reps is None:
            return 0.0
        return self.weight_kg * self.reps

    def to_dict(self) -> dict[str, Any]:
        return {
            "set_index": self.set_index,
            "set_type": self.set_type,
            "weight_kg": self.weight_kg,
            "weight_lbs": self.weight_lbs,
            "reps": self.reps,
            "rpe": self.rpe,
            "distance_km": self.distance_km,
            "duration_seconds": self.duration_seconds,
            "volume_kg": round(self.volume_kg, 2),
        }


@dataclass
class Exercise:
    name: str
    muscle_group: str = "other"
    notes: str = ""
    sets: list[WorkoutSet] = field(default_factory=list)

    @property
    def volume_kg(self) -> float:
        return sum(s.volume_kg for s in self.sets)

    def to_dict(self) -> dict[str, Any]:
        return {
            "name": self.name,
            "muscle_group": self.muscle_group,
            "notes": self.notes,
            "sets": [s.to_dict() for s in self.sets],
            "volume_kg": round(self.volume_kg, 2),
        }


@dataclass
class Session:
    title: str
    start_time: datetime | None
    end_time: datetime | None
    exercises: list[Exercise] = field(default_factory=list)

    @property
    def volume_kg(self) -> float:
        return sum(e.volume_kg for e in self.exercises)

    @property
    def duration_minutes(self) -> float | None:
        if self.start_time and self.end_time:
            return round((self.end_time - self.start_time).total_seconds() / 60.0, 1)
        return None

    @property
    def total_sets(self) -> int:
        return sum(len(e.sets) for e in self.exercises)

    def to_dict(self) -> dict[str, Any]:
        return {
            "title": self.title,
            "start_time": self.start_time.isoformat() if self.start_time else None,
            "end_time": self.end_time.isoformat() if self.end_time else None,
            "duration_minutes": self.duration_minutes,
            "volume_kg": round(self.volume_kg, 2),
            "total_sets": self.total_sets,
            "exercises": [e.to_dict() for e in self.exercises],
        }


# --------------------------------------------------------------------------------------
# Muscle-group inference (bilingual keyword match: English + Spanish)
# --------------------------------------------------------------------------------------
_MUSCLE_KEYWORDS: dict[str, tuple[str, ...]] = {
    "chest": ("bench", "chest", "press de banca", "pecho", "aperturas", "fly", "vuelos",
              "pec", "dips", "fondos"),
    "back": ("row", "remo", "pull", "jalon", "jalón", "dominada", "pulldown", "lat",
             "espalda", "deadlift", "peso muerto", "pullover"),
    "shoulders": ("shoulder", "hombro", "press de hombros", "overhead", "lateral",
                  "elevacion", "elevación", "raise", "delt", "arnold", "face pull",
                  "posteriores"),
    "biceps": ("curl", "biceps", "bíceps", "predicador", "preacher"),
    "triceps": ("triceps", "tríceps", "extension de codo", "pushdown", "skull",
                "frances", "francés", "press cerrado"),
    "legs": ("squat", "sentadilla", "leg", "pierna", "lunge", "zancada", "press a una",
             "extension de pierna", "extensión de pierna", "leg press", "prensa",
             "curl femoral", "hamstring", "femoral", "hack", "goblet"),
    "glutes": ("glute", "gluteo", "glúteo", "hip thrust", "empuje de cadera", "puente"),
    "calves": ("calf", "calves", "gemelo", "gastrocnemio", "soleo", "pantorrilla"),
    "core": ("ab", "core", "plank", "plancha", "crunch", "abdominal", "oblicuo",
             "russian twist"),
    "cardio": ("run", "carrera", "treadmill", "cinta", "bike", "bicicleta", "row erg",
               "remo ergometro", "elliptical", "eliptica", "elíptica", "cardio"),
    "forearms": ("forearm", "antebrazo", "wrist", "muñeca", "invertido", "reverse curl",
                 "grip", "agarre"),
}


def infer_muscle_group(exercise_name: str) -> str:
    name = exercise_name.lower()
    for group, keywords in _MUSCLE_KEYWORDS.items():
        if any(kw in name for kw in keywords):
            return group
    return "other"


# --------------------------------------------------------------------------------------
# Parsing helpers
# --------------------------------------------------------------------------------------
def _normalize_headers(df: pd.DataFrame) -> dict[str, str]:
    """Map normalized field name -> actual column present in the dataframe."""
    lower_to_actual = {str(c).strip().lower(): c for c in df.columns}
    resolved: dict[str, str] = {}
    for field_name, aliases in COLUMN_ALIASES.items():
        for alias in aliases:
            if alias in lower_to_actual:
                resolved[field_name] = lower_to_actual[alias]
                break
    return resolved


def _to_float(value: Any) -> float | None:
    if value is None:
        return None
    if isinstance(value, float) and pd.isna(value):
        return None
    s = str(value).strip().replace(",", ".")
    if s == "" or s.lower() in {"nan", "none", "null"}:
        return None
    try:
        return float(s)
    except ValueError:
        return None


def _to_int(value: Any) -> int | None:
    f = _to_float(value)
    return None if f is None else int(round(f))


def parse_datetime(value: Any) -> datetime | None:
    if value is None or (isinstance(value, float) and pd.isna(value)):
        return None
    raw = str(value).strip()
    if not raw:
        return None
    lowered = raw.lower()
    for es, en in _ES_MONTHS.items():
        lowered = lowered.replace(f" {es} ", f" {en} ")
    candidate = lowered.title().replace("Am", "AM").replace("Pm", "PM")
    for fmt in DATE_FORMATS:
        try:
            return datetime.strptime(candidate, fmt)
        except ValueError:
            continue
    parsed = pd.to_datetime(raw, errors="coerce", dayfirst=True)
    return None if pd.isna(parsed) else parsed.to_pydatetime()


def _resolve_weight_kg(row: pd.Series, cols: dict[str, str]) -> float | None:
    """Return weight in kilograms regardless of the source unit."""
    if "weight_kg" in cols:
        kg = _to_float(row.get(cols["weight_kg"]))
        if kg is not None:
            return round(kg, 4)
    if "weight_lbs" in cols:
        lbs = _to_float(row.get(cols["weight_lbs"]))
        if lbs is not None:
            return round(lbs * LB_TO_KG, 4)
    if "weight" in cols:
        w = _to_float(row.get(cols["weight"]))
        if w is None:
            return None
        unit = ""
        if "weight_unit" in cols:
            unit = str(row.get(cols["weight_unit"], "")).strip().lower()
        if unit in {"lb", "lbs", "pound", "pounds"}:
            return round(w * LB_TO_KG, 4)
        return round(w, 4)  # assume kg by default
    return None


# --------------------------------------------------------------------------------------
# Public API
# --------------------------------------------------------------------------------------
def parse_workouts(source: str | bytes | io.IOBase | pd.DataFrame) -> list[Session]:
    """Parse a workout export into a chronologically sorted list of Sessions.

    `source` may be a file path, raw CSV bytes/str, a file-like object, or a DataFrame.
    """
    if isinstance(source, pd.DataFrame):
        df = source.copy()
    elif isinstance(source, (bytes, bytearray)):
        df = pd.read_csv(io.BytesIO(source))
    elif isinstance(source, str) and ("\n" in source or "," in source) and not source.endswith(".csv"):
        df = pd.read_csv(io.StringIO(source))
    else:
        df = pd.read_csv(source)

    if df.empty:
        return []

    cols = _normalize_headers(df)
    if "exercise" not in cols:
        raise ValueError(
            "Could not find an exercise column. Expected one of: "
            + ", ".join(COLUMN_ALIASES["exercise"])
        )

    sessions: dict[tuple[str, str], Session] = {}
    exercises: dict[tuple[str, str, str], Exercise] = {}

    for _, row in df.iterrows():
        title = str(row.get(cols.get("session_title", ""), "") or "Workout").strip() or "Workout"
        start_raw = str(row.get(cols.get("start_time", ""), "") or "")
        start_dt = parse_datetime(start_raw) if "start_time" in cols else None
        end_dt = parse_datetime(row.get(cols["end_time"])) if "end_time" in cols else None

        session_key = (title, start_raw)
        if session_key not in sessions:
            sessions[session_key] = Session(title=title, start_time=start_dt, end_time=end_dt)
        session = sessions[session_key]

        ex_name = str(row.get(cols["exercise"], "") or "").strip()
        if not ex_name:
            continue
        ex_key = (*session_key, ex_name)
        if ex_key not in exercises:
            notes = ""
            if "notes" in cols:
                notes = str(row.get(cols["notes"], "") or "").strip()
            exercise = Exercise(
                name=ex_name,
                muscle_group=infer_muscle_group(ex_name),
                notes=notes,
            )
            exercises[ex_key] = exercise
            session.exercises.append(exercise)
        exercise = exercises[ex_key]

        workout_set = WorkoutSet(
            set_index=_to_int(row.get(cols.get("set_index", ""))) or len(exercise.sets),
            set_type=str(row.get(cols.get("set_type", ""), "normal") or "normal").strip() or "normal",
            weight_kg=_resolve_weight_kg(row, cols),
            reps=_to_int(row.get(cols.get("reps", ""))),
            rpe=_to_float(row.get(cols.get("rpe", ""))),
            distance_km=_to_float(row.get(cols.get("distance_km", ""))),
            duration_seconds=_to_float(row.get(cols.get("duration_seconds", ""))),
        )
        exercise.sets.append(workout_set)

    ordered = list(sessions.values())
    ordered.sort(key=lambda s: s.start_time or datetime.min)
    return ordered


def sessions_to_dicts(sessions: Iterable[Session]) -> list[dict[str, Any]]:
    return [s.to_dict() for s in sessions]