123 lines
3.7 KiB
Python
123 lines
3.7 KiB
Python
import csv
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
|
|
from django.contrib.auth import get_user_model
|
|
|
|
from scrobbles.models import Scrobble
|
|
from scrobbles.notifications import ScrobbleNtfyNotification
|
|
from tasks.models import Task
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
FLOAT_COLUMNS = {
|
|
"WEIGHT": "weight",
|
|
"BMI": "bmi",
|
|
"BODY_FAT": "body_fat",
|
|
"BONE": "bone",
|
|
"MUSCLE": "muscle",
|
|
"WATER": "water",
|
|
"VISCERAL_FAT": "visceral_fat",
|
|
"WAIST": "waist",
|
|
"CALORIES": "calories",
|
|
"BMR": "bmr",
|
|
"TDEE": "tdee",
|
|
"HEART_RATE": "heart_rate",
|
|
"CHEST": "chest",
|
|
"BICEPS": "biceps",
|
|
"NECK": "neck",
|
|
"THIGH": "thigh",
|
|
"HIPS": "hips",
|
|
"CALIPER": "caliper",
|
|
"CALIPER_1": "caliper_1",
|
|
"CALIPER_2": "caliper_2",
|
|
"CALIPER_3": "caliper_3",
|
|
"LBM": "lbm",
|
|
"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
|
|
|
|
log_dict["unit_type"] = user.profile.weigh_in_units
|
|
|
|
weight_val = log_dict.get("weight")
|
|
if weight_val is not None:
|
|
unit = "kg" if user.profile.weigh_in_units == "metric" else "lbs"
|
|
log_dict["title"] = f"{weight_val} {unit}"
|
|
|
|
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,
|
|
visibility="private",
|
|
)
|
|
new_scrobbles.append(new_scrobble)
|
|
|
|
created = Scrobble.objects.bulk_create(new_scrobbles)
|
|
logger.info(f"Created {len(created)} weigh-in scrobbles")
|
|
for scrobble in created:
|
|
ScrobbleNtfyNotification(scrobble).send()
|
|
return created
|