282 lines
11 KiB
Python
282 lines
11 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Schlanker Datenvertrag fuer den Trainingsplan-Editor."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import re
|
|
from typing import Any
|
|
|
|
PLAN_SCHEMA_VERSION = 4
|
|
CONTRACT_VERSION = 3
|
|
|
|
|
|
def slug(value: Any) -> str:
|
|
text = str(value or "").strip().lower()
|
|
text = text.replace("ä", "ae").replace("ö", "oe").replace("ü", "ue").replace("ß", "ss")
|
|
return re.sub(r"[^a-z0-9]+", "-", text).strip("-") or "item"
|
|
|
|
|
|
def stable_id(prefix: str, *parts: Any) -> str:
|
|
seed = "|".join(str(part or "") for part in parts)
|
|
digest = hashlib.sha1(seed.encode("utf-8")).hexdigest()[:10]
|
|
return f"{prefix}-{slug(parts[-1] if parts else prefix)[:30]}-{digest}"
|
|
|
|
|
|
def default_config() -> dict[str, Any]:
|
|
return {
|
|
"schema_version": PLAN_SCHEMA_VERSION,
|
|
"contract_version": CONTRACT_VERSION,
|
|
"plan_id": "neuer-trainingsplan",
|
|
"meta": {"title": "Neuer Trainingsplan"},
|
|
"exercises": [],
|
|
"days": [],
|
|
}
|
|
|
|
|
|
def _string(value: Any) -> str:
|
|
return str(value or "").strip()
|
|
|
|
|
|
def _int(value: Any, fallback: int = 0) -> int:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return fallback
|
|
|
|
|
|
def _text_items(value: Any, fallback: Any = None) -> dict[str, list[str]]:
|
|
source = value if value not in (None, "") else fallback
|
|
if isinstance(source, dict):
|
|
source = source.get("items")
|
|
if isinstance(source, list):
|
|
items = [_string(item) for item in source if _string(item)]
|
|
else:
|
|
text = _string(source)
|
|
parts = re.split(r"<br\s*/?>|\n|·", text, flags=re.I)
|
|
items = [_string(re.sub(r"<[^>]+>", "", item)) for item in parts]
|
|
items = [item for item in items if item]
|
|
return {"items": items}
|
|
|
|
|
|
def _normalize_exercise(raw: Any, plan_id: str, index: int) -> dict[str, Any] | None:
|
|
if not isinstance(raw, dict):
|
|
return None
|
|
name = _string(raw.get("name") or raw.get("title"))
|
|
exercise_id = _string(raw.get("id") or raw.get("exercise_id"))
|
|
if not exercise_id:
|
|
exercise_id = stable_id("exercise", plan_id, index, name or "uebung")
|
|
exercise = {"id": exercise_id, "name": name}
|
|
for key in ("cue", "description", "equipment", "notes"):
|
|
value = _string(raw.get(key))
|
|
if value:
|
|
exercise[key] = value
|
|
tags = raw.get("tags")
|
|
if isinstance(tags, list):
|
|
clean_tags = [_string(tag) for tag in tags if _string(tag)]
|
|
if clean_tags:
|
|
exercise["tags"] = clean_tags
|
|
return exercise
|
|
|
|
|
|
def _catalog_sources(cfg: dict[str, Any]) -> list[Any]:
|
|
sources: list[Any] = []
|
|
raw_exercises = cfg.get("exercises")
|
|
if isinstance(raw_exercises, list):
|
|
sources.extend(raw_exercises)
|
|
raw_catalog = cfg.get("exercise_catalog")
|
|
if isinstance(raw_catalog, dict):
|
|
sources.extend(raw_catalog.values())
|
|
elif isinstance(raw_catalog, list):
|
|
sources.extend(raw_catalog)
|
|
return sources
|
|
|
|
|
|
def _legacy_day_exercises(day: dict[str, Any]) -> list[dict[str, Any]]:
|
|
result: list[dict[str, Any]] = []
|
|
direct = day.get("exercises")
|
|
if isinstance(direct, list):
|
|
result.extend(item for item in direct if isinstance(item, dict))
|
|
rotations = day.get("rotations")
|
|
if not isinstance(rotations, list):
|
|
rotations = []
|
|
if isinstance(day.get("rotA"), list):
|
|
rotations.append({"label": "Rotation A", "exercises": day["rotA"]})
|
|
if isinstance(day.get("rotB"), list):
|
|
rotations.append({"label": "Rotation B", "exercises": day["rotB"]})
|
|
for rotation in rotations:
|
|
if not isinstance(rotation, dict):
|
|
continue
|
|
label = _string(rotation.get("label"))
|
|
exercises = rotation.get("exercises") if isinstance(rotation.get("exercises"), list) else []
|
|
for exercise in exercises:
|
|
if not isinstance(exercise, dict):
|
|
continue
|
|
item = copy.deepcopy(exercise)
|
|
item["__legacy_catalog"] = True
|
|
if label and not _string(item.get("notes")):
|
|
item["notes"] = label
|
|
result.append(item)
|
|
return result
|
|
|
|
|
|
def _ensure_catalog_entry(
|
|
catalog: dict[str, dict[str, Any]],
|
|
raw: dict[str, Any],
|
|
plan_id: str,
|
|
index: int,
|
|
) -> str:
|
|
exercise_id = _string(raw.get("exercise_id") or raw.get("id"))
|
|
name = _string(raw.get("name") or raw.get("title"))
|
|
if not exercise_id:
|
|
exercise_id = stable_id("exercise", plan_id, index, name or "uebung")
|
|
if exercise_id not in catalog:
|
|
source = {
|
|
"id": exercise_id,
|
|
"name": name,
|
|
"cue": raw.get("cue"),
|
|
"description": raw.get("description"),
|
|
"equipment": raw.get("equipment"),
|
|
"tags": raw.get("tags"),
|
|
}
|
|
catalog[exercise_id] = _normalize_exercise(source, plan_id, index) or {
|
|
"id": exercise_id,
|
|
"name": name,
|
|
}
|
|
return exercise_id
|
|
|
|
|
|
def _normalize_day_exercise(raw: dict[str, Any], day_id: str, exercise_id: str, index: int) -> dict[str, Any]:
|
|
item_id = _string(raw.get("id"))
|
|
if not item_id or item_id == exercise_id:
|
|
item_id = stable_id("day-exercise", day_id, exercise_id, index)
|
|
item: dict[str, Any] = {"id": item_id, "exercise_id": exercise_id}
|
|
for key in ("sets", "duration_seconds"):
|
|
value = _int(raw.get(key), 0)
|
|
if value > 0:
|
|
item[key] = value
|
|
for key in ("reps", "tempo", "rest", "notes", "legacy_id"):
|
|
value = _string(raw.get(key))
|
|
if value:
|
|
item[key] = value
|
|
return item
|
|
|
|
|
|
def normalize_training_config(config: Any, plan_id: str | None = None) -> dict[str, Any]:
|
|
cfg = copy.deepcopy(config) if isinstance(config, dict) else {}
|
|
meta = cfg.get("meta") if isinstance(cfg.get("meta"), dict) else {}
|
|
resolved_plan_id = _string(cfg.get("plan_id") or plan_id or stable_id("plan", meta.get("title") or "training"))
|
|
title = _string(meta.get("title") or cfg.get("title") or "Neuer Trainingsplan")
|
|
normalized: dict[str, Any] = {
|
|
"schema_version": PLAN_SCHEMA_VERSION,
|
|
"contract_version": CONTRACT_VERSION,
|
|
"plan_id": resolved_plan_id,
|
|
"meta": {"title": title},
|
|
"exercises": [],
|
|
"days": [],
|
|
}
|
|
subtitle = _string(meta.get("subtitle") or cfg.get("subtitle"))
|
|
if subtitle:
|
|
normalized["meta"]["subtitle"] = subtitle
|
|
|
|
catalog: dict[str, dict[str, Any]] = {}
|
|
for index, raw in enumerate(_catalog_sources(cfg)):
|
|
exercise = _normalize_exercise(raw, resolved_plan_id, index)
|
|
if exercise:
|
|
catalog[exercise["id"]] = exercise
|
|
|
|
legacy_prepost = cfg.get("prepost") if isinstance(cfg.get("prepost"), dict) else {}
|
|
days = cfg.get("days") if isinstance(cfg.get("days"), list) else []
|
|
exercise_counter = len(catalog)
|
|
for day_index, raw_day in enumerate(days):
|
|
if not isinstance(raw_day, dict):
|
|
continue
|
|
day_num = _int(raw_day.get("num"), day_index + 1) or day_index + 1
|
|
name = _string(raw_day.get("name") or raw_day.get("focus") or f"Tag {day_num}")
|
|
day_id = _string(raw_day.get("id")) or stable_id("day", resolved_plan_id, day_num, name)
|
|
prepost = legacy_prepost.get(str(day_num)) if isinstance(legacy_prepost.get(str(day_num)), dict) else {}
|
|
day: dict[str, Any] = {
|
|
"id": day_id,
|
|
"num": day_num,
|
|
"name": name,
|
|
"warmup": _text_items(raw_day.get("warmup"), prepost.get("warmup")),
|
|
"cooldown": _text_items(raw_day.get("cooldown"), prepost.get("cooldown")),
|
|
"exercises": [],
|
|
}
|
|
focus = _string(raw_day.get("focus"))
|
|
if focus and focus != name:
|
|
day["focus"] = focus
|
|
notes = _string(raw_day.get("notes") or raw_day.get("note"))
|
|
if notes:
|
|
day["notes"] = notes
|
|
|
|
for item_index, raw_item in enumerate(_legacy_day_exercises(raw_day)):
|
|
should_create = bool(raw_item.pop("__legacy_catalog", False) or _string(raw_item.get("name")))
|
|
if should_create:
|
|
exercise_id = _ensure_catalog_entry(catalog, raw_item, resolved_plan_id, exercise_counter)
|
|
exercise_counter += 1
|
|
else:
|
|
exercise_id = _string(raw_item.get("exercise_id"))
|
|
day["exercises"].append(_normalize_day_exercise(raw_item, day_id, exercise_id, item_index))
|
|
normalized["days"].append(day)
|
|
|
|
normalized["exercises"] = sorted(catalog.values(), key=lambda item: (slug(item.get("name")), item.get("id")))
|
|
normalized["days"] = sorted(normalized["days"], key=lambda item: item.get("num", 0))
|
|
return normalized
|
|
|
|
|
|
def migrate_training_config(config: Any, plan_id: str | None = None) -> dict[str, Any]:
|
|
return normalize_training_config(config, plan_id=plan_id)
|
|
|
|
|
|
def validate_training_config(config: Any) -> dict[str, Any]:
|
|
cfg = normalize_training_config(config)
|
|
errors: list[dict[str, Any]] = []
|
|
warnings: list[dict[str, Any]] = []
|
|
ids: dict[str, str] = {}
|
|
|
|
def register(value: Any, location: str) -> None:
|
|
sid = _string(value)
|
|
if not sid:
|
|
errors.append({"code": "missing_id", "location": location, "message": "Stabile ID fehlt."})
|
|
elif sid in ids:
|
|
errors.append({"code": "duplicate_id", "location": location, "message": f"ID wird bereits bei {ids[sid]} verwendet."})
|
|
else:
|
|
ids[sid] = location
|
|
|
|
if not _string(cfg.get("meta", {}).get("title")):
|
|
errors.append({"code": "empty_title", "location": "Plan", "message": "Plantitel fehlt."})
|
|
|
|
exercise_ids: set[str] = set()
|
|
for index, exercise in enumerate(cfg.get("exercises", [])):
|
|
location = f"Übung {index + 1}"
|
|
register(exercise.get("id"), location)
|
|
exercise_ids.add(_string(exercise.get("id")))
|
|
if not _string(exercise.get("name")):
|
|
errors.append({"code": "empty_exercise", "location": location, "message": "Übungsname fehlt."})
|
|
|
|
if not cfg.get("days"):
|
|
warnings.append({"code": "no_days", "location": "Plan", "message": "Noch keine Trainingstage angelegt."})
|
|
|
|
for day_index, day in enumerate(cfg.get("days", [])):
|
|
location = f"Tag {day_index + 1}"
|
|
register(day.get("id"), location)
|
|
if not _string(day.get("name")):
|
|
errors.append({"code": "empty_day", "location": location, "message": "Tagesname fehlt."})
|
|
if not day.get("exercises"):
|
|
warnings.append({"code": "empty_day", "location": location, "message": "Tag enthält keine Übungen."})
|
|
for item_index, item in enumerate(day.get("exercises", [])):
|
|
item_location = f"{location}, Übung {item_index + 1}"
|
|
register(item.get("id"), item_location)
|
|
exercise_id = _string(item.get("exercise_id"))
|
|
if exercise_id not in exercise_ids:
|
|
errors.append({"code": "unknown_exercise", "location": item_location, "message": "Hinterlegte Übung existiert nicht in der Übungsbibliothek."})
|
|
sets = _int(item.get("sets"), 0)
|
|
if "sets" in item and not 1 <= sets <= 40:
|
|
errors.append({"code": "invalid_sets", "location": item_location, "message": "Sätze müssen zwischen 1 und 40 liegen."})
|
|
duration = _int(item.get("duration_seconds"), 0)
|
|
if "duration_seconds" in item and duration <= 0:
|
|
errors.append({"code": "invalid_duration", "location": item_location, "message": "Dauer muss größer als 0 sein."})
|
|
|
|
return {"errors": errors, "warnings": warnings, "config": cfg}
|