diff --git a/PROJECT.org b/PROJECT.org index 42c90b4..8e80ed7 100644 --- a/PROJECT.org +++ b/PROJECT.org @@ -722,6 +722,30 @@ Added a `workouts` Django app so we can scrobble gym sessions. - Templates, admin, DRF viewsets (`exercises`, `workout-routines`), MCP tools, and `tests/workouts_tests/` (unit conversion, form round-trips, importer, imperial POST) included. +** TODO [#B] Add Flexify workout importer :workouts:importer: +:PROPERTIES: +:ID: 96c9fbf5-a0d5-4f09-a72c-bb491b7ece18 +:END: + +*** Description + +Import gym sessions and routines from a Flexify backup. + +- `import_flexify()` in `workouts/importer.py` reads a Flexify sqlite backup + (`data/flexify-example.sqlite`) plus an optional plans CSV + (`data/flexify-plans-example.csv`). Plans become `WorkoutRoutine`s with + `WorkoutRoutineExercise` planned-exercise rows; logged `gym_sets` are + clustered into sessions (gap > 60 min or plan change) and converted into + workout scrobbles. +- A curated `FLEXIFY_EXERCISE_ALIASES` map resolves Flexify exercise names to + the wrkout catalog, falling back to `find_or_create` for unmatched names. +- `python manage.py import_flexify --sqlite ... --csv ... --user ...` command. +- `scan_webdav_for_flexify()` watches `var/workouts/` on WebDAV (via the + existing import_from_webdav sweep), pairing a plans CSV with the sqlite + backup, and queues `process_workout_import` celery tasks. +- `WorkoutImport` model (`BaseFileImportMixin`, new `file_hash` field for + change detection), admin, DRF `workout-imports` viewset, planned-exercises + block on the routine detail template, and MCP routine dict update. * Version 64.5 [1/1] ** DONE [#B] Clean up issues with org-mode notes :orgmode:notes:scrobbles: :PROPERTIES: diff --git a/data/flexify-example.sqlite b/data/flexify-example.sqlite new file mode 100644 index 0000000..dc6f927 Binary files /dev/null and b/data/flexify-example.sqlite differ diff --git a/data/flexify-plans-example.csv b/data/flexify-plans-example.csv new file mode 100644 index 0000000..a7c4bd6 --- /dev/null +++ b/data/flexify-plans-example.csv @@ -0,0 +1,4 @@ +id,days,title,sequence,exercises +1,Monday,Middle Distance - Push Day,,Barbell bench press;Squat;Lat pull-down;Leg press +2,Wednesday,Middle Distance - Pull Day,,Deadlift;Overhead triceps extension;Dumbbell biceps curl;Barbell bent-over row +3,Friday,Middle Distance - Arm Day,,Leg press;Pull-up;Push-up;Crunch diff --git a/tests/workouts_tests/test_flexify_importer.py b/tests/workouts_tests/test_flexify_importer.py new file mode 100644 index 0000000..5cfc0d3 --- /dev/null +++ b/tests/workouts_tests/test_flexify_importer.py @@ -0,0 +1,307 @@ +import os + +import pytest +from django.contrib.auth import get_user_model +from scrobbles.importers.webdav import scan_webdav_for_flexify +from scrobbles.models import Scrobble +from workouts.importer import ( + FLEXIFY_EXERCISE_ALIASES, + aggregate_bodyweight, + aggregate_workouts, + cluster_sessions, + import_flexify, + read_flexify_sqlite, + read_plans_csv, + resolve_exercise, +) +from workouts.models import Exercise, WorkoutRoutine, WorkoutRoutineExercise + +User = get_user_model() + +SAMPLE_SQLITE = os.path.join( + os.path.dirname(__file__), "..", "..", "data", "flexify-example.sqlite" +) +SAMPLE_CSV = os.path.join( + os.path.dirname(__file__), "..", "..", "data", "flexify-plans-example.csv" +) + + +@pytest.fixture +def user(db): + return User.objects.create(email="lifter@example.com") + + +def test_read_plans_csv(): + plans = read_plans_csv(SAMPLE_CSV) + assert len(plans) == 3 + assert plans[0]["days"] == "Monday" + assert plans[0]["title"] == "Middle Distance - Push Day" + assert plans[0]["exercises"] == [ + "Barbell bench press", + "Squat", + "Lat pull-down", + "Leg press", + ] + + +def test_read_flexify_sqlite(): + data = read_flexify_sqlite(SAMPLE_SQLITE) + assert len(data["plans"]) == 3 + assert len(data["plan_exercises"]) == 12 + assert len(data["catalog"]) == 58 + assert data["strength_unit"] == "lb" + assert len(data["sets"]) == 12 + first = data["sets"][0] + assert first["name"] == "Leg press" + assert first["unit"] == "lb" + assert first["plan_id"] == 3 + + +def test_resolve_exercise_alias(db): + exercise, created = resolve_exercise("Barbell bench press") + assert exercise.name == "Barbell Bench Press - Medium Grip" + assert created is True + + +def test_resolve_exercise_exact_name(db): + exercise, created = resolve_exercise("Leg press") + assert exercise.name == "Leg Press" + assert created is True + + +def test_resolve_exercise_dry_run_does_not_create(db): + before = Exercise.objects.count() + exercise, created = resolve_exercise("Some Unknown Exercise", dry_run=True) + assert exercise is None + assert created is True + assert Exercise.objects.count() == before + + +def test_resolve_exercise_reuses_existing(db): + first, _ = resolve_exercise("Push-up") + second, created = resolve_exercise("Push-up") + assert second.id == first.id + assert created is False + + +def test_cluster_sessions_splits_on_time_gap(): + sets = [ + {"name": "A", "created": 1000, "id": 1, "plan_id": 1}, + {"name": "B", "created": 1010, "id": 2, "plan_id": 1}, + {"name": "C", "created": 1000 + 61 * 60, "id": 3, "plan_id": 1}, + {"name": "D", "created": 1001 + 61 * 60, "id": 4, "plan_id": 1}, + ] + sessions = cluster_sessions(sets, gap_minutes=60) + assert len(sessions) == 2 + assert [s["name"] for s in sessions[0]] == ["A", "B"] + assert [s["name"] for s in sessions[1]] == ["C", "D"] + + +def test_cluster_sessions_splits_on_plan_change(): + sets = [ + {"name": "A", "created": 1000, "id": 1, "plan_id": 1}, + {"name": "B", "created": 1005, "id": 2, "plan_id": 2}, + ] + sessions = cluster_sessions(sets, gap_minutes=60) + assert len(sessions) == 2 + assert [s["name"] for s in sessions[0]] == ["A"] + assert [s["name"] for s in sessions[1]] == ["B"] + + +def test_aggregate_workouts_weights(db): + session = read_flexify_sqlite(SAMPLE_SQLITE)["sets"][:3] + workouts = aggregate_workouts(session) + assert len(workouts) == 1 + entry = workouts[0] + assert entry["sets"] == 3 + assert entry["reps"] == 10 + assert entry["weight_kg"] == pytest.approx(54.43, abs=0.01) + + +def test_aggregate_bodyweight(): + session = read_flexify_sqlite(SAMPLE_SQLITE)["sets"] + assert aggregate_bodyweight(session) == pytest.approx(83.91, abs=0.01) + + +def test_import_flexify_dry_run(user): + result = import_flexify( + sqlite_path=SAMPLE_SQLITE, + csv_path=SAMPLE_CSV, + user_id=user.id, + dry_run=True, + ) + assert result["routines_created"] == 3 + assert result["routines_updated"] == 0 + assert result["scrobbles_created"] == 1 + assert Scrobble.objects.count() == 0 + assert WorkoutRoutine.objects.count() == 0 + + +def test_import_flexify_full(user): + result = import_flexify( + sqlite_path=SAMPLE_SQLITE, + csv_path=SAMPLE_CSV, + user_id=user.id, + ) + assert result["routines_created"] == 3 + assert result["scrobbles_created"] == 1 + assert result["scrobbles_skipped"] == 0 + + scrobble = Scrobble.objects.get(user=user) + assert scrobble.source == "Flexify Import" + assert scrobble.played_to_completion is True + assert scrobble.workout_routine.title == "Middle Distance - Arm Day" + + entry = scrobble.logdata.workouts[0] + assert entry["exercise_id"] is not None + assert entry["sets"] == 3 + assert entry["reps"] == 10 + assert entry["weight_kg"] == pytest.approx(54.43, abs=0.01) + assert scrobble.logdata.bodyweight_kg == pytest.approx(83.91, abs=0.01) + assert scrobble.logdata.duration_minutes == 2 + + planned = WorkoutRoutineExercise.objects.filter( + routine=scrobble.workout_routine + ).order_by("sequence") + assert [pe.exercise.name for pe in planned] == [ + "Leg Press", + "Pullups", + "Pushups", + "Crunches", + ] + assert result["exercises_created"] == 11 + + +def test_import_flexify_idempotent(user): + import_flexify(sqlite_path=SAMPLE_SQLITE, csv_path=SAMPLE_CSV, user_id=user.id) + result = import_flexify( + sqlite_path=SAMPLE_SQLITE, + csv_path=SAMPLE_CSV, + user_id=user.id, + ) + assert result["routines_created"] == 0 + assert result["routines_updated"] == 3 + assert result["scrobbles_created"] == 0 + assert result["scrobbles_existing"] == 1 + assert Scrobble.objects.filter(user=user).count() == 1 + + +def test_import_flexify_csv_only(user): + result = import_flexify( + csv_path=SAMPLE_CSV, + user_id=user.id, + ) + assert result["routines_created"] == 3 + assert result["scrobbles_created"] == 0 + + +def test_import_flexify_sqlite_only(user): + result = import_flexify( + sqlite_path=SAMPLE_SQLITE, + user_id=user.id, + ) + assert result["scrobbles_created"] == 1 + routine_titles = set(WorkoutRoutine.objects.values_list("title", flat=True)) + assert "Friday" in routine_titles + + +def test_alias_targets_exist_in_wrkout_catalog(): + assert FLEXIFY_EXERCISE_ALIASES["Pull-up"] == "Pullups" + assert FLEXIFY_EXERCISE_ALIASES["Push-up"] == "Pushups" + assert FLEXIFY_EXERCISE_ALIASES["Crunch"] == "Crunches" + assert FLEXIFY_EXERCISE_ALIASES["Deadlift"] == "Barbell Deadlift" + assert FLEXIFY_EXERCISE_ALIASES["Squat"] == "Barbell Squat" + + +class FakeWebDAVClient: + def __init__(self, files): + self.files = { + "var/workouts/": None, + **{f"var/workouts/{name}": data for name, data in files.items()}, + } + + def info(self, path): + if not any(k.startswith(path) or k == path for k in self.files): + raise FileNotFoundError(path) + + def list(self, path, get_info=False): + names = [k[len(path) :] for k in self.files if k.startswith(path) and k != path] + if get_info: + return [{"path": f"{path}{name}"} for name in names] + return names + + def mkdir(self, path, recursive=False): + self.files.setdefault(path, None) + + def download_sync(self, remote_path, local_path): + with open(local_path, "wb") as f: + f.write(self.files[remote_path]) + + def move(self, src, dst): + self.files[dst] = self.files.pop(src) + + +def test_scan_webdav_for_flexify(user, monkeypatch, tmp_path): + with open(SAMPLE_SQLITE, "rb") as f: + sqlite_bytes = f.read() + with open(SAMPLE_CSV, "rb") as f: + csv_bytes = f.read() + + client = FakeWebDAVClient( + { + "flexify-backup.sqlite": sqlite_bytes, + "flexify-plans.csv": csv_bytes, + } + ) + + import scrobbles.tasks as scrobbles_tasks + from workouts.models import WorkoutImport + + queued = [] + monkeypatch.setattr( + scrobbles_tasks, + "process_workout_import", + type("Task", (), {"delay": lambda self, pk: queued.append(pk)})(), + ) + + count = scan_webdav_for_flexify(client, user.id) + assert count == 1 + assert len(queued) == 1 + assert "var/workouts/flexify-backup.sqlite" not in client.files + assert "var/workouts/flexify-plans.csv" not in client.files + assert any( + k.startswith("var/workouts/processed/") and k.endswith(".sqlite") + for k in client.files + ) + + imp = WorkoutImport.objects.get(pk=queued[0]) + assert imp.user_id == user.id + assert imp.original_filename == "flexify-backup.sqlite" + assert imp.file_hash + assert imp.csv_file + + +def test_scan_webdav_for_flexify_skips_unchanged(user, monkeypatch, tmp_path): + with open(SAMPLE_SQLITE, "rb") as f: + sqlite_bytes = f.read() + + client = FakeWebDAVClient({"flexify-backup.sqlite": sqlite_bytes}) + + import scrobbles.tasks as scrobbles_tasks + from workouts.models import WorkoutImport + + queued = [] + monkeypatch.setattr( + scrobbles_tasks, + "process_workout_import", + type("Task", (), {"delay": lambda self, pk: queued.append(pk)})(), + ) + + assert scan_webdav_for_flexify(client, user.id) == 1 + assert len(queued) == 1 + + client.files["var/workouts/flexify-backup.sqlite"] = sqlite_bytes + + count = scan_webdav_for_flexify(client, user.id) + assert count == 0 + assert len(queued) == 1 diff --git a/vrobbler/apps/scrobbles/importers/webdav.py b/vrobbler/apps/scrobbles/importers/webdav.py index e96c098..f2033da 100644 --- a/vrobbler/apps/scrobbles/importers/webdav.py +++ b/vrobbler/apps/scrobbles/importers/webdav.py @@ -20,6 +20,7 @@ DEFAULT_BGSTATS_PATH = "var/bgstats/" DEFAULT_EBIRD_PATH = "var/ebird/" DEFAULT_SCALE_PATH = "var/scale/" DEFAULT_UDISC_PATH = "var/udisc/" +DEFAULT_WORKOUTS_PATH = "var/workouts/" def import_from_webdav_for_all_users( @@ -50,6 +51,7 @@ def import_from_webdav_for_all_users( ebird_count = 0 scale_count = 0 udisc_count = 0 + workout_count = 0 for user_id in webdav_enabled_user_ids: client = get_webdav_client(user_id) @@ -84,9 +86,13 @@ def import_from_webdav_for_all_users( udisc_count += scan_webdav_for_udisc( client, user_id, include_processed=include_processed ) + logger.info("Scanning WebDAV workouts for user %s", user_id) + workout_count += scan_webdav_for_flexify( + client, user_id, include_processed=include_processed + ) logger.info( - "Started %d KOReader, %d Trail GPX, %d Retroarch, %d BGStats, %d eBird, %d Scale, %d uDisc WebDAV imports", + "Started %d KOReader, %d Trail GPX, %d Retroarch, %d BGStats, %d eBird, %d Scale, %d uDisc, %d Workout WebDAV imports", ko_count, gpx_count, retro_count, @@ -94,6 +100,7 @@ def import_from_webdav_for_all_users( ebird_count, scale_count, udisc_count, + workout_count, extra={ "koreader": ko_count, "trail_gpx": gpx_count, @@ -102,9 +109,19 @@ def import_from_webdav_for_all_users( "ebird": ebird_count, "scale": scale_count, "udisc": udisc_count, + "workouts": workout_count, }, ) - return ko_count, gpx_count, retro_count, bgstats_count, ebird_count, scale_count, udisc_count + return ( + ko_count, + gpx_count, + retro_count, + bgstats_count, + ebird_count, + scale_count, + udisc_count, + workout_count, + ) def scan_webdav_for_koreader( @@ -342,8 +359,8 @@ def scan_webdav_for_retroarch( retroarch overwrites the same .lrtl filename. """ import hashlib - import shutil import re + import shutil import zipfile from datetime import datetime @@ -379,8 +396,7 @@ def scan_webdav_for_retroarch( f["name"] = os.path.basename(f["path"]) lrtl_files = [ - f for f in root_files - if (f.get("name") or "").lower().endswith(".lrtl") + f for f in root_files if (f.get("name") or "").lower().endswith(".lrtl") ] # Optionally include historical files from processed/ @@ -396,10 +412,13 @@ def scan_webdav_for_retroarch( for f in processed_files: if f.get("path") and not f.get("name"): f["name"] = os.path.basename(f["path"]) - lrtl_files.extend([ - f for f in processed_files - if (f.get("name") or "").lower().endswith(".lrtl") - ]) + lrtl_files.extend( + [ + f + for f in processed_files + if (f.get("name") or "").lower().endswith(".lrtl") + ] + ) if not lrtl_files: logger.info("No .lrtl files found on webdav", extra={"user_id": user_id}) @@ -408,6 +427,7 @@ def scan_webdav_for_retroarch( # Sort chronologically: processed files (timestamp prefixed) first, # current root files last. ts_pattern = re.compile(r"^(\d{12})-") + def sort_key(f): name = f.get("name") or "" m = ts_pattern.match(name) @@ -924,9 +944,183 @@ def scan_webdav_for_udisc(webdav_client, user_id, include_processed=False): process_udisc_csv_import.delay(imp.id) new_imports += 1 except Exception as e: - logger.error( - f"Failed to import processed uDisc CSV file {fname}: {e}" - ) + logger.error(f"Failed to import processed uDisc CSV file {fname}: {e}") + finally: + os.unlink(tmp.name) + + return new_imports + + +def scan_webdav_for_flexify(webdav_client, user_id, include_processed=False): + """Download Flexify sqlite backups from WebDAV var/workouts/ and queue imports. + + A plans CSV next to the sqlite file (matching stem, else the first CSV in + the directory) is attached to the import so plans get their Flexify titles. + Files are moved to var/workouts/processed/ after downloading so they are + not re-imported on subsequent scans unless *include_processed* is True. + """ + from scrobbles.tasks import process_workout_import + from workouts.models import WorkoutImport + + workouts_path = DEFAULT_WORKOUTS_PATH + try: + webdav_client.info(workouts_path) + except: + logger.info("No var/workouts/ directory on webdav", extra={"user_id": user_id}) + return 0 + + try: + files = webdav_client.list(workouts_path, get_info=True) + except Exception as e: + logger.warning( + "Could not list var/workouts/", + extra={"user_id": user_id, "error": str(e)}, + ) + return 0 + + processed_dir = f"{workouts_path}processed/" + try: + webdav_client.mkdir(processed_dir, recursive=True) + except Exception: + pass + + def basename(f): + return os.path.basename(f.get("path") or f.get("name") or "") + + sqlite_files = [ + f for f in files if basename(f).lower().endswith((".sqlite", ".sqlite3", ".db")) + ] + csv_files = [basename(f) for f in files if basename(f).lower().endswith(".csv")] + if not sqlite_files: + logger.info( + "No Flexify sqlite files found on webdav", extra={"user_id": user_id} + ) + return 0 + + if WorkoutImport.objects.filter( + user_id=user_id, processed_finished__isnull=True + ).exists(): + logger.info( + "Workout import already pending for user", extra={"user_id": user_id} + ) + return 0 + + new_imports = 0 + + for f in sqlite_files: + fname = basename(f) + tmp = tempfile.NamedTemporaryFile(delete=False, suffix=fname) + try: + webdav_client.download_sync( + remote_path=f"{workouts_path}{fname}", local_path=tmp.name + ) + new_hash = get_file_md5_hash(tmp.name) + + last_import = ( + WorkoutImport.objects.filter( + user_id=user_id, + original_filename=fname, + processed_finished__isnull=False, + ) + .order_by("-processed_finished") + .first() + ) + if last_import and last_import.file_hash == new_hash: + logger.debug( + "Workout sqlite %s unchanged (hash match), skipping", + fname, + extra={"user_id": user_id, "hash": new_hash}, + ) + continue + + imp = WorkoutImport.objects.create( + user_id=user_id, + original_filename=fname, + file_hash=new_hash, + ) + with open(tmp.name, "rb") as f: + imp.sqlite_file.save(fname, f, save=True) + + stem = os.path.splitext(fname)[0] + csv_fname = next((c for c in csv_files if c.startswith(stem)), None) + if not csv_fname and csv_files: + csv_fname = csv_files[0] + if csv_fname: + csv_tmp = tempfile.NamedTemporaryFile(delete=False, suffix=csv_fname) + try: + webdav_client.download_sync( + remote_path=f"{workouts_path}{csv_fname}", + local_path=csv_tmp.name, + ) + with open(csv_tmp.name, "rb") as f: + imp.csv_file.save(csv_fname, f, save=True) + except Exception as e: + logger.error(f"Failed to fetch Flexify plans CSV {csv_fname}: {e}") + finally: + os.unlink(csv_tmp.name) + + stem_ext, ext = os.path.splitext(fname) + ts = datetime.now().strftime("%Y%m%dT%H%M%S") + webdav_client.move( + f"{workouts_path}{fname}", + f"{processed_dir}{stem_ext}_{ts}{ext}", + ) + if csv_fname: + try: + webdav_client.move( + f"{workouts_path}{csv_fname}", + f"{processed_dir}{csv_fname}", + ) + except Exception: + pass + + process_workout_import.delay(imp.id) + new_imports += 1 + except Exception as e: + logger.error(f"Failed to import Flexify file {fname}: {e}") + finally: + os.unlink(tmp.name) + + if include_processed: + try: + processed_files = webdav_client.list(processed_dir, get_info=True) + except Exception as e: + logger.warning( + "Could not list var/workouts/processed/", + extra={"user_id": user_id, "error": str(e)}, + ) + return new_imports + + for f in processed_files: + fname = basename(f) + if not fname.lower().endswith((".sqlite", ".sqlite3", ".db")): + continue + + tmp = tempfile.NamedTemporaryFile(delete=False, suffix=fname) + try: + webdav_client.download_sync( + remote_path=f"{processed_dir}{fname}", local_path=tmp.name + ) + new_hash = get_file_md5_hash(tmp.name) + if WorkoutImport.objects.filter( + user_id=user_id, + original_filename=fname, + file_hash=new_hash, + processed_finished__isnull=False, + ).exists(): + continue + + imp = WorkoutImport.objects.create( + user_id=user_id, + original_filename=fname, + file_hash=new_hash, + ) + with open(tmp.name, "rb") as f: + imp.sqlite_file.save(fname, f, save=True) + process_workout_import.delay(imp.id) + new_imports += 1 + except Exception as e: + logger.error(f"Failed to import processed Flexify file {fname}: {e}") finally: os.unlink(tmp.name) diff --git a/vrobbler/apps/scrobbles/tasks.py b/vrobbler/apps/scrobbles/tasks.py index be3e7f7..1e6d12a 100644 --- a/vrobbler/apps/scrobbles/tasks.py +++ b/vrobbler/apps/scrobbles/tasks.py @@ -312,6 +312,16 @@ def process_trail_gpx_import(import_id): trail_gpx_import.process() +@shared_task +def process_workout_import(import_id): + WorkoutImport = apps.get_model("workouts", "WorkoutImport") + workout_import = WorkoutImport.objects.filter(id=import_id).first() + if not workout_import: + logger.warn(f"WorkoutImport not found with id {import_id}") + return + workout_import.process() + + @shared_task def process_ebird_csv_import(import_id): EBirdCSVImport = apps.get_model("scrobbles", "EBirdCSVImport") diff --git a/vrobbler/apps/workouts/admin.py b/vrobbler/apps/workouts/admin.py index 65f4c25..3e46dec 100644 --- a/vrobbler/apps/workouts/admin.py +++ b/vrobbler/apps/workouts/admin.py @@ -1,6 +1,11 @@ from django.contrib import admin from scrobbles.admin import ScrobbleInline -from workouts.models import Exercise, WorkoutRoutine +from workouts.models import ( + Exercise, + WorkoutImport, + WorkoutRoutine, + WorkoutRoutineExercise, +) @admin.register(Exercise) @@ -11,6 +16,11 @@ class ExerciseAdmin(admin.ModelAdmin): search_fields = ("name", "primary_muscles", "secondary_muscles") +class WorkoutRoutineExerciseInline(admin.TabularInline): + model = WorkoutRoutineExercise + extra = 0 + + @admin.register(WorkoutRoutine) class WorkoutRoutineAdmin(admin.ModelAdmin): date_hierarchy = "created" @@ -19,4 +29,29 @@ class WorkoutRoutineAdmin(admin.ModelAdmin): search_fields = ("title", "description") inlines = [ ScrobbleInline, + WorkoutRoutineExerciseInline, ] + + +@admin.register(WorkoutImport) +class WorkoutImportAdmin(admin.ModelAdmin): + list_display = ( + "uuid", + "user", + "original_filename", + "processing_started", + "processed_finished", + "process_count", + ) + ordering = ("-created",) + readonly_fields = ( + "uuid", + "created", + "modified", + "processing_started", + "processed_finished", + "process_log", + "process_count", + "error_log", + "file_hash", + ) diff --git a/vrobbler/apps/workouts/api/serializers.py b/vrobbler/apps/workouts/api/serializers.py index 1f72621..07b767c 100644 --- a/vrobbler/apps/workouts/api/serializers.py +++ b/vrobbler/apps/workouts/api/serializers.py @@ -1,5 +1,10 @@ from rest_framework import serializers -from workouts.models import Exercise, WorkoutRoutine +from workouts.models import ( + Exercise, + WorkoutImport, + WorkoutRoutine, + WorkoutRoutineExercise, +) class ExerciseSerializer(serializers.HyperlinkedModelSerializer): @@ -8,7 +13,41 @@ class ExerciseSerializer(serializers.HyperlinkedModelSerializer): fields = "__all__" +class WorkoutRoutineExerciseSerializer(serializers.ModelSerializer): + exercise_name = serializers.CharField(source="exercise.name", read_only=True) + + class Meta: + model = WorkoutRoutineExercise + fields = [ + "id", + "exercise", + "exercise_name", + "sequence", + "target_sets", + "target_reps", + "target_weight_kg", + "enabled", + ] + + class WorkoutRoutineSerializer(serializers.HyperlinkedModelSerializer): + routine_exercises = WorkoutRoutineExerciseSerializer(many=True, read_only=True) + class Meta: model = WorkoutRoutine fields = "__all__" + + +class WorkoutImportSerializer(serializers.HyperlinkedModelSerializer): + class Meta: + model = WorkoutImport + fields = "__all__" + read_only_fields = ( + "uuid", + "processing_started", + "processed_finished", + "process_log", + "process_count", + "error_log", + "file_hash", + ) diff --git a/vrobbler/apps/workouts/api/views.py b/vrobbler/apps/workouts/api/views.py index 504a5b9..fa8e6ec 100644 --- a/vrobbler/apps/workouts/api/views.py +++ b/vrobbler/apps/workouts/api/views.py @@ -13,3 +13,9 @@ class WorkoutRoutineViewSet(viewsets.ModelViewSet): queryset = models.WorkoutRoutine.objects.all().order_by("-created") serializer_class = serializers.WorkoutRoutineSerializer permission_classes = [permissions.IsAuthenticated] + + +class WorkoutImportViewSet(viewsets.ModelViewSet): + queryset = models.WorkoutImport.objects.all().order_by("-created") + serializer_class = serializers.WorkoutImportSerializer + permission_classes = [permissions.IsAuthenticated] diff --git a/vrobbler/apps/workouts/importer.py b/vrobbler/apps/workouts/importer.py index 609666c..17a00ec 100644 --- a/vrobbler/apps/workouts/importer.py +++ b/vrobbler/apps/workouts/importer.py @@ -1,13 +1,24 @@ +import csv import json import logging import os +import sqlite3 import tarfile import tempfile import zipfile +from collections import Counter, defaultdict +from datetime import datetime, timezone as dt_timezone import requests from django.core.files.base import ContentFile -from workouts.models import Exercise +from scrobbles.models import Scrobble +from workouts.models import ( + Exercise, + WorkoutRoutine, + WorkoutRoutineExercise, + WorkoutSetEntry, +) +from workouts.utils import lbs_to_kg logger = logging.getLogger(__name__) @@ -16,6 +27,9 @@ TARBALL_URL = ( ) IMAGES_DIR_NAME = "images" +SESSION_GAP_MINUTES = 60 +DEFAULT_IMPORT_SOURCE = "Flexify Import" + def download_tarball(url=TARBALL_URL, destination=None): if not destination: @@ -142,3 +156,421 @@ def import_wrkout_exercises(tarball_path=None, import_images=True, dry_run=False "skipped": skipped, "image_failures": image_failures, } + + +FLEXIFY_EXERCISE_ALIASES = { + "Arnold press": "Arnold Dumbbell Press", + "Back extension": "Hyperextensions (Back Extensions)", + "Barbell bench press": "Barbell Bench Press - Medium Grip", + "Barbell bent-over row": "Bent Over Barbell Row", + "Barbell biceps curl": "Barbell Curl", + "Barbell shrug": "Barbell Shrug", + "Cable fly": "Flat Bench Cable Flyes", + "Cable lateral raise": "Cable Seated Lateral Raise", + "Cable pull-down": "Wide-Grip Lat Pulldown", + "Chest fly": "Dumbbell Flyes", + "Chin-up": "Chin-Up", + "Crunch": "Crunches", + "Deadlift": "Barbell Deadlift", + "Decline bench press": "Decline Barbell Bench Press", + "Dumbbell bench press": "Dumbbell Bench Press", + "Dumbbell bent-over row": "Bent Over Two-Dumbbell Row", + "Dumbbell biceps curl": "Dumbbell Bicep Curl", + "Dumbbell fly": "Dumbbell Flyes", + "Dumbbell lateral raise": "Side Lateral Raise", + "Dumbbell shoulder press": "Dumbbell Shoulder Press", + "Dumbbell shrug": "Dumbbell Shrug", + "Good morning": "Good Morning", + "Hanging leg raise": "Hanging Leg Raise", + "Hyperextension": "Hyperextensions (Back Extensions)", + "Incline bench press": "Barbell Incline Bench Press - Medium Grip", + "Lat pull-down": "Wide-Grip Lat Pulldown", + "Leg curl": "Seated Leg Curl", + "Leg extension": "Leg Extensions", + "Leg press": "Leg Press", + "Overhead triceps extension": "Cable Rope Overhead Triceps Extension", + "Pull-down": "Wide-Grip Lat Pulldown", + "Pull-up": "Pullups", + "Push-up": "Pushups", + "Reverse grip pull-down": "Underhand Cable Pulldowns", + "Reverse grip pushdown": "Reverse Grip Triceps Pushdown", + "Squat": "Barbell Squat", + "Standing calf raise": "Standing Calf Raises", + "T-bar row": "T-Bar Row with Handle", + "Triceps dip": "Dips - Triceps Version", + "Upright row": "Upright Barbell Row", +} + + +def resolve_exercise(name, dry_run=False): + """Map a Flexify exercise name to the closest wrkout Exercise. + + Returns (exercise, created). When no good alias exists the Flexify name + itself is used, creating the Exercise on first import. + """ + target = FLEXIFY_EXERCISE_ALIASES.get(name) or name + exercise = Exercise.objects.filter(name__iexact=target).first() + created = False + if not exercise: + created = True + if not dry_run: + exercise, _ = Exercise.objects.get_or_create(name=target) + return exercise, created + + +def read_plans_csv(path): + plans = [] + with open(path, newline="", encoding="utf-8-sig") as f: + for row in csv.DictReader(f): + exercises = [ + e.strip() for e in (row.get("exercises") or "").split(";") if e.strip() + ] + plans.append( + { + "days": (row.get("days") or "").strip(), + "title": (row.get("title") or "").strip(), + "sequence": row.get("sequence") or None, + "exercises": exercises, + } + ) + return plans + + +def _available_columns(con, table): + return {row[1] for row in con.execute(f"PRAGMA table_info({table})")} + + +def _select_rows(con, table, columns, where=None, order_by=None): + existing = [c for c in columns if c in _available_columns(con, table)] + if not existing: + return [] + sql = f"SELECT {', '.join(existing)} FROM {table}" + if where: + sql += f" WHERE {where}" + if order_by: + sql += f" ORDER BY {order_by}" + return [dict(zip(existing, row)) for row in con.execute(sql)] + + +def read_flexify_sqlite(path): + try: + con = sqlite3.connect(f"file:{path}?mode=ro", uri=True) + except sqlite3.Error as exc: + raise ValueError(f"Not a readable SQLite file: {path}") from exc + con.row_factory = sqlite3.Row + try: + plans = _select_rows( + con, "plans", ["id", "days", "sequence", "title"], order_by="id" + ) + plan_exercises = _select_rows( + con, + "plan_exercises", + ["plan_id", "exercise", "sequence", "enabled"], + order_by="plan_id, sequence", + ) + sets = _select_rows( + con, + "gym_sets", + [ + "id", + "name", + "weight", + "reps", + "unit", + "created", + "plan_id", + "body_weight", + "notes", + "hidden", + "cardio", + "distance", + "duration", + ], + where="hidden = 0", + order_by="created, id", + ) + catalog = [ + r["name"] + for r in _select_rows( + con, "gym_sets", ["name"], where="hidden = 1", order_by="name" + ) + ] + strength_unit = None + if "settings" in [ + r[0] + for r in con.execute("SELECT name FROM sqlite_master WHERE type='table'") + ]: + row = con.execute("SELECT strength_unit FROM settings LIMIT 1").fetchone() + if row and row[0]: + strength_unit = row[0] + return { + "plans": plans, + "plan_exercises": plan_exercises, + "sets": sets, + "catalog": catalog, + "strength_unit": strength_unit, + } + except sqlite3.Error as exc: + raise ValueError(f"Could not read Flexify data from {path}: {exc}") from exc + finally: + con.close() + + +def cluster_sessions(sets, gap_minutes=SESSION_GAP_MINUTES): + """Group logged sets into sessions: a new session starts when the plan_id + changes or the gap between consecutive sets exceeds gap_minutes.""" + sessions = [] + current = None + for s in sorted(sets, key=lambda x: (x.get("created") or 0, x.get("id") or 0)): + if current is None: + current = [] + sessions.append(current) + else: + gap = (s.get("created") or 0) - (current[-1].get("created") or 0) + if s.get("plan_id") != current[-1].get("plan_id") or gap > gap_minutes * 60: + current = [] + sessions.append(current) + current.append(s) + return sessions + + +def _to_kg(value, unit): + if value is None: + return None + try: + value = float(value) + except (TypeError, ValueError): + return None + if (unit or "").lower() == "lb": + return lbs_to_kg(value) + return round(value, 2) + + +def _whole(value): + try: + value = float(value) + except (TypeError, ValueError): + return None + return int(value) if value.is_integer() else value + + +def aggregate_workouts(session, dry_run=False): + by_name = defaultdict(list) + for s in session: + by_name[s["name"]].append(s) + + workouts = [] + for name, rows in by_name.items(): + exercise, _ = resolve_exercise(name, dry_run=dry_run) + reps_counter = Counter(_whole(r.get("reps")) for r in rows if r.get("reps")) + weights = [ + _to_kg(r.get("weight"), r.get("unit")) + for r in rows + if r.get("weight") not in (None, "") + ] + weight_kg = max(weights) if weights else None + if weight_kg == 0: + weight_kg = None + notes = ( + "; ".join(str(r["notes"]).strip() for r in rows if r.get("notes")) or None + ) + workouts.append( + WorkoutSetEntry( + exercise_id=exercise.id if exercise else None, + sets=len(rows), + reps=reps_counter.most_common(1)[0][0] if reps_counter else None, + weight_kg=weight_kg, + notes=notes, + ).asdict + ) + return workouts + + +def aggregate_bodyweight(session): + values = [] + for r in session: + kg = _to_kg(r.get("body_weight"), r.get("unit")) + if kg: + values.append(round(kg, 2)) + if not values: + return None + return Counter(values).most_common(1)[0][0] + + +def _find_or_create_routine(title, counts): + routine = WorkoutRoutine.objects.filter(title__iexact=title).first() + if routine: + counts["routines_updated"] += 1 + else: + routine = WorkoutRoutine.objects.create(title=title) + counts["routines_created"] += 1 + return routine + + +def _plan_title(plan_def, fallback="Workout"): + title = plan_def.get("title") or plan_def.get("days") or fallback + return title.strip() or fallback + + +def import_flexify( + sqlite_path=None, + csv_path=None, + user_id=None, + dry_run=False, + source=DEFAULT_IMPORT_SOURCE, +): + """Import Flexify plans and workout history. + + Plans come from the CSV (preferred, has titles) and/or the sqlite + ``plans``/``plan_exercises`` tables. Logged ``gym_sets`` become workout + scrobbles, clustered into sessions by time gap and plan. + """ + sqlite_data = read_flexify_sqlite(sqlite_path) if sqlite_path else None + csv_plans = read_plans_csv(csv_path) if csv_path else [] + if sqlite_data is None and not csv_plans: + raise ValueError("Provide --sqlite and/or --csv") + + counts = { + "routines_created": 0, + "routines_updated": 0, + "exercises_created": 0, + "scrobbles_created": 0, + "scrobbles_skipped": 0, + "scrobbles_existing": 0, + } + created_scrobbles = [] + + plan_defs = [] + seen_days = set() + for p in csv_plans: + plan_defs.append( + {"days": p["days"], "title": p["title"], "exercises": p["exercises"]} + ) + seen_days.add(p["days"].lower()) + if sqlite_data: + by_plan = defaultdict(list) + for pe in sqlite_data["plan_exercises"]: + by_plan[pe["plan_id"]].append(pe["exercise"]) + for p in sqlite_data["plans"]: + if p.get("days") and p["days"].lower() in seen_days: + continue + plan_defs.append( + { + "days": p.get("days") or "", + "title": p.get("title") or "", + "exercises": by_plan.get(p["id"], []), + } + ) + if p.get("days"): + seen_days.add(p["days"].lower()) + + for plan_def in plan_defs: + title = _plan_title(plan_def) + if dry_run: + if not WorkoutRoutine.objects.filter(title__iexact=title).exists(): + counts["routines_created"] += 1 + else: + counts["routines_updated"] += 1 + else: + _find_or_create_routine(title, counts) + for name in plan_def["exercises"]: + _, created = resolve_exercise(name, dry_run=dry_run) + if created: + counts["exercises_created"] += 1 + + if sqlite_data: + plans_by_id = {p["id"]: p for p in sqlite_data["plans"]} + routine_title_by_days = {} + for plan_def in plan_defs: + if plan_def["days"]: + routine_title_by_days[plan_def["days"].lower()] = _plan_title(plan_def) + for session in cluster_sessions(sqlite_data["sets"]): + if not session: + continue + plan_def = plans_by_id.get(session[0].get("plan_id")) + if plan_def and plan_def.get("days"): + title = routine_title_by_days.get( + plan_def["days"].lower(), _plan_title(plan_def) + ) + elif plan_def: + title = _plan_title(plan_def) + else: + title = "Flexify Workout" + routine = None + if not dry_run: + routine = WorkoutRoutine.objects.filter(title__iexact=title).first() + if not routine: + routine = WorkoutRoutine.objects.create(title=title) + counts["routines_created"] += 1 + + workouts = aggregate_workouts(session, dry_run=dry_run) + if not workouts: + counts["scrobbles_skipped"] += 1 + continue + + timestamp = datetime.fromtimestamp( + session[0]["created"], tz=dt_timezone.utc + ).replace(microsecond=0) + if ( + not dry_run + and routine + and Scrobble.objects.filter( + user_id=user_id, + workout_routine=routine, + timestamp=timestamp, + source=source, + ).exists() + ): + counts["scrobbles_existing"] += 1 + continue + + bodyweight_kg = aggregate_bodyweight(session) + duration_minutes = int( + (session[-1]["created"] - session[0]["created"]) / 60 + ) + log = {"workouts": workouts} + if bodyweight_kg: + log["bodyweight_kg"] = bodyweight_kg + if duration_minutes: + log["duration_minutes"] = duration_minutes + + counts["scrobbles_created"] += 1 + if dry_run: + continue + scrobble = Scrobble.objects.create( + user_id=user_id, + workout_routine=routine, + media_type=Scrobble.MediaType.WORKOUT, + timestamp=timestamp, + played_to_completion=True, + log=log, + source=source, + ) + created_scrobbles.append(scrobble) + + if not dry_run: + for plan_def in plan_defs: + routine = WorkoutRoutine.objects.filter( + title__iexact=_plan_title(plan_def) + ).first() + if not routine: + continue + for sequence, name in enumerate(plan_def["exercises"]): + exercise, _ = resolve_exercise(name) + we, _ = WorkoutRoutineExercise.objects.get_or_create( + routine=routine, exercise=exercise + ) + we.sequence = sequence + we.enabled = True + we.save() + + logger.info(f"Flexify import finished: {counts}") + return { + "routines_created": counts["routines_created"], + "routines_updated": counts["routines_updated"], + "exercises_created": counts["exercises_created"], + "scrobbles_created": counts["scrobbles_created"], + "scrobbles_skipped": counts["scrobbles_skipped"], + "scrobbles_existing": counts["scrobbles_existing"], + "scrobbles": created_scrobbles, + } diff --git a/vrobbler/apps/workouts/management/commands/import_flexify.py b/vrobbler/apps/workouts/management/commands/import_flexify.py new file mode 100644 index 0000000..280b85f --- /dev/null +++ b/vrobbler/apps/workouts/management/commands/import_flexify.py @@ -0,0 +1,66 @@ +from django.contrib.auth import get_user_model +from django.core.management.base import BaseCommand, CommandError +from workouts.importer import import_flexify + +User = get_user_model() + + +class Command(BaseCommand): + help = "Import workouts and routines from a Flexify backup" + + def add_arguments(self, parser): + parser.add_argument( + "--sqlite", + dest="sqlite", + default=None, + help="Path to a Flexify sqlite backup file", + ) + parser.add_argument( + "--csv", + dest="csv", + default=None, + help="Path to a Flexify plans CSV file", + ) + parser.add_argument( + "--user", + dest="user", + required=True, + help="Email of the user to attach the scrobbles to", + ) + parser.add_argument( + "--dry-run", + action="store_true", + help="Report what would be imported without writing anything", + ) + parser.add_argument( + "--source", + dest="source", + default="Flexify Import", + help="Scrobble source label (default: Flexify Import)", + ) + + def handle(self, *args, **options): + user = User.objects.filter(email=options["user"]).first() + if not user: + raise CommandError(f"No user found with email {options['user']}") + + if not options.get("sqlite") and not options.get("csv"): + raise CommandError("Provide --sqlite and/or --csv") + + result = import_flexify( + sqlite_path=options.get("sqlite"), + csv_path=options.get("csv"), + user_id=user.id, + dry_run=options.get("dry_run"), + source=options.get("source"), + ) + self.stdout.write( + self.style.SUCCESS( + f"Import finished: {result['routines_created']} routines created, " + f"{result['routines_updated']} routines updated, " + f"{result['exercises_created']} exercises created, " + f"{result['scrobbles_created']} scrobbles created, " + f"{result['scrobbles_existing']} scrobbles already present, " + f"{result['scrobbles_skipped']} skipped" + ) + ) diff --git a/vrobbler/apps/workouts/mcp.py b/vrobbler/apps/workouts/mcp.py index 5b12cfb..0b21277 100644 --- a/vrobbler/apps/workouts/mcp.py +++ b/vrobbler/apps/workouts/mcp.py @@ -66,4 +66,19 @@ def _routine_to_dict(r: WorkoutRoutine) -> dict: result = {"uuid": str(r.uuid), "title": r.title} if r.description: result["description"] = r.description + exercises = [] + for we in r.routine_exercises.all(): + exercises.append( + { + "sequence": we.sequence, + "exercise_uuid": str(we.exercise.uuid), + "exercise_name": we.exercise.name, + "target_sets": we.target_sets, + "target_reps": we.target_reps, + "target_weight_kg": we.target_weight_kg, + "enabled": we.enabled, + } + ) + if exercises: + result["planned_exercises"] = exercises return result diff --git a/vrobbler/apps/workouts/migrations/0002_workoutroutineexercise_workoutimport.py b/vrobbler/apps/workouts/migrations/0002_workoutroutineexercise_workoutimport.py new file mode 100644 index 0000000..7997481 --- /dev/null +++ b/vrobbler/apps/workouts/migrations/0002_workoutroutineexercise_workoutimport.py @@ -0,0 +1,134 @@ +# Generated by Django 4.2.29 on 2026-08-07 16:27 + +import uuid + +import django.db.models.deletion +import django_extensions.db.fields +import workouts.models +from django.conf import settings +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + migrations.swappable_dependency(settings.AUTH_USER_MODEL), + ("workouts", "0001_initial"), + ] + + operations = [ + migrations.CreateModel( + name="WorkoutRoutineExercise", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ( + "created", + django_extensions.db.fields.CreationDateTimeField( + auto_now_add=True, verbose_name="created" + ), + ), + ( + "modified", + django_extensions.db.fields.ModificationDateTimeField( + auto_now=True, verbose_name="modified" + ), + ), + ("sequence", models.PositiveIntegerField(default=0)), + ("target_sets", models.PositiveIntegerField(blank=True, null=True)), + ("target_reps", models.PositiveIntegerField(blank=True, null=True)), + ("target_weight_kg", models.FloatField(blank=True, null=True)), + ("enabled", models.BooleanField(default=True)), + ( + "exercise", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="routine_exercises", + to="workouts.exercise", + ), + ), + ( + "routine", + models.ForeignKey( + on_delete=django.db.models.deletion.CASCADE, + related_name="routine_exercises", + to="workouts.workoutroutine", + ), + ), + ], + options={ + "ordering": ["routine", "sequence"], + }, + ), + migrations.CreateModel( + name="WorkoutImport", + fields=[ + ( + "id", + models.BigAutoField( + auto_created=True, + primary_key=True, + serialize=False, + verbose_name="ID", + ), + ), + ( + "created", + django_extensions.db.fields.CreationDateTimeField( + auto_now_add=True, verbose_name="created" + ), + ), + ( + "modified", + django_extensions.db.fields.ModificationDateTimeField( + auto_now=True, verbose_name="modified" + ), + ), + ("uuid", models.UUIDField(default=uuid.uuid4, editable=False)), + ("processing_started", models.DateTimeField(blank=True, null=True)), + ("processed_finished", models.DateTimeField(blank=True, null=True)), + ("process_log", models.TextField(blank=True, null=True)), + ("process_count", models.IntegerField(blank=True, null=True)), + ("error_log", models.TextField(blank=True, null=True)), + ( + "sqlite_file", + models.FileField( + blank=True, + null=True, + upload_to=workouts.models.WorkoutImport.get_path, + ), + ), + ( + "csv_file", + models.FileField( + blank=True, + null=True, + upload_to=workouts.models.WorkoutImport.get_path, + ), + ), + ( + "original_filename", + models.CharField(blank=True, max_length=255, null=True), + ), + ( + "user", + models.ForeignKey( + blank=True, + null=True, + on_delete=django.db.models.deletion.DO_NOTHING, + to=settings.AUTH_USER_MODEL, + ), + ), + ], + options={ + "verbose_name": "Workout Import", + }, + ), + ] diff --git a/vrobbler/apps/workouts/migrations/0003_workoutimport_file_hash.py b/vrobbler/apps/workouts/migrations/0003_workoutimport_file_hash.py new file mode 100644 index 0000000..7972101 --- /dev/null +++ b/vrobbler/apps/workouts/migrations/0003_workoutimport_file_hash.py @@ -0,0 +1,18 @@ +# Generated by Django 4.2.29 on 2026-08-07 16:34 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("workouts", "0002_workoutroutineexercise_workoutimport"), + ] + + operations = [ + migrations.AddField( + model_name="workoutimport", + name="file_hash", + field=models.CharField(blank=True, max_length=32, null=True), + ), + ] diff --git a/vrobbler/apps/workouts/models.py b/vrobbler/apps/workouts/models.py index 625f2e0..e399973 100644 --- a/vrobbler/apps/workouts/models.py +++ b/vrobbler/apps/workouts/models.py @@ -4,6 +4,7 @@ from functools import cached_property from typing import Optional from uuid import uuid4 +from django.conf import settings from django.db import models from django.urls import reverse from django_extensions.db.models import TimeStampedModel @@ -11,6 +12,7 @@ from imagekit.models import ImageSpecField from imagekit.processors import ResizeToFit from scrobbles.dataclasses import BaseLogData, WithPeopleLogData from scrobbles.mixins import ScrobblableConstants, ScrobblableMixin +from scrobbles.models import BaseFileImportMixin from workouts.utils import display_weight logger = logging.getLogger(__name__) @@ -197,3 +199,69 @@ class WorkoutRoutine(ScrobblableMixin): if not routine: routine = cls.objects.create(title=title) return routine + + +class WorkoutRoutineExercise(TimeStampedModel): + routine = models.ForeignKey( + WorkoutRoutine, related_name="routine_exercises", on_delete=models.CASCADE + ) + exercise = models.ForeignKey( + Exercise, related_name="routine_exercises", on_delete=models.CASCADE + ) + sequence = models.PositiveIntegerField(default=0) + target_sets = models.PositiveIntegerField(**BNULL) + target_reps = models.PositiveIntegerField(**BNULL) + target_weight_kg = models.FloatField(**BNULL) + enabled = models.BooleanField(default=True) + + class Meta: + ordering = ["routine", "sequence"] + + def __str__(self): + return f"{self.routine.title}: {self.exercise.name}" + + +class WorkoutImport(BaseFileImportMixin): + class Meta: + verbose_name = "Workout Import" + + def get_path(instance, filename): + extension = filename.split(".")[-1] + uuid = instance.uuid + return f"workout-imports/{uuid}.{extension}" + + @property + def import_type(self) -> str: + return "Flexify" + + @property + def upload_file_path(self): + if getattr(settings, "USE_S3_STORAGE"): + return self.sqlite_file.url + return self.sqlite_file.path + + sqlite_file = models.FileField(upload_to=get_path, **BNULL) + csv_file = models.FileField(upload_to=get_path, **BNULL) + original_filename = models.CharField(max_length=255, **BNULL) + file_hash = models.CharField(max_length=32, **BNULL) + + def process(self, force=False): + from workouts.importer import import_flexify + + if self.processed_finished and not force: + logger.info(f"{self} already processed on {self.processed_finished}") + return + + self.mark_started() + try: + result = import_flexify( + sqlite_path=self.upload_file_path, + csv_path=self.csv_file.path if self.csv_file else None, + user_id=self.user_id, + ) + self.record_log(result["scrobbles"]) + except Exception as exc: + self.record_error(f"Import failed: {exc}") + logger.exception(f"Import failed for {self}") + finally: + self.mark_finished() diff --git a/vrobbler/templates/workouts/workoutroutine_detail.html b/vrobbler/templates/workouts/workoutroutine_detail.html index 69b9625..e0b3c1a 100644 --- a/vrobbler/templates/workouts/workoutroutine_detail.html +++ b/vrobbler/templates/workouts/workoutroutine_detail.html @@ -17,6 +17,23 @@
+{% if object.routine_exercises.all %} +