[tasks] Add weigh-in task importer

vrobbler/apps/scrobbles/importers/scale.py

@@ -0,0 +1,111 @@
+import csv
+import logging
+from datetime import datetime, timedelta
+
+from django.contrib.auth import get_user_model
+
+from scrobbles.models import Scrobble
+from tasks.models import Task
+
+logger = logging.getLogger(__name__)
+
+FLOAT_COLUMNS = {
+    "WEIGHT": "weight_kg",
+    "BMI": "bmi",
+    "BODY_FAT": "body_fat_pct",
+    "BONE": "bone_kg",
+    "MUSCLE": "muscle_kg",
+    "WATER": "water_pct",
+    "VISCERAL_FAT": "visceral_fat",
+    "WAIST": "waist_cm",
+    "CALORIES": "calories",
+    "BMR": "bmr",
+    "TDEE": "tdee",
+    "HEART_RATE": "heart_rate",
+    "CHEST": "chest_cm",
+    "BICEPS": "biceps_cm",
+    "NECK": "neck_cm",
+    "THIGH": "thigh_cm",
+    "HIPS": "hips_cm",
+    "CALIPER": "caliper_mm",
+    "CALIPER_1": "caliper_1_mm",
+    "CALIPER_2": "caliper_2_mm",
+    "CALIPER_3": "caliper_3_mm",
+    "LBM": "lbm_kg",
+    "WHR": "whr",
+    "WHTR": "whtr",
+}
+
+
+def parse_float(value):
+    if not value:
+        return None
+    try:
+        return round(float(value), 2)
+    except (ValueError, TypeError):
+        return None
+
+
+def import_scale_csv(file_path, user_id):
+    user = get_user_model().objects.get(id=user_id)
+    weigh_in = Task.find_or_create("Weigh-in")
+    new_scrobbles = []
+
+    with open(file_path, newline="", encoding="utf-8-sig") as f:
+        reader = csv.DictReader(f)
+        for row in reader:
+            date_str = (row.get("DATE") or "").strip()
+            time_str = (row.get("TIME") or "").strip()
+            if not date_str or not time_str:
+                logger.warning("Skipping row with no DATE or TIME")
+                continue
+
+            try:
+                dt = datetime.strptime(f"{date_str} {time_str}", "%Y-%m-%d %H:%M:%S.%f")
+            except ValueError:
+                try:
+                    dt = datetime.strptime(f"{date_str} {time_str}", "%Y-%m-%d %H:%M:%S")
+                except ValueError:
+                    logger.warning(f"Could not parse date/time: {date_str} {time_str}")
+                    continue
+
+            dt = dt.replace(microsecond=0)
+            stop = user.profile.get_timestamp_with_tz(dt)
+            start = stop - timedelta(minutes=5)
+
+            log_dict = {}
+            for csv_col, log_key in FLOAT_COLUMNS.items():
+                val = parse_float(row.get(csv_col))
+                if val is not None:
+                    log_dict[log_key] = val
+
+            comment = (row.get("COMMENT") or "").strip()
+            if comment:
+                log_dict["comment"] = comment
+
+            existing = Scrobble.objects.filter(
+                timestamp=start,
+                task=weigh_in,
+                user=user,
+            ).first()
+            if existing:
+                logger.debug(f"Skipping existing scrobble for {start}")
+                continue
+
+            new_scrobble = Scrobble(
+                user=user,
+                timestamp=start,
+                stop_timestamp=stop,
+                timezone=stop.tzinfo.name,
+                source="openScale CSV Import",
+                task=weigh_in,
+                log=log_dict,
+                played_to_completion=True,
+                in_progress=False,
+                media_type=Scrobble.MediaType.TASK,
+            )
+            new_scrobbles.append(new_scrobble)
+
+    created = Scrobble.objects.bulk_create(new_scrobbles)
+    logger.info(f"Created {len(created)} weigh-in scrobbles")
+    return created