Hotfix

plugins/trainingstracker/app.py

@@ -1234,6 +1234,8 @@ def _default_tracker(plan: dict[str, Any]) -> dict[str, Any]:
         "revision": 1,
         "source_file": plan["source_file"],
         "source_hash": plan["source_hash"],
+        "plan_id": plan.get("plan_id"),
+        "plan_revision": plan.get("published_revision"),
         "profile": {
             "start_date": "",
             "display_name": "",
@@ -1274,6 +1276,7 @@ def _load_tracker(plan: dict[str, Any]) -> dict[str, Any]:
     tracker = _read_json(_tracker_path(plan["source_file"]), None)
     if not isinstance(tracker, dict):
         return _default_tracker(plan)
+    original_tracker = deepcopy(tracker)
     tracker.setdefault("version", TRACKER_SCHEMA_VERSION)
     tracker.setdefault("revision", 1)
     tracker.setdefault("profile", {})
@@ -1290,7 +1293,18 @@ def _load_tracker(plan: dict[str, Any]) -> dict[str, Any]:
     tracker.pop("analyses", None)
     tracker["version"] = TRACKER_SCHEMA_VERSION
     tracker["source_file"] = plan["source_file"]
-    tracker["source_changed"] = tracker.get("source_hash") not in (None, plan["source_hash"])
+    source_changed = tracker.get("source_hash") not in (None, plan["source_hash"])
+    tracker, migrated = _migrate_tracker_sessions(plan, tracker)
+    if migrated:
+        tracker["revision"] = int(tracker.get("revision") or 1) + 1
+        tracker["updated_at"] = _utc_now()
+        persisted = deepcopy(tracker)
+        persisted.pop("source_changed", None)
+        backup_path = SESSIONS_DIR / ".migration-backups" / f"{plan['source_file']}.pre-result-data-v2.json"
+        if not backup_path.exists():
+            _atomic_json_write(backup_path, original_tracker)
+        _atomic_json_write(_tracker_path(plan["source_file"]), persisted)
+    tracker["source_changed"] = source_changed
     return tracker
 
 
@@ -1394,6 +1408,347 @@ def _format_result_data(value: Any) -> str:
     return f"{prefix}{joined(data['values'])}{unit}{side}".strip()
 
 
+def _item_has_user_data(item: Any) -> bool:
+    if not isinstance(item, dict):
+        return False
+    status = str(item.get("completion_status") or ("completed" if item.get("done") else "planned"))
+    if status != "planned" or bool(item.get("done")):
+        return True
+    if isinstance(item.get("result_data"), dict):
+        return True
+    return any(_has_text(item.get(field)) for field in (
+        "result", "note", "progression", "exercise_name", "progression_id", "skip_reason"
+    ))
+
+
+def _merge_migrated_item(legacy: dict[str, Any], stable: dict[str, Any]) -> dict[str, Any]:
+    """Vereinigt doppelte Legacy-/Stable-ID-Einträge ohne befüllte Werte zu verlieren.
+
+    Ein durch die neue UI angelegter leerer Stable-ID-Platzhalter darf einen
+    bereits abgeschlossenen Legacy-Eintrag insbesondere nicht wieder auf
+    ``planned``/``done=false`` zurücksetzen.
+    """
+    merged = deepcopy(legacy)
+    legacy_status = str(legacy.get("completion_status") or ("completed" if legacy.get("done") else "planned"))
+    stable_status = str(stable.get("completion_status") or ("completed" if stable.get("done") else "planned"))
+    for key, value in stable.items():
+        if key not in merged:
+            merged[key] = deepcopy(value)
+            continue
+        empty = value is None or value == "" or value == [] or value == {}
+        if empty:
+            continue
+        if key in {"done", "completion_status"} and stable_status == "planned" and legacy_status != "planned":
+            continue
+        merged[key] = deepcopy(value)
+    if stable_status != "planned":
+        merged["completion_status"] = stable_status
+        merged["done"] = stable_status in {"completed", "partial"}
+    elif legacy_status != "planned":
+        merged["completion_status"] = legacy_status
+        merged["done"] = legacy_status in {"completed", "partial"}
+    return merged
+
+
+def _exercise_lookup_for_day(plan: dict[str, Any], day_num: int) -> tuple[dict[str, dict[str, Any]], dict[str, str]]:
+    by_id: dict[str, dict[str, Any]] = {}
+    legacy_to_stable: dict[str, str] = {}
+    day = next((entry for entry in plan.get("days", []) if isinstance(entry, dict) and int(entry.get("num") or 0) == day_num), None)
+    if not isinstance(day, dict):
+        return by_id, legacy_to_stable
+    for rotation in day.get("rotations", []) if isinstance(day.get("rotations"), list) else []:
+        if not isinstance(rotation, dict):
+            continue
+        for exercise in rotation.get("exercises", []) if isinstance(rotation.get("exercises"), list) else []:
+            if not isinstance(exercise, dict):
+                continue
+            stable_id = str(exercise.get("id") or "")
+            legacy_id = str(exercise.get("legacy_id") or "")
+            if stable_id:
+                by_id[stable_id] = exercise
+            if legacy_id and stable_id:
+                by_id[legacy_id] = exercise
+                legacy_to_stable[legacy_id] = stable_id
+    return by_id, legacy_to_stable
+
+
+def _progression_step_for_item(plan: dict[str, Any], exercise: dict[str, Any], item: dict[str, Any]) -> dict[str, Any] | None:
+    progression_id = str(item.get("progression_id") or exercise.get("progression_id") or "")
+    stage = plan.get("stages", {}).get(progression_id) if isinstance(plan.get("stages"), dict) else None
+    steps = stage.get("steps") if isinstance(stage, dict) and isinstance(stage.get("steps"), list) else []
+    wanted_id = str(item.get("progression_step_id") or "")
+    if wanted_id:
+        found = next((step for step in steps if isinstance(step, dict) and str(step.get("id") or "") == wanted_id), None)
+        if found:
+            return found
+    wanted_name = _plain_text(item.get("progression") or "").casefold()
+    if wanted_name:
+        return next((
+            step for step in steps
+            if isinstance(step, dict) and _plain_text(step.get("name") or "").casefold() == wanted_name
+        ), None)
+    return None
+
+
+def _resolved_schema_for_item(plan: dict[str, Any], exercise: dict[str, Any], item: dict[str, Any]) -> dict[str, Any]:
+    schema = deepcopy(exercise.get("result_schema") if isinstance(exercise.get("result_schema"), dict) else {})
+    step = _progression_step_for_item(plan, exercise, item)
+    if isinstance(step, dict) and isinstance(step.get("result_schema"), dict):
+        schema.update(step["result_schema"])
+    manual = item.get("result_format")
+    if isinstance(manual, dict):
+        for key in ("mode", "weight_mode", "laterality", "sides_mode", "sets", "default_sets", "locked_sets"):
+            if key in manual:
+                schema[key] = manual[key]
+    mode = str(schema.get("mode") or "auto")
+    if mode not in {"auto", "reps", "seconds", "minutes", "none"}:
+        mode = "auto"
+    weight_mode = str(schema.get("weight_mode") or "none")
+    if weight_mode not in {"none", "optional", "required"}:
+        weight_mode = "none"
+    laterality = str(schema.get("laterality") or "bilateral")
+    if laterality not in {"bilateral", "unilateral"}:
+        laterality = "bilateral"
+    sides_mode = str(schema.get("sides_mode") or "same")
+    if sides_mode not in {"same", "separate"}:
+        sides_mode = "same"
+    try:
+        sets = max(1, min(20, int(schema.get("sets") or schema.get("default_sets") or 1)))
+    except (TypeError, ValueError):
+        sets = 1
+    return {
+        "mode": mode,
+        "weight_mode": weight_mode,
+        "laterality": laterality,
+        "sides_mode": sides_mode if laterality == "unilateral" else "same",
+        "default_sets": sets,
+        "locked_sets": bool(schema.get("locked_sets")),
+    }
+
+
+def _parse_legacy_result_data(value: Any, schema: dict[str, Any]) -> tuple[dict[str, Any] | None, dict[str, Any] | None]:
+    """Überführt eindeutige alte Freitextergebnisse verlustfrei in result_data 2.
+
+    Die historische Eingabe ist für die konkrete Session bindend. Weicht ihre
+    eindeutig erkennbare Messart vom heutigen Plan ab, wird deshalb zusätzlich
+    ein manueller ``result_format``-Override gespeichert, statt Werte zu verwerfen.
+    """
+    raw = str(value or "").strip()
+    if not raw:
+        return None, None
+    low = raw.lower().replace("×", "x")
+    explicit_mode: str | None = None
+    if re.search(r"(?:\d|\s)(?:min\.?|minuten)\b", low, flags=re.I):
+        explicit_mode = "minutes"
+    elif re.search(r"(?:\d|\s)(?:s|sek\.?|sekunden)\b", low, flags=re.I):
+        explicit_mode = "seconds"
+    recommended_mode = str(schema.get("mode") or "auto")
+    mode = explicit_mode or (recommended_mode if recommended_mode in {"reps", "seconds", "minutes"} else "reps")
+
+    weight_match = re.search(r"(\d+(?:[.,]\d+)?)\s*kg\b", low, flags=re.I)
+    weight = _result_number(weight_match.group(1)) if weight_match else None
+
+    def number_list(part: Any) -> list[float | None]:
+        return [
+            number for number in (_result_number(match.group(0)) for match in re.finditer(r"\d+(?:[.,]\d+)?", str(part or "")))
+            if number is not None
+        ]
+
+    def clean_part(part: Any) -> str:
+        text = re.sub(r"\d+(?:[.,]\d+)?\s*kg\b", " ", str(part or ""), flags=re.I)
+        return re.sub(
+            r"\b(?:reps?|wiederholungen|s|sek\.?|sekunden|min\.?|minuten|je seite|pro seite)\b",
+            " ", text, flags=re.I,
+        )
+
+    separate = re.search(
+        r"\bL(?:inks)?\s*:?\s*([^·;|]+)[·;|]\s*R(?:echts)?\s*:?\s*(.+)$",
+        raw, flags=re.I,
+    )
+    if separate:
+        left = number_list(clean_part(separate.group(1)))
+        right = number_list(clean_part(separate.group(2)))
+        if not left and not right:
+            return None, None
+        sets = max(len(left), len(right), 1)
+        data = _sanitize_result_data({
+            "version": RESULT_DATA_VERSION,
+            "mode": mode,
+            "laterality": "unilateral",
+            "sides_mode": "separate",
+            "sets": sets,
+            "weight_kg": weight,
+            "values": [],
+            "left_values": left,
+            "right_values": right,
+        })
+    else:
+        work = clean_part(raw)
+        repeated = re.search(r"(\d+(?:[.,]\d+)?)\s*x\s*(\d+(?:[.,]\d+)?)", work, flags=re.I)
+        if repeated:
+            count = max(1, min(20, int(float(repeated.group(1).replace(",", ".")))))
+            repeated_value = _result_number(repeated.group(2))
+            values = [repeated_value] * count if repeated_value is not None else []
+        else:
+            values = number_list(work)
+        if not values and weight is None:
+            return None, None
+        per_side = bool(re.search(r"(?:je|pro)\s+seite|/\s*seite", low, flags=re.I))
+        laterality = "unilateral" if per_side else str(schema.get("laterality") or "bilateral")
+        if laterality not in {"bilateral", "unilateral"}:
+            laterality = "bilateral"
+        # Eine einzige historische Zahlenreihe ist nie eine belastbare Links-/
+        # Rechts-Trennung. Bei einseitigen Übungen bedeutet sie daher "je Seite".
+        sides_mode = "same"
+        sets = max(1, min(20, len(values) or int(schema.get("default_sets") or 1)))
+        data = _sanitize_result_data({
+            "version": RESULT_DATA_VERSION,
+            "mode": mode,
+            "laterality": laterality,
+            "sides_mode": sides_mode,
+            "sets": sets,
+            "weight_kg": weight,
+            "values": values,
+            "left_values": [],
+            "right_values": [],
+        })
+    if data is None:
+        return None, None
+
+    actual_weight_mode = str(schema.get("weight_mode") or "none")
+    if weight is not None and actual_weight_mode == "none":
+        actual_weight_mode = "optional"
+    differs = (
+        data["mode"] != recommended_mode
+        or data["laterality"] != str(schema.get("laterality") or "bilateral")
+        or data["sides_mode"] != str(schema.get("sides_mode") or "same")
+        or actual_weight_mode != str(schema.get("weight_mode") or "none")
+    )
+    result_format = None
+    if differs:
+        result_format = {
+            "mode": data["mode"],
+            "laterality": data["laterality"],
+            "sides_mode": data["sides_mode"],
+            "weight_mode": actual_weight_mode,
+        }
+    return data, result_format
+
+
+def _migrate_tracker_sessions(plan: dict[str, Any], tracker: dict[str, Any]) -> tuple[dict[str, Any], bool]:
+    """Migriert alle vorhandenen Sessions serverseitig und atomar auf stabile IDs.
+
+    Dadurch hängt die Datenmigration nicht davon ab, welche Session zufällig im
+    Browser geöffnet oder verändert wurde. Uneindeutige Freitexte bleiben als
+    ``result`` erhalten; eindeutige Werte werden zusätzlich strukturiert gespeichert.
+    """
+    changed = False
+    tracker = deepcopy(tracker)
+    desired_root = {
+        "version": TRACKER_SCHEMA_VERSION,
+        "source_file": plan.get("source_file"),
+        "source_hash": plan.get("source_hash"),
+        "plan_id": plan.get("plan_id"),
+        "plan_revision": plan.get("published_revision"),
+    }
+    for key, value in desired_root.items():
+        if tracker.get(key) != value:
+            tracker[key] = value
+            changed = True
+    sessions = tracker.get("sessions") if isinstance(tracker.get("sessions"), dict) else {}
+    if tracker.get("sessions") is not sessions:
+        tracker["sessions"] = sessions
+        changed = True
+
+    for session_key, session in sessions.items():
+        if not isinstance(session, dict):
+            continue
+        _, day_num = _session_key_parts(str(session_key))
+        if day_num == 9999:
+            continue
+        by_id, legacy_to_stable = _exercise_lookup_for_day(plan, day_num)
+        items = session.get("items") if isinstance(session.get("items"), dict) else {}
+        if session.get("items") is not items:
+            session["items"] = items
+            changed = True
+
+        for legacy_id, stable_id in list(legacy_to_stable.items()):
+            if legacy_id not in items or legacy_id == stable_id:
+                continue
+            legacy_item = items.pop(legacy_id)
+            if isinstance(legacy_item, dict):
+                stable_item = items.get(stable_id)
+                items[stable_id] = _merge_migrated_item(
+                    legacy_item,
+                    stable_item if isinstance(stable_item, dict) else {},
+                )
+            changed = True
+
+        for item_id, item in list(items.items()):
+            if not isinstance(item, dict):
+                continue
+            exercise = by_id.get(str(item_id))
+            status = str(item.get("completion_status") or ("completed" if item.get("done") else "planned"))
+            if status not in VALID_ITEM_STATUSES:
+                status = "planned"
+            if item.get("completion_status") != status:
+                item["completion_status"] = status
+                changed = True
+            done = status in {"completed", "partial"}
+            if bool(item.get("done")) != done:
+                item["done"] = done
+                changed = True
+            if status != "skipped" and "skip_reason" in item:
+                item.pop("skip_reason", None)
+                changed = True
+
+            if isinstance(exercise, dict) and _item_has_user_data(item):
+                metadata = {
+                    "progression_id": str(item.get("progression_id") or exercise.get("progression_id") or ""),
+                    "exercise_name": _plain_text(item.get("exercise_name") or exercise.get("name") or ""),
+                }
+                for key, value in metadata.items():
+                    if value and item.get(key) != value:
+                        item[key] = value
+                        changed = True
+                step = _progression_step_for_item(plan, exercise, item)
+                if isinstance(step, dict) and step.get("id") and not item.get("progression_step_id"):
+                    item["progression_step_id"] = str(step["id"])
+                    changed = True
+
+            if isinstance(item.get("result_data"), dict):
+                data = _sanitize_result_data(item.get("result_data"))
+                if data is not None:
+                    formatted = _format_result_data(data)
+                    if item.get("result_data") != data:
+                        item["result_data"] = data
+                        changed = True
+                    if item.get("result") != formatted:
+                        item["result"] = formatted
+                        changed = True
+                else:
+                    item.pop("result_data", None)
+                    changed = True
+            elif isinstance(exercise, dict) and _has_text(item.get("result")):
+                schema = _resolved_schema_for_item(plan, exercise, item)
+                data, result_format = _parse_legacy_result_data(item.get("result"), schema)
+                if data is not None:
+                    item["result_data"] = data
+                    item["result"] = _format_result_data(data)
+                    if result_format is not None:
+                        item["result_format"] = result_format
+                    changed = True
+
+        for key, value in {
+            "plan_id": plan.get("plan_id"),
+            "plan_revision": plan.get("published_revision"),
+        }.items():
+            if session.get(key) != value:
+                session[key] = value
+                changed = True
+    return tracker, changed
+
+
 def _normalize_tracker_payload(plan: dict[str, Any], incoming: Any, existing: dict[str, Any]) -> dict[str, Any]:
     if not isinstance(incoming, dict):
         raise ValueError("Tracker-Daten müssen ein JSON-Objekt sein")
@@ -1436,6 +1791,7 @@ def _normalize_tracker_payload(plan: dict[str, Any], incoming: Any, existing: di
     for week, status in list(clean["week_statuses"].items()):
         if not isinstance(status, dict): clean["week_statuses"].pop(week, None); continue
         status["status"] = "closed" if status.get("status") == "closed" else "open"
+    clean, _ = _migrate_tracker_sessions(plan, clean)
     return clean