350 lines
14 KiB
Python
350 lines
14 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""Schlanker Datenvertrag fuer den Trainingsplan-Editor."""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import hashlib
|
|
import html
|
|
import re
|
|
from typing import Any
|
|
|
|
PLAN_SCHEMA_VERSION = 5
|
|
CONTRACT_VERSION = 4
|
|
|
|
|
|
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 _html_text(value: Any) -> str:
|
|
return html.unescape(_string(value))
|
|
|
|
|
|
def _legacy_stage_lookup(cfg: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
|
stages = cfg.get("stages") if isinstance(cfg.get("stages"), dict) else {}
|
|
lookup: dict[str, dict[str, Any]] = {}
|
|
for key, stage in stages.items():
|
|
if not isinstance(stage, dict):
|
|
continue
|
|
for candidate in (key, stage.get("key"), stage.get("id")):
|
|
sid = _string(candidate)
|
|
if sid:
|
|
lookup[sid] = stage
|
|
return lookup
|
|
|
|
|
|
def _legacy_stage_for(raw: dict[str, Any], lookup: dict[str, dict[str, Any]]) -> dict[str, Any] | None:
|
|
for candidate in (raw.get("key"), raw.get("progression_id"), raw.get("stage_id")):
|
|
sid = _string(candidate)
|
|
if sid and sid in lookup:
|
|
return lookup[sid]
|
|
return None
|
|
|
|
|
|
def _progression_sources(raw: dict[str, Any], stage: dict[str, Any] | None) -> list[Any]:
|
|
for key in ("progressions", "variants"):
|
|
value = raw.get(key)
|
|
if isinstance(value, list) and value:
|
|
return value
|
|
if isinstance(stage, dict) and isinstance(stage.get("steps"), list):
|
|
return stage["steps"]
|
|
return []
|
|
|
|
|
|
def _normalize_progressions(
|
|
raw: dict[str, Any],
|
|
plan_id: str,
|
|
exercise_id: str,
|
|
stage: dict[str, Any] | None,
|
|
) -> tuple[list[dict[str, Any]], str]:
|
|
progressions: list[dict[str, Any]] = []
|
|
seen: set[str] = set()
|
|
selected = _string(raw.get("current_progression_id"))
|
|
selected_name = _html_text(raw.get("current_progression"))
|
|
|
|
for index, item in enumerate(_progression_sources(raw, stage)):
|
|
if isinstance(item, dict):
|
|
name = _html_text(item.get("name") or item.get("title"))
|
|
else:
|
|
name = _html_text(item)
|
|
if not name:
|
|
continue
|
|
fingerprint = slug(name)
|
|
if fingerprint in seen:
|
|
continue
|
|
seen.add(fingerprint)
|
|
progression_id = _string(item.get("id")) if isinstance(item, dict) else ""
|
|
if not progression_id:
|
|
progression_id = stable_id("progression", plan_id, exercise_id, index, name)
|
|
progressions.append({"id": progression_id, "name": name})
|
|
if selected_name and slug(selected_name) == fingerprint:
|
|
selected = progression_id
|
|
|
|
ids = {item["id"] for item in progressions}
|
|
if selected not in ids:
|
|
selected = progressions[0]["id"] if progressions else ""
|
|
return progressions, selected
|
|
|
|
|
|
def _exercise_identity(raw: dict[str, Any], plan_id: str, index: int, name: str) -> str:
|
|
exercise_id = _string(raw.get("exercise_id") or raw.get("id"))
|
|
if exercise_id:
|
|
return exercise_id
|
|
key = _string(raw.get("key") or raw.get("progression_id"))
|
|
if key:
|
|
return stable_id("exercise", plan_id, key)
|
|
return stable_id("exercise", plan_id, slug(name or "uebung"))
|
|
|
|
|
|
def _normalize_exercise(
|
|
raw: Any,
|
|
plan_id: str,
|
|
index: int,
|
|
stage_lookup: dict[str, dict[str, Any]] | None = None,
|
|
) -> dict[str, Any] | None:
|
|
if not isinstance(raw, dict):
|
|
return None
|
|
name = _html_text(raw.get("name") or raw.get("title"))
|
|
exercise_id = _exercise_identity(raw, plan_id, index, name)
|
|
exercise = {"id": exercise_id, "name": name}
|
|
stage = _legacy_stage_for(raw, stage_lookup or {})
|
|
progressions, selected = _normalize_progressions(raw, plan_id, exercise_id, stage)
|
|
if progressions:
|
|
exercise["progressions"] = progressions
|
|
exercise["current_progression_id"] = selected
|
|
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
|
|
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
|
|
result.append(item)
|
|
return result
|
|
|
|
|
|
def _ensure_catalog_entry(
|
|
catalog: dict[str, dict[str, Any]],
|
|
raw: dict[str, Any],
|
|
plan_id: str,
|
|
index: int,
|
|
stage_lookup: dict[str, dict[str, Any]],
|
|
) -> str:
|
|
name = _html_text(raw.get("name") or raw.get("title"))
|
|
exercise_id = _exercise_identity(raw, plan_id, index, name)
|
|
if exercise_id not in catalog:
|
|
source = {**raw, "id": exercise_id, "name": name}
|
|
catalog[exercise_id] = _normalize_exercise(source, plan_id, index, stage_lookup) or {
|
|
"id": exercise_id,
|
|
"name": name,
|
|
}
|
|
elif not catalog[exercise_id].get("progressions"):
|
|
source = {**raw, "id": exercise_id, "name": name}
|
|
exercise = _normalize_exercise(source, plan_id, index, stage_lookup)
|
|
if exercise and exercise.get("progressions"):
|
|
catalog[exercise_id]["progressions"] = exercise["progressions"]
|
|
catalog[exercise_id]["current_progression_id"] = exercise["current_progression_id"]
|
|
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)
|
|
return {"id": item_id, "exercise_id": exercise_id}
|
|
|
|
|
|
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]] = {}
|
|
stage_lookup = _legacy_stage_lookup(cfg)
|
|
for index, raw in enumerate(_catalog_sources(cfg)):
|
|
exercise = _normalize_exercise(raw, resolved_plan_id, index, stage_lookup)
|
|
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": [],
|
|
}
|
|
seen_day_exercises: set[str] = set()
|
|
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, stage_lookup)
|
|
exercise_counter += 1
|
|
else:
|
|
exercise_id = _string(raw_item.get("exercise_id"))
|
|
if exercise_id in seen_day_exercises:
|
|
continue
|
|
seen_day_exercises.add(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."})
|
|
progression_ids: set[str] = set()
|
|
for progression_index, progression in enumerate(exercise.get("progressions", [])):
|
|
progression_location = f"{location}, Progression {progression_index + 1}"
|
|
register(progression.get("id"), progression_location)
|
|
progression_ids.add(_string(progression.get("id")))
|
|
if not _string(progression.get("name")):
|
|
errors.append({"code": "empty_progression", "location": progression_location, "message": "Progressionsname fehlt."})
|
|
current_progression_id = _string(exercise.get("current_progression_id"))
|
|
if current_progression_id and current_progression_id not in progression_ids:
|
|
errors.append({"code": "unknown_progression", "location": location, "message": "Aktuelle Progression existiert nicht in dieser Übung."})
|
|
|
|
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."})
|
|
|
|
return {"errors": errors, "warnings": warnings, "config": cfg}
|