3321 lines
156 KiB
Python
3321 lines
156 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""Mobile Tracking-App für boehmitools-Trainingspläne.
|
||
|
||
Quelle, strikt read-only:
|
||
<boehmitools>/data/trainingsplan/plans/*.json
|
||
|
||
Tracker-Daten:
|
||
<boehmitools>/data/trainingstracker/sessions/<Original-Dateiname>.json
|
||
|
||
Analyse-Daten, vollständig getrennt von den Sessions. Pro Woche und
|
||
Gesamtplan existiert jeweils nur eine überschreibbare Datei:
|
||
<boehmitools>/data/trainingstracker/analyses/<Original-Dateiname>/
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import html
|
||
import json
|
||
import os
|
||
import re
|
||
import tempfile
|
||
import threading
|
||
import time
|
||
from contextlib import contextmanager
|
||
from copy import deepcopy
|
||
from datetime import date, datetime, timedelta, timezone
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from flask import Flask, Response, jsonify, request, send_file
|
||
|
||
from contract import (PLAN_SCHEMA_VERSION, CONTRACT_VERSION, TRACKER_SCHEMA_VERSION, RESULT_DATA_VERSION, PROMPT_VERSION, ANALYSIS_SCHEMA_VERSION, PROGRESSION_NORMALIZATION_VERSION)
|
||
from session_store import merge_session_patch, VALID_ITEM_STATUSES
|
||
from analysis_store import current_filename, cleanup_legacy_files
|
||
|
||
PLUGIN_DIR = Path(__file__).resolve().parent
|
||
STATIC_DIR = PLUGIN_DIR / "static"
|
||
_WRITE_LOCK = threading.RLock()
|
||
_ANALYSIS_GUARD = threading.RLock()
|
||
_ANALYSIS_THREADS: dict[str, threading.Thread] = {}
|
||
_ANALYSIS_TIMEOUT_SECONDS = 3 * 60
|
||
_ANALYSIS_HEARTBEAT_SECONDS = 30
|
||
|
||
|
||
# Version der deterministischen, planunabhängigen Normalisierung. Eine Änderung
|
||
# macht bestehende Analysen über den Dataset-Hash automatisch veraltet.
|
||
|
||
# Die Faktoren sind keine medizinischen oder biomechanischen Naturkonstanten.
|
||
# Sie dienen ausschließlich dazu, Varianten desselben Bewegungsmusters in einer
|
||
# nachvollziehbaren Trendkurve abzubilden. Push-up-Anker orientieren sich an veröffentlichten Kraftplattenmessungen; die
|
||
# eigentliche Referenzskala ist bewusst nichtlinear, damit ein Stufenwechsel mit
|
||
# weniger Wiederholungen nicht fälschlich als Rückschritt erscheint. Alle übrigen
|
||
# Werte sind konservative Heuristiken aus Hebel, Unterstützung, ROM und Stufenfolge.
|
||
_EQUIVALENCE_TABLES: list[dict[str, Any]] = [
|
||
{
|
||
"id": "push_up", "label": "Push-up", "mode": "reps", "priority": 100,
|
||
"reference": "voller Push-up = 1,00 Referenz-Rep",
|
||
"variants": [
|
||
{"label": "Wand-Push-up", "factor": 0.05, "patterns": [r"wand", r"wall"]},
|
||
{"label": "Incline hoch, ca. 60 cm", "factor": 0.10, "patterns": [r"inkline.?\(?hoch", r"incline.?high", r"hände hoch"]},
|
||
{"label": "Incline mittel", "factor": 0.20, "patterns": [r"inkline.?mittel", r"incline.?medium", r"halbe höhe"]},
|
||
{"label": "Incline tief, ca. 30 cm", "factor": 0.35, "patterns": [r"inkline.?tief", r"incline.?low", r"tiefe erhöh"]},
|
||
{"label": "Incline tiefer / Knie", "factor": 0.40, "patterns": [r"tiefer.?/? ?knie"]},
|
||
{"label": "Knie-Push-up", "factor": 0.45, "patterns": [r"knie.?liegest", r"knee.?push", r"\bknie\b"]},
|
||
{"label": "Teilwiederholung", "factor": 0.35, "patterns": [r"teil.?rep", r"partial", r"kleiner rom"]},
|
||
{"label": "Negativ", "factor": 0.70, "patterns": [r"negativ", r"eccentric", r"absenk"]},
|
||
{"label": "Negativ + erste volle", "factor": 0.85, "patterns": [r"negativ.*erste volle"]},
|
||
{"label": "Voller Push-up", "factor": 1.00, "patterns": [r"volle? push", r"standard", r"normal", r"boden"]},
|
||
{"label": "Decline / Zusatzlast", "factor": 1.25, "patterns": [r"decline", r"füße erhöht", r"zusatzlast", r"gewichtete"]},
|
||
],
|
||
},
|
||
{
|
||
"id": "dip", "label": "Dip", "mode": "reps", "priority": 82,
|
||
"reference": "voller Dip am Barren = 1,00 Referenz-Rep",
|
||
"variants": [
|
||
{"label": "Bank-Dip, Knie stark gebeugt, kleiner ROM", "factor": 0.25, "patterns": [r"knie stark gebeugt.*klein"]},
|
||
{"label": "Bank-Dip, Knie gebeugt, voller ROM", "factor": 0.40, "patterns": [r"knie gebeugt.*voll"]},
|
||
{"label": "Bank-Dip, Füße etwas weiter", "factor": 0.55, "patterns": [r"füße etwas weiter"]},
|
||
{"label": "Bank-Dip, Füße weit / Beine gerade", "factor": 0.70, "patterns": [r"füße weiter", r"beine gerade"]},
|
||
{"label": "Assistierter Dip", "factor": 0.65, "patterns": [r"assistiert"]},
|
||
{"label": "Negativer Dip", "factor": 0.80, "patterns": [r"negativ"]},
|
||
{"label": "Voller Dip", "factor": 1.00, "patterns": [r"volle? dip", r"standard dip"]},
|
||
],
|
||
},
|
||
{
|
||
"id": "pike_press", "label": "Pike Push-up", "mode": "reps", "priority": 78,
|
||
"reference": "voller Pike Push-up = 1,00 Referenz-Rep",
|
||
"variants": [
|
||
{"label": "Hold + kleine Senkung", "factor": 0.30, "patterns": [r"kleine senkung", r"partial"]},
|
||
{"label": "Pike Push-up, Hände erhöht", "factor": 0.50, "patterns": [r"hände erhöht", r"erhöht"]},
|
||
{"label": "Pike Push-up am Boden, kleiner ROM", "factor": 0.75, "patterns": [r"boden.*klein", r"kleiner rom"]},
|
||
{"label": "Voller Pike Push-up", "factor": 1.00, "patterns": [r"voll", r"pike push"]},
|
||
],
|
||
},
|
||
{
|
||
"id": "inverted_row", "label": "Inverted Row", "mode": "reps", "priority": 92,
|
||
"reference": "Straight-Leg Row = 1,00 Referenz-Rep",
|
||
"variants": [
|
||
{"label": "Aufrechte assistierte Row", "factor": 0.25, "patterns": [r"aufrecht", r"stark assist"]},
|
||
{"label": "Bent-Leg Row", "factor": 0.50, "patterns": [r"bent.?leg", r"beine gebeugt", r"knie gebeugt"]},
|
||
{"label": "Straight-Leg Row", "factor": 1.00, "patterns": [r"straight.?leg", r"beine gestreckt", r"volle row"]},
|
||
{"label": "Füße erhöht", "factor": 1.15, "patterns": [r"füße erhöht", r"feet elevated"]},
|
||
],
|
||
},
|
||
{
|
||
"id": "pull_up", "label": "Pull-up", "mode": "reps", "priority": 100,
|
||
"reference": "voller Pull-up = 1,00 Referenz-Rep",
|
||
"variants": [
|
||
{"label": "Scapular Pull-up", "factor": 0.20, "patterns": [r"scapular"]},
|
||
{"label": "Stark assistiert / Fußhilfe", "factor": 0.35, "patterns": [r"fußhilfe", r"stark assist"]},
|
||
{"label": "Assistierter Pull-up", "factor": 0.55, "patterns": [r"assistiert", r"band"]},
|
||
{"label": "Negativer Pull-up", "factor": 0.70, "patterns": [r"negativ"]},
|
||
{"label": "Leicht assistiert", "factor": 0.80, "patterns": [r"leicht assist"]},
|
||
{"label": "Voller Pull-up", "factor": 1.00, "patterns": [r"volle? pull", r"klimmzug", r"standard"]},
|
||
{"label": "Pull-up mit Zusatzlast", "factor": 1.15, "patterns": [r"zusatzlast", r"weighted"]},
|
||
],
|
||
},
|
||
{
|
||
"id": "squat", "label": "Squat", "mode": "reps", "priority": 100,
|
||
"reference": "voller Bodyweight Squat = 1,00 Referenz-Rep",
|
||
"variants": [
|
||
{"label": "Chair / Box Squat", "factor": 0.35, "patterns": [r"chair", r"box", r"stuhl"]},
|
||
{"label": "Assistierter Squat", "factor": 0.55, "patterns": [r"assistiert", r"festhalten", r"türrahmen"]},
|
||
{"label": "Teilwiederholung", "factor": 0.45, "patterns": [r"teil", r"partial"]},
|
||
{"label": "Negativer Squat", "factor": 0.75, "patterns": [r"negativ", r"3s runter"]},
|
||
{"label": "Bodyweight Squat", "factor": 1.00, "patterns": [r"körpergewicht", r"bodyweight", r"volle squat", r"kniebeuge"]},
|
||
{"label": "Goblet Squat", "factor": 1.00, "patterns": [r"goblet"]},
|
||
{"label": "Einbeinige Variante", "factor": 1.45, "patterns": [r"pistol", r"einbeinig", r"shrimp"]},
|
||
],
|
||
},
|
||
{
|
||
"id": "plank", "label": "Plank", "mode": "seconds", "priority": 100,
|
||
"reference": "volle Plank = 1,00 Referenzsekunde",
|
||
"variants": [
|
||
{"label": "Erhöhte Plank", "factor": 0.55, "patterns": [r"erhöht", r"incline"]},
|
||
{"label": "Plank auf Knien", "factor": 0.65, "patterns": [r"knien", r"knie"]},
|
||
{"label": "Volle Plank", "factor": 1.00, "patterns": [r"voll", r"standard", r"high plank"]},
|
||
{"label": "Plank mit Schulter-Taps", "factor": 1.15, "patterns": [r"schulter.?tap", r"shoulder.?tap"]},
|
||
{"label": "Long-Lever / erschwerter Hebel", "factor": 1.25, "patterns": [r"long.?lever", r"hebel", r"schwieriger"]},
|
||
],
|
||
},
|
||
{
|
||
"id": "side_plank", "label": "Side Plank", "mode": "seconds", "priority": 82,
|
||
"reference": "volle Side Plank = 1,00 Referenzsekunde",
|
||
"variants": [
|
||
{"label": "Side Plank auf Knien", "factor": 0.65, "patterns": [r"knien", r"knie"]},
|
||
{"label": "Volle Side Plank", "factor": 1.00, "patterns": [r"voll", r"standard"]},
|
||
{"label": "Star / langer Hebel", "factor": 1.25, "patterns": [r"star", r"langer hebel", r"bein oben"]},
|
||
],
|
||
},
|
||
{
|
||
"id": "hollow_hold", "label": "Hollow Hold", "mode": "seconds", "priority": 82,
|
||
"reference": "voller Hollow Hold = 1,00 Referenzsekunde",
|
||
"variants": [
|
||
{"label": "Tuck", "factor": 0.65, "patterns": [r"tuck", r"knie an"]},
|
||
{"label": "Halb gestreckt", "factor": 0.82, "patterns": [r"halb"]},
|
||
{"label": "Voll", "factor": 1.00, "patterns": [r"voll", r"gestreckt"]},
|
||
],
|
||
},
|
||
{
|
||
"id": "active_hang", "label": "Active Hang", "mode": "seconds", "priority": 92,
|
||
"reference": "freier Active Hang = 1,00 Referenzsekunde",
|
||
"variants": [
|
||
{"label": "Mit Fußhilfe", "factor": 0.60, "patterns": [r"fußhilfe", r"assist"]},
|
||
{"label": "Freier Active Hang", "factor": 1.00, "patterns": [r"active hang", r"frei"]},
|
||
{"label": "Einarmig", "factor": 1.60, "patterns": [r"einarm"]},
|
||
],
|
||
},
|
||
{
|
||
"id": "deep_squat_hold", "label": "Deep Squat Hold", "mode": "seconds", "priority": 90,
|
||
"reference": "ungewichteter Deep Squat Hold = 1,00 Referenzsekunde",
|
||
"variants": [
|
||
{"label": "Assistiert", "factor": 0.75, "patterns": [r"assist", r"festhalten"]},
|
||
{"label": "Ungewichtet", "factor": 1.00, "patterns": [r"ungewichtet", r"bodyweight", r"deep squat"]},
|
||
{"label": "Gewichtet", "factor": 1.00, "patterns": [r"gewichtet", r"goblet"]},
|
||
],
|
||
},
|
||
{
|
||
"id": "wall_sit", "label": "Wall Sit", "mode": "seconds", "priority": 75,
|
||
"reference": "sauberer Wall Sit = 1,00 Referenzsekunde",
|
||
"variants": [{"label": "Wall Sit", "factor": 1.00, "patterns": [r"wall sit", r"wandsitz"]}],
|
||
},
|
||
]
|
||
|
||
_EQUIVALENCE_SOURCES = [
|
||
{
|
||
"title": "CaliHoss Calisthenics Course",
|
||
"url": "https://calihoss.me/learn/calisthenics/#playlist-2-progression",
|
||
"note": "Progression über Technik, ROM, Unterstützung, Hebel, Widerstand und mechanische Schwierigkeit; 4–8 Reps bzw. 10–30-s-Holds.",
|
||
},
|
||
{
|
||
"title": "Suprak et al. (2011): The effect of position on the percentage of body mass supported during traditional and modified push-up variants",
|
||
"url": "https://pubmed.ncbi.nlm.nih.gov/21273908/",
|
||
"note": "Biomechanischer Anker für die Reihenfolge traditioneller und Knie-Push-ups; die Analysefaktoren sind bewusst nichtlinear.",
|
||
},
|
||
{
|
||
"title": "Ebben et al. (2011): Kinetic analysis of several variations of push-ups",
|
||
"url": "https://pubmed.ncbi.nlm.nih.gov/21873902/",
|
||
"note": "Biomechanischer Anker für die Reihenfolge verschiedener Handhöhen; die Analysefaktoren sind bewusst nichtlinear.",
|
||
},
|
||
]
|
||
|
||
|
||
def _discover_boehmitools_root() -> Path:
|
||
explicit = os.environ.get("BOEHMITOOLS_ROOT")
|
||
if explicit:
|
||
return Path(explicit).expanduser().resolve()
|
||
for parent in [PLUGIN_DIR, *PLUGIN_DIR.parents]:
|
||
if (parent / "data" / "trainingsplan" / "plans").is_dir():
|
||
return parent
|
||
# Standardfall bei .../boehmitools/plugins/trainingstracker
|
||
if PLUGIN_DIR.parent.name == "plugins":
|
||
return PLUGIN_DIR.parent.parent
|
||
return PLUGIN_DIR
|
||
|
||
|
||
ROOT_DIR = _discover_boehmitools_root()
|
||
DATA_DIR = Path(os.environ.get(
|
||
"TRAININGSTRACKER_DATA_DIR", ROOT_DIR / "data" / "trainingstracker"
|
||
)).expanduser().resolve()
|
||
PLANS_DIR = Path(os.environ.get(
|
||
"TRAININGSTRACKER_PLANS_DIR", ROOT_DIR / "data" / "trainingsplan" / "plans"
|
||
)).expanduser().resolve()
|
||
SESSIONS_DIR = DATA_DIR / "sessions"
|
||
ANALYSES_DIR = DATA_DIR / "analyses"
|
||
PROPOSALS_DIR = DATA_DIR / "proposals"
|
||
SETTINGS_FILE = DATA_DIR / "settings.json"
|
||
SESSIONS_DIR.mkdir(parents=True, exist_ok=True)
|
||
ANALYSES_DIR.mkdir(parents=True, exist_ok=True)
|
||
PROPOSALS_DIR.mkdir(parents=True, exist_ok=True)
|
||
|
||
app = Flask(__name__, static_folder=None)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Dateisystem und JSON
|
||
# ---------------------------------------------------------------------------
|
||
def _utc_now() -> str:
|
||
return datetime.now(timezone.utc).isoformat(timespec="seconds")
|
||
|
||
|
||
def _atomic_json_write(path: Path, payload: Any) -> None:
|
||
"""Schreibt JSON atomar, damit auch ein abgebrochener Request nichts zerlegt."""
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with _WRITE_LOCK:
|
||
fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
|
||
try:
|
||
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
||
json.dump(payload, handle, ensure_ascii=False, indent=2)
|
||
handle.write("\n")
|
||
handle.flush()
|
||
os.fsync(handle.fileno())
|
||
os.replace(temp_name, path)
|
||
finally:
|
||
if os.path.exists(temp_name):
|
||
os.unlink(temp_name)
|
||
|
||
|
||
def _read_json(path: Path, default: Any = None) -> Any:
|
||
try:
|
||
with path.open("r", encoding="utf-8") as handle:
|
||
return json.load(handle)
|
||
except (FileNotFoundError, json.JSONDecodeError, OSError):
|
||
return deepcopy(default)
|
||
|
||
|
||
def _source_hash(path: Path) -> str:
|
||
digest = hashlib.sha256()
|
||
with path.open("rb") as handle:
|
||
for chunk in iter(lambda: handle.read(1024 * 128), b""):
|
||
digest.update(chunk)
|
||
return digest.hexdigest()
|
||
|
||
|
||
def _known_plan_files() -> dict[str, Path]:
|
||
"""Ermittelt valide Trainingspläne. Rezeptsammlungen werden ausgeschlossen."""
|
||
result: dict[str, Path] = {}
|
||
if not PLANS_DIR.is_dir():
|
||
return result
|
||
for path in sorted(PLANS_DIR.glob("*.json"), key=lambda p: p.name.casefold()):
|
||
try:
|
||
raw = _read_json(path)
|
||
_, cfg = _unwrap_plan(raw, path.stem)
|
||
if _is_training_config(cfg):
|
||
result[path.name] = path
|
||
except Exception:
|
||
continue
|
||
return result
|
||
|
||
|
||
def _validated_plan_path(plan_id: str) -> Path | None:
|
||
# Exakte Auswahl aus dem Scan verhindert Traversal und fremde Dateien.
|
||
return _known_plan_files().get(plan_id)
|
||
|
||
|
||
def _tracker_path(plan_filename: str) -> Path:
|
||
# Original-Dateiname im separaten sessions-Ordner, wie vorgegeben.
|
||
return SESSIONS_DIR / Path(plan_filename).name
|
||
|
||
|
||
def _analysis_plan_dir(plan_filename: str) -> Path:
|
||
# Exakter Plan-Dateiname als eigener Archivordner, ohne Annahmen zum Linux-Root.
|
||
return ANALYSES_DIR / Path(plan_filename).name
|
||
|
||
|
||
def _analysis_index_path(plan_filename: str) -> Path:
|
||
return _analysis_plan_dir(plan_filename) / "index.json"
|
||
|
||
|
||
def _analysis_state_path(plan_filename: str) -> Path:
|
||
return _analysis_plan_dir(plan_filename) / "state.json"
|
||
|
||
|
||
def _analysis_lock_path(plan_filename: str) -> Path:
|
||
return _analysis_plan_dir(plan_filename) / ".analysis.lock"
|
||
|
||
|
||
@contextmanager
|
||
def _analysis_file_lock(plan_filename: str):
|
||
"""Prozessübergreifende Sperre für Job-Claim und Statuswechsel auf Linux."""
|
||
import fcntl
|
||
path = _analysis_lock_path(plan_filename)
|
||
path.parent.mkdir(parents=True, exist_ok=True)
|
||
with path.open("a+", encoding="utf-8") as handle:
|
||
fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
|
||
try:
|
||
yield
|
||
finally:
|
||
fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
|
||
|
||
|
||
def _empty_analysis_index(plan_filename: str) -> dict[str, Any]:
|
||
return {
|
||
"version": 2,
|
||
"source_file": Path(plan_filename).name,
|
||
"latest": {"weeks": {}, "overall": None},
|
||
"updated_at": _utc_now(),
|
||
}
|
||
|
||
|
||
def _read_analysis_index(plan_filename: str) -> dict[str, Any]:
|
||
raw = _read_json(_analysis_index_path(plan_filename), None)
|
||
index = raw if isinstance(raw, dict) else _empty_analysis_index(plan_filename)
|
||
latest = index.get("latest") if isinstance(index.get("latest"), dict) else {}
|
||
index = {
|
||
"version": 2,
|
||
"source_file": Path(plan_filename).name,
|
||
"latest": {
|
||
"weeks": latest.get("weeks") if isinstance(latest.get("weeks"), dict) else {},
|
||
"overall": latest.get("overall") if isinstance(latest.get("overall"), str) else None,
|
||
},
|
||
"updated_at": str(index.get("updated_at") or _utc_now()),
|
||
}
|
||
return index
|
||
|
||
|
||
def _write_analysis_index(plan_filename: str, index: dict[str, Any]) -> None:
|
||
index = _sanitize(index)
|
||
if not isinstance(index, dict):
|
||
raise ValueError("Ungültiger Analyseindex")
|
||
index["version"] = 2
|
||
index["source_file"] = Path(plan_filename).name
|
||
index["updated_at"] = _utc_now()
|
||
index.pop("history", None)
|
||
_atomic_json_write(_analysis_index_path(plan_filename), index)
|
||
|
||
|
||
def _public_analysis_record(record: Any) -> dict[str, Any] | None:
|
||
if not isinstance(record, dict) or record.get("status") not in (None, "success"):
|
||
return None
|
||
response = record.get("response") if isinstance(record.get("response"), dict) else record.get("result")
|
||
if not isinstance(response, dict):
|
||
return None
|
||
keys = (
|
||
"id", "type", "week", "created_at", "model", "source_hash",
|
||
"sessions_considered", "weeks_considered", "archive_file", "analysis_kind",
|
||
)
|
||
public = {key: deepcopy(record.get(key)) for key in keys if key in record}
|
||
public["result"] = deepcopy(response)
|
||
public["visuals"] = deepcopy(record.get("visuals") if isinstance(record.get("visuals"), dict) else {})
|
||
return public
|
||
|
||
|
||
def _read_analysis_record(plan_filename: str, relative_name: Any) -> dict[str, Any] | None:
|
||
if not isinstance(relative_name, str) or not relative_name:
|
||
return None
|
||
base = _analysis_plan_dir(plan_filename).resolve()
|
||
candidate = (base / relative_name).resolve()
|
||
try:
|
||
candidate.relative_to(base)
|
||
except ValueError:
|
||
return None
|
||
raw = _read_json(candidate, None)
|
||
return raw if isinstance(raw, dict) else None
|
||
|
||
|
||
def _migrate_single_analysis_store(plan_filename: str) -> dict[str, Any]:
|
||
"""Reduziert ältere Analysearchive auf genau eine aktuelle Datei je Scope.
|
||
|
||
Alte Versionen führten Zeitstempeldateien und optional eine Historie im Index.
|
||
Beim ersten Lesen werden nur die im Index als aktuell markierten Datensätze
|
||
übernommen, auf deterministische Dateinamen umgeschrieben und alle älteren
|
||
Dateien desselben Scopes entfernt.
|
||
"""
|
||
with _WRITE_LOCK:
|
||
index = _read_analysis_index(plan_filename)
|
||
directory = _analysis_plan_dir(plan_filename)
|
||
changed = False
|
||
weeks = index["latest"]["weeks"]
|
||
for week_key, relative_name in list(weeks.items()):
|
||
record = _read_analysis_record(plan_filename, relative_name)
|
||
if not isinstance(record, dict):
|
||
weeks.pop(week_key, None)
|
||
changed = True
|
||
continue
|
||
try:
|
||
week = int(record.get("week") or week_key)
|
||
except (TypeError, ValueError):
|
||
weeks.pop(week_key, None)
|
||
changed = True
|
||
continue
|
||
record["type"] = "week"
|
||
record["week"] = week
|
||
target = current_filename(record)
|
||
if relative_name != target or not (directory / target).exists():
|
||
record["archive_file"] = target
|
||
_atomic_json_write(directory / target, record)
|
||
changed = True
|
||
cleanup_legacy_files(directory, record, target)
|
||
if week_key != str(week):
|
||
weeks.pop(week_key, None)
|
||
if weeks.get(str(week)) != target:
|
||
weeks[str(week)] = target
|
||
changed = True
|
||
|
||
overall_name = index["latest"].get("overall")
|
||
if overall_name:
|
||
record = _read_analysis_record(plan_filename, overall_name)
|
||
if isinstance(record, dict):
|
||
record["type"] = "overall"
|
||
target = current_filename(record)
|
||
if overall_name != target or not (directory / target).exists():
|
||
record["archive_file"] = target
|
||
_atomic_json_write(directory / target, record)
|
||
changed = True
|
||
cleanup_legacy_files(directory, record, target)
|
||
if index["latest"].get("overall") != target:
|
||
index["latest"]["overall"] = target
|
||
changed = True
|
||
else:
|
||
index["latest"]["overall"] = None
|
||
changed = True
|
||
if changed:
|
||
_write_analysis_index(plan_filename, index)
|
||
return index
|
||
|
||
|
||
def _read_analysis_cache(plan: dict[str, Any]) -> dict[str, Any]:
|
||
index = _migrate_single_analysis_store(plan["source_file"])
|
||
latest = index["latest"]
|
||
weeks: dict[str, Any] = {}
|
||
for week, relative_name in latest["weeks"].items():
|
||
public = _public_analysis_record(_read_analysis_record(plan["source_file"], relative_name))
|
||
if public is not None:
|
||
weeks[str(week)] = public
|
||
overall = _public_analysis_record(_read_analysis_record(plan["source_file"], latest.get("overall")))
|
||
return {"weeks": weeks, "overall": overall}
|
||
|
||
|
||
def _write_proposals(plan: dict[str, Any], archive: dict[str, Any]) -> None:
|
||
response = archive.get("response") if isinstance(archive.get("response"), dict) else {}
|
||
adjustments = response.get("plan_adjustments") if isinstance(response.get("plan_adjustments"), list) else []
|
||
scope_key = f"{archive.get('type', 'analysis')}:{archive.get('week') or 'overall'}"
|
||
proposals = []
|
||
for index, item in enumerate(adjustments[:3]):
|
||
if not isinstance(item, dict):
|
||
continue
|
||
proposal = deepcopy(item)
|
||
proposal.setdefault("id", f"{scope_key}-{index+1}")
|
||
proposal.setdefault("status", "open")
|
||
proposal["scope_key"] = scope_key
|
||
proposal["analysis_created_at"] = archive.get("created_at")
|
||
proposal["analysis_type"] = archive.get("type")
|
||
proposal["week"] = archive.get("week")
|
||
proposals.append(proposal)
|
||
path = PROPOSALS_DIR / Path(plan["source_file"]).name
|
||
previous = _read_json(path, {})
|
||
previous_items = previous.get("proposals") if isinstance(previous, dict) and isinstance(previous.get("proposals"), list) else []
|
||
retained = [
|
||
deepcopy(item) for item in previous_items
|
||
if isinstance(item, dict) and str(item.get("scope_key") or "") != scope_key
|
||
]
|
||
payload = {
|
||
"version": 2, "source_file": plan["source_file"], "plan_id": plan.get("plan_id"),
|
||
"published_revision": plan.get("published_revision"), "updated_at": _utc_now(),
|
||
"proposals": retained + proposals,
|
||
}
|
||
_atomic_json_write(path, payload)
|
||
|
||
|
||
def _archive_analysis_record(plan: dict[str, Any], record: dict[str, Any]) -> dict[str, Any]:
|
||
"""Speichert genau eine aktuelle Analyse je Woche sowie eine Gesamtanalyse.
|
||
|
||
Neue Auswertungen überschreiben die bisherige Datei. Es wird bewusst keine
|
||
Analysehistorie oder Versionierung geführt.
|
||
"""
|
||
archive = _sanitize(record)
|
||
if not isinstance(archive, dict):
|
||
raise ValueError("Ungültiger Analysedatensatz")
|
||
filename = current_filename(archive)
|
||
archive["version"] = 2
|
||
archive["status"] = "success"
|
||
archive["source_file"] = plan["source_file"]
|
||
archive["archive_file"] = filename
|
||
directory = _analysis_plan_dir(plan["source_file"])
|
||
path = directory / filename
|
||
_atomic_json_write(path, archive)
|
||
cleanup_legacy_files(directory, archive, filename)
|
||
|
||
with _WRITE_LOCK:
|
||
index = _read_analysis_index(plan["source_file"])
|
||
if archive.get("type") == "week":
|
||
index["latest"]["weeks"][str(int(archive.get("week") or 0))] = filename
|
||
else:
|
||
index["latest"]["overall"] = filename
|
||
_write_analysis_index(plan["source_file"], index)
|
||
_write_proposals(plan, archive)
|
||
public = _public_analysis_record(archive)
|
||
if public is None:
|
||
raise ValueError("Analyse konnte nicht gelesen werden")
|
||
return public
|
||
|
||
|
||
def _read_analysis_state(plan: dict[str, Any]) -> dict[str, Any]:
|
||
return _normalize_analysis_state(_read_json(_analysis_state_path(plan["source_file"]), {}))
|
||
|
||
|
||
def _write_analysis_state(plan: dict[str, Any], state: dict[str, Any]) -> dict[str, Any]:
|
||
clean = _normalize_analysis_state(_sanitize(state))
|
||
clean["source_file"] = plan["source_file"]
|
||
clean["updated_at"] = _utc_now()
|
||
_atomic_json_write(_analysis_state_path(plan["source_file"]), clean)
|
||
return clean
|
||
|
||
|
||
def _read_settings() -> dict[str, Any]:
|
||
settings = _read_json(SETTINGS_FILE, {})
|
||
return settings if isinstance(settings, dict) else {}
|
||
|
||
|
||
def _save_selected(plan_id: str) -> None:
|
||
settings = _read_settings()
|
||
settings["selected_plan"] = plan_id
|
||
settings["updated_at"] = _utc_now()
|
||
_atomic_json_write(SETTINGS_FILE, settings)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Plan-Normalisierung
|
||
# ---------------------------------------------------------------------------
|
||
def _unwrap_plan(raw: Any, fallback_name: str) -> tuple[str, dict[str, Any]]:
|
||
if not isinstance(raw, dict):
|
||
raise ValueError("Plan ist kein JSON-Objekt")
|
||
if isinstance(raw.get("config"), dict):
|
||
return _plain_text(raw.get("name") or fallback_name), raw["config"]
|
||
meta = raw.get("meta") if isinstance(raw.get("meta"), dict) else {}
|
||
return _plain_text(meta.get("title") or fallback_name), raw
|
||
|
||
|
||
def _is_training_config(cfg: dict[str, Any]) -> bool:
|
||
if not isinstance(cfg, dict) or cfg.get("type") == "recipe":
|
||
return False
|
||
days = cfg.get("days")
|
||
return isinstance(days, list) and any(isinstance(day, dict) for day in days)
|
||
|
||
|
||
def _plain_text(value: Any) -> str:
|
||
"""Wandelt HTML-formatierte Planfelder in sichtbaren Klartext um.
|
||
|
||
Der PDF-Editor speichert Sonderzeichen teils als HTML-Entities, zum Beispiel
|
||
``&``. In reinen UI-Labels dürfen diese nicht erneut escaped werden.
|
||
"""
|
||
text = str(value or "")
|
||
text = re.sub(r"<br\s*/?>", "\n", text, flags=re.I)
|
||
text = re.sub(r"<[^>]+>", "", text)
|
||
return html.unescape(text).strip()
|
||
|
||
|
||
def _sequence_items(value: Any) -> list[str]:
|
||
text = str(value or "")
|
||
text = re.sub(r"<br\s*/?>", " · ", text, flags=re.I)
|
||
text = _plain_text(text)
|
||
if not text:
|
||
return []
|
||
parts = re.split(r"\s*(?:·|•|\n)\s*", text)
|
||
return [part.strip() for part in parts if part.strip()]
|
||
|
||
|
||
def _slug(value: Any) -> str:
|
||
text = _plain_text(value).lower()
|
||
text = text.replace("ä", "ae").replace("ö", "oe").replace("ü", "ue").replace("ß", "ss")
|
||
text = re.sub(r"[^a-z0-9]+", "-", text).strip("-")
|
||
return text or "item"
|
||
|
||
|
||
def _normalize_phases(cfg: dict[str, Any]) -> list[dict[str, Any]]:
|
||
phases = cfg.get("phases") if isinstance(cfg.get("phases"), dict) else {}
|
||
items = phases.get("items")
|
||
normalized: list[dict[str, Any]] = []
|
||
if isinstance(items, list):
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
normalized.append({
|
||
"id": str(item.get("id") or f"phase-{len(normalized)+1}"),
|
||
"name": _plain_text(item.get("name") or f"Phase {len(normalized) + 1}"),
|
||
"params": str(item.get("params") or ""),
|
||
"weeks": max(1, _as_int(item.get("weeks"), 1)),
|
||
})
|
||
else:
|
||
names = phases.get("names") if isinstance(phases.get("names"), list) else []
|
||
params = phases.get("params") if isinstance(phases.get("params"), list) else []
|
||
for index, name in enumerate(names):
|
||
normalized.append({
|
||
"id": f"phase-{index+1}",
|
||
"name": _plain_text(name or f"Phase {index + 1}"),
|
||
"params": str(params[index] if index < len(params) else ""),
|
||
"weeks": 2,
|
||
})
|
||
if not normalized:
|
||
normalized.append({"id": "phase-1", "name": "Gesamter Plan", "params": "", "weeks": max(1, _plan_weeks(cfg))})
|
||
|
||
cursor = 1
|
||
for index, item in enumerate(normalized):
|
||
item["index"] = index
|
||
item["start_week"] = cursor
|
||
item["end_week"] = cursor + item["weeks"] - 1
|
||
cursor = item["end_week"] + 1
|
||
return normalized
|
||
|
||
|
||
def _sanitize_plan_result_schema(value: Any, *, partial: bool = True) -> dict[str, Any] | None:
|
||
"""Normalisiert ein im Trainingsplan gespeichertes Ergebnisschema.
|
||
|
||
Stufen-Schemata dürfen partiell sein und nur einzelne Eigenschaften des
|
||
Übungsstandards überschreiben. Unbekannte Felder werden ignoriert.
|
||
"""
|
||
if not isinstance(value, dict):
|
||
return None
|
||
result: dict[str, Any] = {}
|
||
mode = str(value.get("mode") or "")
|
||
if mode in {"auto", "reps", "seconds", "minutes", "none"}:
|
||
result["mode"] = mode
|
||
weight_mode = str(value.get("weight_mode") or "")
|
||
if weight_mode in {"none", "optional", "required"}:
|
||
result["weight_mode"] = weight_mode
|
||
laterality = str(value.get("laterality") or "")
|
||
if laterality in {"bilateral", "unilateral"}:
|
||
result["laterality"] = laterality
|
||
sides_mode = str(value.get("sides_mode") or "")
|
||
if sides_mode in {"same", "separate"}:
|
||
result["sides_mode"] = sides_mode
|
||
try:
|
||
sets = int(value.get("sets") or value.get("default_sets") or 0)
|
||
except (TypeError, ValueError):
|
||
sets = 0
|
||
if 1 <= sets <= 20:
|
||
result["sets"] = sets
|
||
result["locked_sets"] = bool(value.get("locked_sets", True))
|
||
elif "locked_sets" in value:
|
||
result["locked_sets"] = bool(value.get("locked_sets"))
|
||
if not partial:
|
||
result.setdefault("mode", "auto")
|
||
result.setdefault("weight_mode", "none")
|
||
result.setdefault("laterality", "bilateral")
|
||
result.setdefault("sides_mode", "same")
|
||
return result or None
|
||
|
||
|
||
def _normalize_stages(cfg: dict[str, Any]) -> dict[str, dict[str, Any]]:
|
||
stages = cfg.get("stages") if isinstance(cfg.get("stages"), dict) else {}
|
||
result: dict[str, dict[str, Any]] = {}
|
||
for key, value in stages.items():
|
||
if isinstance(value, dict):
|
||
raw_steps = value.get("steps") if isinstance(value.get("steps"), list) else []
|
||
raw_schemas = value.get("result_schemas") if isinstance(value.get("result_schemas"), list) else []
|
||
steps: list[dict[str, Any]] = []
|
||
for index, raw_step in enumerate(raw_steps):
|
||
if isinstance(raw_step, dict):
|
||
name = html.unescape(str(raw_step.get("name") or raw_step.get("label") or ""))
|
||
schema = _sanitize_plan_result_schema(raw_step.get("result_schema"), partial=True)
|
||
step_id = str(raw_step.get("id") or f"{key}-step-{index+1}")
|
||
factor = _result_number(raw_step.get("factor")) or 1.0
|
||
cluster = str(raw_step.get("movement_cluster") or "")
|
||
phase_id = str(raw_step.get("phase_id") or "")
|
||
else:
|
||
name = html.unescape(str(raw_step or ""))
|
||
schema = _sanitize_plan_result_schema(raw_schemas[index], partial=True) if index < len(raw_schemas) else None
|
||
step_id = f"{key}-step-{index+1}"
|
||
factor = 1.0
|
||
cluster = ""
|
||
phase_id = ""
|
||
steps.append({
|
||
"id": step_id, "name": name, "result_schema": schema,
|
||
"factor": factor, "movement_cluster": cluster, "phase_id": phase_id,
|
||
})
|
||
result[str(key)] = {
|
||
"id": str(value.get("id") or f"progression-{key}"),
|
||
"key": str(value.get("key") or key),
|
||
"name": _plain_text(value.get("name") or key),
|
||
"steps": steps,
|
||
}
|
||
elif isinstance(value, list):
|
||
result[str(key)] = {
|
||
"id": f"progression-{key}", "key": str(key), "name": _plain_text(key),
|
||
"steps": [{"id": f"{key}-step-{i+1}", "name": html.unescape(str(step or "")), "result_schema": None, "factor": 1.0, "movement_cluster": "", "phase_id": ""} for i, step in enumerate(value)],
|
||
}
|
||
return result
|
||
|
||
|
||
def _library_lookup(cfg: dict[str, Any]) -> dict[str, dict[str, str]]:
|
||
lookup: dict[str, dict[str, str]] = {}
|
||
groups = cfg.get("library") if isinstance(cfg.get("library"), list) else []
|
||
for group in groups:
|
||
if not isinstance(group, dict):
|
||
continue
|
||
for item in group.get("items", []) if isinstance(group.get("items"), list) else []:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
name = _plain_text(item.get("name") or "")
|
||
if not name:
|
||
continue
|
||
entry = {
|
||
"name": name,
|
||
"desc": str(item.get("desc") or ""),
|
||
"url": str(item.get("url") or ""),
|
||
"group": _plain_text(group.get("group") or ""),
|
||
}
|
||
lookup[_slug(name)] = entry
|
||
return lookup
|
||
|
||
|
||
def _closest_library_entry(name: str, lookup: dict[str, dict[str, str]]) -> dict[str, str] | None:
|
||
exact = lookup.get(_slug(name))
|
||
if exact:
|
||
return exact
|
||
target = _slug(name)
|
||
for key, entry in lookup.items():
|
||
if target in key or key in target:
|
||
return entry
|
||
return None
|
||
|
||
|
||
_RESULT_STATIC_RE = re.compile(
|
||
r"(?:\bhold\b|\bplank\b|\bhang\b|wall\s*sit|wandsitz|isometr|stützposition)",
|
||
flags=re.I,
|
||
)
|
||
_RESULT_DYNAMIC_RE = re.compile(
|
||
r"(?:push.?up|pull.?up|klimmzug|scapular|negative?|liegestütz|wiederhol|\breps?\b|rudern|\brow\b|"
|
||
r"squat|kniebeuge|dip|curl|deadlift|rdl|hinge|swing|raise|heben|climber|crunch|twist|lunge|ausfallschritt)",
|
||
flags=re.I,
|
||
)
|
||
_RESULT_NO_MEASURE_RE = re.compile(
|
||
r"(?:regeneration\s+prüfen|schlaf\s+priorisieren|schmerzcheck|sonst\s+vollständig\s+pausieren|"
|
||
r"(?:push|pull|beine|core)\s+protokollieren|schwierigkeit\s+erhöhen|reps?\s+reserve\s+prüfen|"
|
||
r"mobilität|mobility)",
|
||
flags=re.I,
|
||
)
|
||
_RESULT_MINUTES_RE = re.compile(r"(?:spaziergang|walk|walking)", flags=re.I)
|
||
|
||
_RESULT_REQUIRED_WEIGHT_RE = re.compile(
|
||
r"(?:\bkb\b|kettlebell|rucksack|goblet|floor\s*press|deadlift|\brdl\b|hip\s*hinge|"
|
||
r"\bswing\b|\bhalo\b|\bcurl|rudern.*\bkb\b|row.*\bkb\b|around.the.world)",
|
||
flags=re.I,
|
||
)
|
||
_RESULT_OPTIONAL_WEIGHT_RE = re.compile(
|
||
r"(?:gewicht|gewichtet|zusatzlast|\blast\b|\bkg\b|kettlebell|\bkb\b|rucksack|"
|
||
r"schwerer|wadenheben|calf)",
|
||
flags=re.I,
|
||
)
|
||
_RESULT_UNWEIGHTED_RE = re.compile(
|
||
r"(?:ohne\s+gewicht|ungewichtet|körpergewicht|koerpergewicht|\bbw\b)",
|
||
flags=re.I,
|
||
)
|
||
_RESULT_UNILATERAL_RE = re.compile(
|
||
r"(?:einarm|einbein|je\s+seite|pro\s+seite|links|rechts|seitenwechsel|\bwechsel\b|"
|
||
r"ausfallschritt|\blunge\b|side\s*plank|seitstütz|seitheben|lateral\s*raise|"
|
||
r"russian\s*twist|dead\s*bug|\bhalo\b|around.the.world|bottoms.?up)",
|
||
flags=re.I,
|
||
)
|
||
|
||
|
||
def _result_default_sets(
|
||
rotation_label: Any, cue: Any, training_format: dict[str, Any], exercise_count: int,
|
||
) -> tuple[int, bool]:
|
||
label = _plain_text(rotation_label or "")
|
||
text = _plain_text(f"{label} {cue or ''}")
|
||
if re.search(r"\barbeitssatz\s*\d+", label, flags=re.I):
|
||
return 1, True
|
||
|
||
set_count = None
|
||
match = re.search(r"(?<!\d)(\d{1,2})\s*(?:sätze|saetze|sets)\b", label, flags=re.I)
|
||
if match:
|
||
set_count = int(match.group(1))
|
||
|
||
rounds = training_format.get("rounds")
|
||
if (
|
||
training_format.get("fixed_interval") and isinstance(rounds, int) and rounds >= 1
|
||
and exercise_count >= 1 and rounds % exercise_count == 0
|
||
):
|
||
per_block = max(1, rounds // exercise_count)
|
||
return max(1, min(20, per_block * (set_count or 1))), True
|
||
|
||
if set_count and 1 <= set_count <= 20:
|
||
return set_count, True
|
||
for pattern in (r"(?<!\d)(\d{1,2})\s*(?:runden|intervalle)\b",):
|
||
match = re.search(pattern, text, flags=re.I)
|
||
if match:
|
||
value = int(match.group(1))
|
||
if 1 <= value <= 20:
|
||
return value, True
|
||
return 1, False
|
||
|
||
|
||
def _infer_result_schema(
|
||
exercise: dict[str, Any], stages: dict[str, dict[str, Any]],
|
||
training_format: dict[str, Any], rotation_label: Any, exercise_count: int,
|
||
) -> dict[str, Any]:
|
||
"""Leitet die feste Ergebniseingabe aus Übung, Stufen und Planformat ab."""
|
||
name = _plain_text(exercise.get("name") or "")
|
||
cue = _plain_text(exercise.get("cue") or "")
|
||
progression_id = str(exercise.get("progression_id") or "")
|
||
stage = stages.get(progression_id) if isinstance(stages, dict) else None
|
||
steps = stage.get("steps") if isinstance(stage, dict) and isinstance(stage.get("steps"), list) else []
|
||
stage_text = " ".join(_plain_text(step) for step in steps)
|
||
full_text = f"{name} {cue} {stage_text}"
|
||
|
||
static_name = bool(_RESULT_STATIC_RE.search(name))
|
||
static_stage = bool(_RESULT_STATIC_RE.search(stage_text))
|
||
dynamic_stage = bool(_RESULT_DYNAMIC_RE.search(stage_text))
|
||
explicit_time = bool(re.search(r"(?<!\d)\d{1,3}(?:[–-]\d{1,3})?\s*(?:s|sek(?:unden)?\.?)(?!\w)", cue, flags=re.I))
|
||
has_stage = bool(steps)
|
||
if _RESULT_NO_MEASURE_RE.search(name):
|
||
mode = "none"
|
||
elif _RESULT_MINUTES_RE.search(name):
|
||
mode = "minutes"
|
||
elif explicit_time and not _RESULT_DYNAMIC_RE.search(name):
|
||
mode = "seconds"
|
||
elif static_name and not dynamic_stage:
|
||
mode = "seconds"
|
||
elif static_stage and dynamic_stage:
|
||
mode = "auto"
|
||
elif static_stage:
|
||
mode = "seconds"
|
||
elif not has_stage and not training_format.get("fixed_interval") and not _RESULT_DYNAMIC_RE.search(full_text):
|
||
mode = "none"
|
||
else:
|
||
mode = "reps"
|
||
|
||
if mode in {"none", "minutes"}:
|
||
weight_mode = "none"
|
||
elif _RESULT_REQUIRED_WEIGHT_RE.search(name):
|
||
# Manche Pläne nennen das spätere Gerät bereits im Übungsnamen, beginnen
|
||
# aber ausdrücklich mit Körpergewicht. Dann ist Gewicht abhängig von der
|
||
# gewählten Progressionsstufe und darf nicht pauschal Pflicht sein.
|
||
weight_mode = "optional" if _RESULT_UNWEIGHTED_RE.search(stage_text) else "required"
|
||
elif _RESULT_OPTIONAL_WEIGHT_RE.search(stage_text) or re.search(r"(?:wadenheben|calf)", name, flags=re.I):
|
||
weight_mode = "optional"
|
||
else:
|
||
weight_mode = "none"
|
||
|
||
laterality = "unilateral" if _RESULT_UNILATERAL_RE.search(full_text) else "bilateral"
|
||
default_sets, locked_sets = _result_default_sets(rotation_label, cue, training_format, exercise_count)
|
||
sides_mode = "same"
|
||
if laterality == "unilateral" and re.search(
|
||
r"(?:dg\s*1.*links.*dg\s*2.*rechts|seitenwechsel|\bwechsel\b)",
|
||
f"{name} {cue}", flags=re.I,
|
||
):
|
||
sides_mode = "separate"
|
||
|
||
# Optionaler Plan-Override. Der Tracker bleibt abwärtskompatibel, kann aber
|
||
# präzise Schemata übernehmen, sobald das Trainingsplan-Modul sie anbietet.
|
||
explicit_sets = 0
|
||
explicit = exercise.get("result_schema")
|
||
if isinstance(explicit, dict):
|
||
explicit_mode = str(explicit.get("mode") or "")
|
||
if explicit_mode in {"auto", "reps", "seconds", "minutes", "none"}:
|
||
mode = explicit_mode
|
||
explicit_weight = str(explicit.get("weight_mode") or "")
|
||
if explicit_weight in {"none", "optional", "required"}:
|
||
weight_mode = explicit_weight
|
||
explicit_laterality = str(explicit.get("laterality") or "")
|
||
if explicit_laterality in {"bilateral", "unilateral"}:
|
||
laterality = explicit_laterality
|
||
explicit_sides = str(explicit.get("sides_mode") or "")
|
||
if explicit_sides in {"same", "separate"}:
|
||
sides_mode = explicit_sides
|
||
try:
|
||
explicit_sets = int(explicit.get("sets"))
|
||
except (TypeError, ValueError):
|
||
explicit_sets = 0
|
||
if 1 <= explicit_sets <= 20:
|
||
default_sets = explicit_sets
|
||
locked_sets = bool(explicit.get("locked_sets", True))
|
||
|
||
# Einseitige Tabata-Übungen mit genau zwei Durchgängen pro Block werden
|
||
# als ein Links-/Rechts-Paar je Block dargestellt. Entscheidend ist die
|
||
# endgültige Lateralisierung nach dem Plan-Override, nicht ein Stichwort in
|
||
# einer späteren Progressionsstufe.
|
||
rounds = training_format.get("rounds")
|
||
if (
|
||
not (1 <= explicit_sets <= 20)
|
||
and laterality == "unilateral"
|
||
and training_format.get("fixed_interval")
|
||
and isinstance(rounds, int) and exercise_count >= 1
|
||
and rounds % exercise_count == 0 and rounds // exercise_count == 2
|
||
and default_sets % 2 == 0
|
||
):
|
||
default_sets = max(1, default_sets // 2)
|
||
|
||
return {
|
||
"version": RESULT_DATA_VERSION,
|
||
"mode": mode,
|
||
"weight_mode": weight_mode,
|
||
"laterality": laterality,
|
||
"sides_mode": sides_mode if laterality == "unilateral" else "same",
|
||
"default_sets": default_sets,
|
||
"locked_sets": locked_sets,
|
||
"work_seconds": training_format.get("work_seconds") if training_format.get("fixed_interval") else None,
|
||
"fixed_interval": bool(training_format.get("fixed_interval")),
|
||
}
|
||
|
||
|
||
def _normalize_rotations(day: dict[str, Any], day_num: int, library: dict[str, dict[str, str]]) -> list[dict[str, Any]]:
|
||
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"]})
|
||
result: list[dict[str, Any]] = []
|
||
day_id = str(day.get("id") or f"day-{day_num}")
|
||
for rotation_index, rotation in enumerate(rotations):
|
||
if not isinstance(rotation, dict): continue
|
||
rotation_id = str(rotation.get("id") or f"{day_id}-rotation-{rotation_index+1}")
|
||
exercises = rotation.get("exercises") if isinstance(rotation.get("exercises"), list) else []
|
||
normalized_exercises = []
|
||
for exercise_index, exercise in enumerate(exercises):
|
||
if not isinstance(exercise, dict): continue
|
||
name = _plain_text(exercise.get("name") or f"Übung {exercise_index + 1}")
|
||
key = str(exercise.get("progression_id") or exercise.get("key") or "").strip()
|
||
progression_id = key or f"name:{_slug(name)}"
|
||
normalized_exercises.append({
|
||
"id": str(exercise.get("id") or f"{rotation_id}-exercise-{exercise_index+1}"),
|
||
"legacy_id": str(exercise.get("legacy_id") or f"d{day_num}-r{rotation_index}-e{exercise_index}"),
|
||
"exercise_id": str(exercise.get("exercise_id") or progression_id),
|
||
"name": name, "key": key, "progression_id": progression_id,
|
||
"cue": str(exercise.get("cue") or ""),
|
||
"movement_cluster": str(exercise.get("movement_cluster") or ""),
|
||
"result_schema": _sanitize_plan_result_schema(exercise.get("result_schema"), partial=False),
|
||
"library": _closest_library_entry(name, library),
|
||
})
|
||
result.append({"id": rotation_id, "label": str(rotation.get("label") or f"Block {rotation_index + 1}"), "exercises": normalized_exercises})
|
||
return result
|
||
|
||
|
||
def _as_int(value: Any, default: int) -> int:
|
||
try:
|
||
return int(value)
|
||
except (TypeError, ValueError):
|
||
return default
|
||
|
||
|
||
def _plan_weeks(cfg: dict[str, Any]) -> int:
|
||
meta = cfg.get("meta") if isinstance(cfg.get("meta"), dict) else {}
|
||
return max(1, _as_int(meta.get("weeks"), 8))
|
||
|
||
|
||
def _first_int(patterns: list[str], text: str) -> int | None:
|
||
for pattern in patterns:
|
||
match = re.search(pattern, text, flags=re.I)
|
||
if match:
|
||
try:
|
||
return int(match.group(1))
|
||
except (TypeError, ValueError):
|
||
continue
|
||
return None
|
||
|
||
|
||
def _detect_training_format(cfg: dict[str, Any], days: list[dict[str, Any]]) -> dict[str, Any]:
|
||
"""Leitet bindende Session-Regeln aus dem Plan ab.
|
||
|
||
Die KI bekommt damit nicht nur Übungen, sondern auch das tatsächliche
|
||
Ausführungsformat. Bei Tabata bleiben Arbeits- und Pausenintervalle fest.
|
||
"""
|
||
explicit_format = cfg.get("training_format") if isinstance(cfg.get("training_format"), dict) else {}
|
||
explicit_mode = str(explicit_format.get("mode") or "auto")
|
||
if explicit_mode in {"sets_reps", "tabata", "fixed_interval"}:
|
||
def _positive_int(field: str) -> int | None:
|
||
try:
|
||
number = int(explicit_format.get(field))
|
||
except (TypeError, ValueError):
|
||
return None
|
||
return number if number > 0 else None
|
||
|
||
fixed = explicit_mode in {"tabata", "fixed_interval"} or bool(explicit_format.get("fixed_interval"))
|
||
work = _positive_int("work_seconds")
|
||
rest = _positive_int("rest_seconds")
|
||
rounds = _positive_int("rounds")
|
||
constraints: list[str] = []
|
||
if fixed:
|
||
interval = f"{work} Sekunden" if work else "die vorgegebene Dauer"
|
||
constraints.extend([
|
||
f"Arbeitsintervalle bleiben fest bei {interval}; keine längeren Holds oder verlängerten Arbeitsintervalle empfehlen.",
|
||
"Progression nur innerhalb des Planformats: planmäßige schwierigere Variante, mehr Gewicht, saubererer ROM/Tempo oder bessere Technik; Pausen und Rundenzahl nicht eigenmächtig ändern.",
|
||
])
|
||
else:
|
||
constraints.append(
|
||
"Vorgegebene Wiederholungs-, Haltezeit-, Satz- und Pausenbereiche bleiben bindend; nach Erreichen der Obergrenze die planmäßige Schwierigkeit anpassen statt unbegrenzt Volumen zu addieren."
|
||
)
|
||
front = cfg.get("front") if isinstance(cfg.get("front"), dict) else {}
|
||
return {
|
||
"mode": explicit_mode,
|
||
"is_tabata": explicit_mode == "tabata",
|
||
"fixed_interval": fixed,
|
||
"work_seconds": work,
|
||
"rest_seconds": rest,
|
||
"rounds": rounds,
|
||
"rounds_scope": "block",
|
||
"source": "plan",
|
||
"session_method": _plain_text(front.get("session_how_body") or ""),
|
||
"timer_note": _plain_text(front.get("timer_note") or ""),
|
||
"constraints": constraints,
|
||
}
|
||
|
||
front = cfg.get("front") if isinstance(cfg.get("front"), dict) else {}
|
||
phases = cfg.get("phases") if isinstance(cfg.get("phases"), dict) else {}
|
||
phase_items = phases.get("items") if isinstance(phases.get("items"), list) else []
|
||
fragments: list[str] = [
|
||
str(front.get("session_how_body") or ""),
|
||
str(front.get("timer_note") or ""),
|
||
str(front.get("goals_note") or ""),
|
||
*[str(value or "") for value in front.get("reminders", []) if isinstance(front.get("reminders"), list)],
|
||
*[str(item.get("params") or "") for item in phase_items if isinstance(item, dict)],
|
||
]
|
||
for day in days:
|
||
fragments.append(str(day.get("note") or ""))
|
||
for rotation in day.get("rotations", []) if isinstance(day.get("rotations"), list) else []:
|
||
fragments.append(str(rotation.get("label") or ""))
|
||
for exercise in rotation.get("exercises", []) if isinstance(rotation.get("exercises"), list) else []:
|
||
fragments.append(str(exercise.get("cue") or ""))
|
||
raw_text = " ".join(fragments)
|
||
plain = _plain_text(raw_text)
|
||
low = plain.casefold()
|
||
negated_tabata = bool(re.search(r"\b(?:kein|keine|ohne)\s+tabata\b|\bkein\s+20\s*/\s*10", low))
|
||
is_tabata = not negated_tabata and ("tabata" in low or bool(re.search(r"\b20\s*/\s*10\b", low)))
|
||
work_seconds = _first_int([
|
||
r"(\d+)\s*(?:s|sek(?:unden)?)\s*(?:arbeit|work)",
|
||
r"(?:arbeit|work)\s*(?:von|:)??\s*(\d+)\s*(?:s|sek(?:unden)?)",
|
||
r"(\d+)\s*/\s*\d+",
|
||
], low)
|
||
rest_seconds = _first_int([
|
||
r"(\d+)\s*(?:s|sek(?:unden)?)\s*(?:pause|rest)",
|
||
r"(?:pause|rest)\s*(?:von|:)??\s*(\d+)\s*(?:s|sek(?:unden)?)",
|
||
r"\d+\s*/\s*(\d+)",
|
||
], low)
|
||
rounds = _first_int([
|
||
r"(\d+)\s*[×x]\s*\(",
|
||
r"(\d+)\s*(?:runden|intervalle)",
|
||
r"(?:runden|intervalle)\s*(?:von|:)??\s*(\d+)",
|
||
], low)
|
||
fixed_interval = bool(is_tabata or ((work_seconds and rest_seconds) and not negated_tabata))
|
||
mode = "tabata" if is_tabata else ("fixed_interval" if fixed_interval else "sets_reps")
|
||
constraints: list[str] = []
|
||
if fixed_interval:
|
||
interval = f"{work_seconds} Sekunden" if work_seconds else "die vorgegebene Dauer"
|
||
constraints.append(
|
||
f"Arbeitsintervalle bleiben fest bei {interval}; keine längeren Holds oder verlängerten Arbeitsintervalle empfehlen."
|
||
)
|
||
constraints.append(
|
||
"Progression nur innerhalb des Planformats: planmäßige schwierigere Variante, mehr Gewicht, saubererer ROM/Tempo oder bessere Technik; Pausen und Rundenzahl nicht eigenmächtig ändern."
|
||
)
|
||
else:
|
||
constraints.append(
|
||
"Vorgegebene Wiederholungs-, Haltezeit-, Satz- und Pausenbereiche bleiben bindend; nach Erreichen der Obergrenze die planmäßige Schwierigkeit anpassen statt unbegrenzt Volumen zu addieren."
|
||
)
|
||
return {
|
||
"mode": mode,
|
||
"is_tabata": is_tabata,
|
||
"fixed_interval": fixed_interval,
|
||
"work_seconds": work_seconds,
|
||
"rest_seconds": rest_seconds,
|
||
"rounds": rounds,
|
||
"rounds_scope": "block",
|
||
"source": "text_fallback",
|
||
"session_method": _plain_text(front.get("session_how_body") or ""),
|
||
"timer_note": _plain_text(front.get("timer_note") or ""),
|
||
"constraints": constraints,
|
||
}
|
||
|
||
|
||
def _published_config_hash(config: dict[str, Any]) -> str:
|
||
raw = json.dumps(config, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
||
return hashlib.sha256(raw).hexdigest()
|
||
|
||
|
||
def _normalize_plan(path: Path) -> dict[str, Any]:
|
||
raw = _read_json(path)
|
||
plan_name, cfg = _unwrap_plan(raw, path.stem)
|
||
wrapper = raw if isinstance(raw, dict) and isinstance(raw.get("config"), dict) else {}
|
||
if not _is_training_config(cfg):
|
||
raise ValueError("Keine Trainingsplan-Datei")
|
||
|
||
raw_schema_version = int(cfg.get("schema_version") or 1)
|
||
raw_contract_version = int(cfg.get("contract_version") or 1)
|
||
if raw_schema_version > PLAN_SCHEMA_VERSION or raw_contract_version > CONTRACT_VERSION:
|
||
raise ValueError(
|
||
f"Der Trainingsplan verwendet Schema {raw_schema_version}/Vertrag {raw_contract_version}; "
|
||
f"dieser Tracker unterstützt höchstens {PLAN_SCHEMA_VERSION}/{CONTRACT_VERSION}."
|
||
)
|
||
compatibility_warnings = []
|
||
if raw_schema_version < PLAN_SCHEMA_VERSION:
|
||
compatibility_warnings.append(
|
||
f"Legacy-Plan-Schema {raw_schema_version}; Fallback-Erkennung bleibt aktiv. Im Trainingsplan-Editor neu veröffentlichen."
|
||
)
|
||
if raw_contract_version < CONTRACT_VERSION:
|
||
compatibility_warnings.append(
|
||
f"Legacy-Trainingsvertrag {raw_contract_version}; einige stabile IDs oder Ergebnisschemata können fehlen."
|
||
)
|
||
|
||
meta = cfg.get("meta") if isinstance(cfg.get("meta"), dict) else {}
|
||
front = cfg.get("front") if isinstance(cfg.get("front"), dict) else {}
|
||
prepost = cfg.get("prepost") if isinstance(cfg.get("prepost"), dict) else {}
|
||
library = _library_lookup(cfg)
|
||
phases = _normalize_phases(cfg)
|
||
stages = _normalize_stages(cfg)
|
||
|
||
days: list[dict[str, Any]] = []
|
||
for fallback_index, day in enumerate(cfg.get("days", []), start=1):
|
||
if not isinstance(day, dict):
|
||
continue
|
||
day_num = max(1, _as_int(day.get("num"), fallback_index))
|
||
pp = prepost.get(str(day_num)) if isinstance(prepost.get(str(day_num)), dict) else {}
|
||
days.append({
|
||
"id": str(day.get("id") or f"day-{day_num}"),
|
||
"num": day_num,
|
||
"focus": _plain_text(day.get("focus") or f"Tag {day_num}"),
|
||
"light": bool(day.get("light")),
|
||
"color": str(day.get("color") or ""),
|
||
"badge": deepcopy(day.get("badge") if isinstance(day.get("badge"), dict) else {}),
|
||
"note": str(day.get("note") or ""),
|
||
"note_bg": str(day.get("note_bg") or ""),
|
||
"rotations": _normalize_rotations(day, day_num, library),
|
||
"warmup": {
|
||
"text": str(pp.get("warmup") or ""),
|
||
"items": _sequence_items(pp.get("warmup")),
|
||
},
|
||
"cooldown": {
|
||
"text": str(pp.get("cooldown") or ""),
|
||
"items": _sequence_items(pp.get("cooldown")),
|
||
},
|
||
"stretch": {
|
||
"text": str(pp.get("stretch") or ""),
|
||
"items": _sequence_items(pp.get("stretch")),
|
||
},
|
||
})
|
||
days.sort(key=lambda item: item["num"])
|
||
|
||
weeks = _plan_weeks(cfg)
|
||
phase_by_week: list[int] = []
|
||
for week in range(1, weeks + 1):
|
||
phase_index = len(phases) - 1
|
||
for phase in phases:
|
||
if phase["start_week"] <= week <= phase["end_week"]:
|
||
phase_index = phase["index"]
|
||
break
|
||
phase_by_week.append(phase_index)
|
||
|
||
training_format = _detect_training_format(cfg, days)
|
||
for day in days:
|
||
for rotation in day.get("rotations", []):
|
||
for exercise in rotation.get("exercises", []):
|
||
exercise["result_schema"] = _infer_result_schema(
|
||
exercise, stages, training_format, rotation.get("label") or "",
|
||
len(rotation.get("exercises", [])),
|
||
)
|
||
|
||
return {
|
||
"source_file": path.name,
|
||
# Entwurfsänderungen im Planeditor dürfen laufende Sessions und Analysen
|
||
# nicht veralten lassen. Relevant ist ausschließlich die veröffentlichte Config.
|
||
"source_hash": _published_config_hash(cfg),
|
||
"plan_id": str(cfg.get("plan_id") or wrapper.get("plan_id") or path.stem),
|
||
"schema_version": raw_schema_version,
|
||
"contract_version": raw_contract_version,
|
||
"compatibility": {
|
||
"supported": True,
|
||
"legacy": bool(compatibility_warnings),
|
||
"warnings": compatibility_warnings,
|
||
"supported_plan_schema": PLAN_SCHEMA_VERSION,
|
||
"supported_contract": CONTRACT_VERSION,
|
||
},
|
||
"published_revision": int(wrapper.get("published_revision") or 1),
|
||
"published_at": str(wrapper.get("published_at") or ""),
|
||
"name": plan_name,
|
||
"title": _plain_text(meta.get("title") or plan_name),
|
||
"subtitle": _plain_text(meta.get("subtitle") or ""),
|
||
"weeks": weeks,
|
||
"days": days,
|
||
"phases": phases,
|
||
"phase_by_week": phase_by_week,
|
||
"stages": stages,
|
||
"exercise_catalog": deepcopy(cfg.get("exercise_catalog") if isinstance(cfg.get("exercise_catalog"), dict) else {}),
|
||
"front": {
|
||
"goals_head": _plain_text(front.get("goals_head") or ""),
|
||
"goals_box": str(front.get("goals_box") or ""),
|
||
"goals_note": str(front.get("goals_note") or ""),
|
||
"reminder_head": _plain_text(front.get("reminder_head") or ""),
|
||
"reminders": [str(value or "") for value in front.get("reminders", [])]
|
||
if isinstance(front.get("reminders"), list) else [],
|
||
"session_how_head": _plain_text(front.get("session_how_head") or ""),
|
||
"session_how_body": str(front.get("session_how_body") or ""),
|
||
"timer_note": str(front.get("timer_note") or ""),
|
||
},
|
||
"training_format": training_format,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Tracker-Daten
|
||
# ---------------------------------------------------------------------------
|
||
def _default_tracker(plan: dict[str, Any]) -> dict[str, Any]:
|
||
return {
|
||
"version": TRACKER_SCHEMA_VERSION,
|
||
"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": "",
|
||
"plan_notes": "",
|
||
},
|
||
"progressions": {}, # Legacy-Fallback aus Version 1; nicht mehr aktiv bearbeitet.
|
||
"sessions": {},
|
||
"week_statuses": {},
|
||
"created_at": _utc_now(),
|
||
"updated_at": _utc_now(),
|
||
}
|
||
|
||
|
||
def _normalize_analysis_state(value: Any) -> dict[str, Any]:
|
||
state = value if isinstance(value, dict) else {}
|
||
status = str(state.get("status") or "idle")
|
||
if status not in {"idle", "running", "done", "error"}: status = "idle"
|
||
clean = dict(state); clean["status"] = status
|
||
if status == "running":
|
||
lease = str(clean.get("lease_expires_at") or "")
|
||
try:
|
||
expired = datetime.now(timezone.utc) > datetime.fromisoformat(lease)
|
||
except (TypeError, ValueError):
|
||
started = str(clean.get("started_at") or "")
|
||
try: expired = (datetime.now(timezone.utc) - datetime.fromisoformat(started)).total_seconds() > _ANALYSIS_TIMEOUT_SECONDS
|
||
except (TypeError, ValueError): expired = False
|
||
if expired:
|
||
clean.update({
|
||
"status": "error",
|
||
"error": "Die vorherige Analyse hat ihre Job-Lease verloren oder wurde unterbrochen.",
|
||
"message": "Analysejob unterbrochen.",
|
||
"finished_at": _utc_now(),
|
||
})
|
||
return clean
|
||
|
||
|
||
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", {})
|
||
tracker.setdefault("progressions", {})
|
||
tracker.setdefault("sessions", {})
|
||
tracker.setdefault("week_statuses", {})
|
||
tracker.setdefault("created_at", _utc_now())
|
||
|
||
# Analysefelder aus älteren Plugin-Versionen werden weder gelesen noch
|
||
# wieder gespeichert. Die eigentliche Bereinigung bestehender Dateien kann
|
||
# separat erfolgen; jeder neue Autosave schreibt bereits das saubere Modell.
|
||
tracker.pop("analysis_cache", None)
|
||
tracker.pop("analysis_state", None)
|
||
tracker.pop("analyses", None)
|
||
tracker["version"] = TRACKER_SCHEMA_VERSION
|
||
tracker["source_file"] = plan["source_file"]
|
||
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
|
||
|
||
|
||
def _sanitize(value: Any, depth: int = 0) -> Any:
|
||
"""Begrenzt Nutzdaten ohne die flexible Tracker-Struktur zu zerstören."""
|
||
if depth > 12:
|
||
return None
|
||
if value is None or isinstance(value, (bool, int, float)):
|
||
return value
|
||
if isinstance(value, str):
|
||
return value[:20000]
|
||
if isinstance(value, list):
|
||
return [_sanitize(item, depth + 1) for item in value[:2000]]
|
||
if isinstance(value, dict):
|
||
result = {}
|
||
for index, (key, item) in enumerate(value.items()):
|
||
if index >= 5000:
|
||
break
|
||
clean_key = str(key)[:200]
|
||
result[clean_key] = _sanitize(item, depth + 1)
|
||
return result
|
||
return str(value)[:20000]
|
||
|
||
|
||
def _result_number(value: Any) -> float | None:
|
||
if value in (None, "") or isinstance(value, bool):
|
||
return None
|
||
try:
|
||
number = float(str(value).replace(",", "."))
|
||
except (TypeError, ValueError):
|
||
return None
|
||
if number < 0 or number > 100000:
|
||
return None
|
||
return round(number, 3)
|
||
|
||
|
||
def _result_values(value: Any, limit: int = 20) -> list[float | None]:
|
||
if not isinstance(value, list):
|
||
return []
|
||
return [_result_number(item) for item in value[:limit]]
|
||
|
||
|
||
def _sanitize_result_data(value: Any) -> dict[str, Any] | None:
|
||
if not isinstance(value, dict):
|
||
return None
|
||
mode = str(value.get("mode") or "")
|
||
if mode not in {"reps", "seconds", "minutes"}:
|
||
return None
|
||
laterality = str(value.get("laterality") or "bilateral")
|
||
if laterality not in {"bilateral", "unilateral"}:
|
||
laterality = "bilateral"
|
||
sides_mode = str(value.get("sides_mode") or "same")
|
||
if sides_mode not in {"same", "separate"}:
|
||
sides_mode = "same"
|
||
try:
|
||
sets = max(1, min(20, int(value.get("sets") or 1)))
|
||
except (TypeError, ValueError):
|
||
sets = 1
|
||
clean: dict[str, Any] = {
|
||
"version": RESULT_DATA_VERSION,
|
||
"mode": mode,
|
||
"laterality": laterality,
|
||
"sides_mode": sides_mode if laterality == "unilateral" else "same",
|
||
"sets": sets,
|
||
"weight_kg": _result_number(value.get("weight_kg")),
|
||
"values": _result_values(value.get("values")),
|
||
"left_values": _result_values(value.get("left_values")),
|
||
"right_values": _result_values(value.get("right_values")),
|
||
}
|
||
for key in ("values", "left_values", "right_values"):
|
||
values = clean[key][:sets]
|
||
values.extend([None] * (sets - len(values)))
|
||
clean[key] = values
|
||
has_values = any(number is not None for number in clean["values"] + clean["left_values"] + clean["right_values"])
|
||
if not has_values and clean["weight_kg"] is None:
|
||
return None
|
||
return clean
|
||
|
||
|
||
def _format_result_number(value: Any) -> str:
|
||
number = _result_number(value)
|
||
if number is None:
|
||
return ""
|
||
return str(int(number)) if float(number).is_integer() else f"{number:.3f}".rstrip("0").rstrip(".")
|
||
|
||
|
||
def _format_result_data(value: Any) -> str:
|
||
data = _sanitize_result_data(value)
|
||
if not data:
|
||
return ""
|
||
unit = " s" if data["mode"] == "seconds" else (" min" if data["mode"] == "minutes" else " Reps")
|
||
weight = _format_result_number(data.get("weight_kg"))
|
||
prefix = f"{weight} kg · " if weight else ""
|
||
|
||
def joined(values: list[Any]) -> str:
|
||
return "/".join(_format_result_number(item) or "–" for item in values)
|
||
|
||
if data["laterality"] == "unilateral" and data["sides_mode"] == "separate":
|
||
return f"{prefix}L {joined(data['left_values'])} · R {joined(data['right_values'])}{unit}".strip()
|
||
side = " je Seite" if data["laterality"] == "unilateral" else ""
|
||
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")
|
||
clean = _sanitize(incoming)
|
||
if not isinstance(clean, dict):
|
||
raise ValueError("Ungültige Tracker-Daten")
|
||
clean["version"] = TRACKER_SCHEMA_VERSION
|
||
clean["source_file"] = plan["source_file"]
|
||
clean["source_hash"] = plan["source_hash"]
|
||
clean["plan_id"] = plan.get("plan_id")
|
||
clean["plan_revision"] = plan.get("published_revision")
|
||
clean["created_at"] = existing.get("created_at") or _utc_now()
|
||
clean["updated_at"] = _utc_now()
|
||
clean.pop("source_changed", None)
|
||
clean.pop("analysis_cache", None); clean.pop("analysis_state", None); clean.pop("analyses", None)
|
||
if not isinstance(clean.get("profile"), dict): clean["profile"] = {}
|
||
if not isinstance(clean.get("progressions"), dict): clean["progressions"] = {}
|
||
if not isinstance(clean.get("sessions"), dict): clean["sessions"] = {}
|
||
if not isinstance(clean.get("week_statuses"), dict): clean["week_statuses"] = {}
|
||
valid_statuses = {"planned", "in_progress", "stopped", "completed"}
|
||
for session in clean["sessions"].values():
|
||
if not isinstance(session, dict): continue
|
||
session.setdefault("plan_id", existing.get("plan_id") or plan.get("plan_id"))
|
||
session.setdefault("plan_revision", existing.get("plan_revision") or plan.get("published_revision"))
|
||
status = str(session.get("status") or "planned")
|
||
session["status"] = status if status in valid_statuses else "planned"
|
||
items = session.get("items") if isinstance(session.get("items"), dict) else {}
|
||
session["items"] = items
|
||
for item in items.values():
|
||
if not isinstance(item, dict): continue
|
||
item_status = str(item.get("completion_status") or ("completed" if item.get("done") else "planned"))
|
||
item["completion_status"] = item_status if item_status in VALID_ITEM_STATUSES else "planned"
|
||
item["done"] = item["completion_status"] in {"completed", "partial"}
|
||
if item["completion_status"] != "skipped": item.pop("skip_reason", None)
|
||
if isinstance(item.get("result_data"), dict):
|
||
data = _sanitize_result_data(item.get("result_data"))
|
||
if data is not None:
|
||
item["result_data"] = data; item["result"] = _format_result_data(data)
|
||
else: item.pop("result_data", None)
|
||
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
|
||
|
||
|
||
def _save_tracker(plan: dict[str, Any], incoming: Any) -> dict[str, Any]:
|
||
existing = _load_tracker(plan)
|
||
expected = incoming.get("expected_revision") if isinstance(incoming, dict) else None
|
||
if expected is not None and int(expected) != int(existing.get("revision") or 1):
|
||
raise RuntimeError(f"revision_conflict:{int(existing.get('revision') or 1)}")
|
||
payload = incoming.get("tracker") if isinstance(incoming, dict) and isinstance(incoming.get("tracker"), dict) else incoming
|
||
clean = _normalize_tracker_payload(plan, payload, existing)
|
||
clean["revision"] = int(existing.get("revision") or 1) + 1
|
||
_atomic_json_write(_tracker_path(plan["source_file"]), clean)
|
||
return clean
|
||
|
||
|
||
def _patch_tracker(plan: dict[str, Any], payload: Any) -> dict[str, Any]:
|
||
if not isinstance(payload, dict): raise ValueError("Ungültiger Session-Patch")
|
||
existing = _load_tracker(plan)
|
||
session_payload = payload.get("session")
|
||
if isinstance(session_payload, dict):
|
||
session_payload = deepcopy(session_payload)
|
||
session_payload.setdefault("plan_id", plan.get("plan_id"))
|
||
session_payload.setdefault("plan_revision", plan.get("published_revision"))
|
||
payload = dict(payload)
|
||
payload["session"] = session_payload
|
||
try:
|
||
merged = merge_session_patch(existing, payload, _utc_now())
|
||
except RuntimeError:
|
||
raise
|
||
clean = _normalize_tracker_payload(plan, merged, existing)
|
||
clean["revision"] = int(merged.get("revision") or int(existing.get("revision") or 1) + 1)
|
||
_atomic_json_write(_tracker_path(plan["source_file"]), clean)
|
||
return clean
|
||
|
||
|
||
def _openai_ready() -> bool:
|
||
return bool(os.environ.get("OPENAI_API_KEY", "").strip() and os.environ.get("OPENAI_MODEL", "").strip())
|
||
|
||
|
||
def _stable_hash(value: Any) -> str:
|
||
raw = json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
||
return hashlib.sha256(raw).hexdigest()
|
||
|
||
|
||
def _session_key_parts(key: str) -> tuple[int, int]:
|
||
match = re.fullmatch(r"w(\d+)-d(\d+)", str(key or ""))
|
||
if not match:
|
||
return (9999, 9999)
|
||
return int(match.group(1)), int(match.group(2))
|
||
|
||
|
||
def _session_date(profile: dict[str, Any], week: int, day_num: int) -> str:
|
||
raw = str(profile.get("start_date") or "").strip()
|
||
if not raw:
|
||
return ""
|
||
try:
|
||
start = date.fromisoformat(raw)
|
||
except ValueError:
|
||
return ""
|
||
return (start + timedelta(days=(week - 1) * 7 + day_num - 1)).isoformat()
|
||
|
||
|
||
def _phase_for_week(plan: dict[str, Any], week: int) -> dict[str, Any]:
|
||
phase_map = plan.get("phase_by_week") if isinstance(plan.get("phase_by_week"), list) else []
|
||
phases = plan.get("phases") if isinstance(plan.get("phases"), list) else []
|
||
index = phase_map[max(0, week - 1)] if week - 1 < len(phase_map) else 0
|
||
if isinstance(index, int) and 0 <= index < len(phases):
|
||
return phases[index]
|
||
return phases[0] if phases else {"name": "", "params": ""}
|
||
|
||
|
||
def _has_text(value: Any) -> bool:
|
||
return bool(str(value or "").strip())
|
||
|
||
|
||
def _session_records(plan: dict[str, Any], tracker: dict[str, Any]) -> list[dict[str, Any]]:
|
||
"""Normalisiert ausschließlich tatsächlich befüllte Sessiondaten."""
|
||
days_by_num = {int(day.get("num", 0)): day for day in plan.get("days", []) if isinstance(day, dict)}
|
||
profile = tracker.get("profile") if isinstance(tracker.get("profile"), dict) else {}
|
||
sessions_out: list[dict[str, Any]] = []
|
||
sessions = tracker.get("sessions") if isinstance(tracker.get("sessions"), dict) else {}
|
||
|
||
for key, session in sorted(sessions.items(), key=lambda pair: _session_key_parts(pair[0])):
|
||
if not isinstance(session, dict):
|
||
continue
|
||
week, day_num = _session_key_parts(key)
|
||
if week == 9999:
|
||
continue
|
||
day = days_by_num.get(day_num, {})
|
||
exercise_by_id: dict[str, dict[str, Any]] = {}
|
||
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 isinstance(exercise, dict):
|
||
enriched = dict(exercise)
|
||
enriched["rotation"] = _plain_text(rotation.get("label") or "")
|
||
enriched["kind"] = "exercise"
|
||
exercise_by_id[str(exercise.get("id") or "")] = enriched
|
||
legacy_id = str(exercise.get("legacy_id") or "")
|
||
if legacy_id:
|
||
exercise_by_id[legacy_id] = enriched
|
||
for routine_name in ("warmup", "cooldown", "stretch"):
|
||
routine = day.get(routine_name) if isinstance(day.get(routine_name), dict) else {}
|
||
for index, label in enumerate(routine.get("items", []) if isinstance(routine.get("items"), list) else []):
|
||
exercise_by_id[f"{routine_name}-{index}"] = {
|
||
"name": _plain_text(label), "progression_id": "", "cue": "",
|
||
"rotation": routine_name, "kind": routine_name,
|
||
}
|
||
|
||
items_out: list[dict[str, Any]] = []
|
||
items = session.get("items") if isinstance(session.get("items"), dict) else {}
|
||
for item_id, item in items.items():
|
||
if not isinstance(item, dict):
|
||
continue
|
||
item_status = str(item.get("completion_status") or ("completed" if item.get("done") else "planned"))
|
||
meaningful = item_status != "planned" or isinstance(item.get("result_data"), dict) or any(_has_text(item.get(field)) for field in (
|
||
"result", "note", "progression", "exercise_name", "progression_id", "skip_reason"
|
||
))
|
||
if not meaningful:
|
||
continue
|
||
exercise = exercise_by_id.get(str(item_id), {})
|
||
items_out.append({
|
||
"item_id": str(item_id),
|
||
"exercise": _plain_text(item.get("exercise_name") or exercise.get("name") or item_id),
|
||
"progression_id": str(item.get("progression_id") or exercise.get("progression_id") or ""),
|
||
"progression": _plain_text(item.get("progression") or ""),
|
||
"result": _plain_text(item.get("result") or ""),
|
||
"result_data": deepcopy(item.get("result_data")) if isinstance(item.get("result_data"), dict) else None,
|
||
"note": _plain_text(item.get("note") or ""),
|
||
"done": bool(item.get("done")),
|
||
"completion_status": item_status,
|
||
"skip_reason": _plain_text(item.get("skip_reason") or ""),
|
||
"progression_step_id": str(item.get("progression_step_id") or ""),
|
||
"exercise_id": str(exercise.get("exercise_id") or ""),
|
||
"movement_cluster": str(exercise.get("movement_cluster") or ""),
|
||
"cue": _plain_text(exercise.get("cue") or ""),
|
||
"rotation": _plain_text(exercise.get("rotation") or ""),
|
||
"kind": str(exercise.get("kind") or "other"),
|
||
})
|
||
|
||
section_notes = {
|
||
name: _plain_text(session.get(f"{name}_note") or "")
|
||
for name in ("warmup", "cooldown", "stretch")
|
||
if _has_text(session.get(f"{name}_note"))
|
||
}
|
||
meaningful_session = (
|
||
session.get("status") in {"in_progress", "stopped", "completed"}
|
||
or _has_text(session.get("note")) or bool(section_notes) or bool(items_out)
|
||
)
|
||
if not meaningful_session:
|
||
continue
|
||
phase = _phase_for_week(plan, week)
|
||
sessions_out.append({
|
||
"session_key": key,
|
||
"week": week,
|
||
"day": day_num,
|
||
"date": _session_date(profile, week, day_num),
|
||
"focus": _plain_text(day.get("focus") or f"Tag {day_num}"),
|
||
"light": bool(day.get("light")),
|
||
"phase": _plain_text(phase.get("name") or ""),
|
||
"phase_instructions": _plain_text(phase.get("params") or ""),
|
||
"status": str(session.get("status") or "planned"),
|
||
"plan_id": str(session.get("plan_id") or tracker.get("plan_id") or plan.get("plan_id") or ""),
|
||
"plan_revision": int(session.get("plan_revision") or tracker.get("plan_revision") or plan.get("published_revision") or 1),
|
||
"started_at": str(session.get("started_at") or ""),
|
||
"stopped_at": str(session.get("stopped_at") or ""),
|
||
"completed_at": str(session.get("completed_at") or ""),
|
||
"session_note": _plain_text(session.get("note") or ""),
|
||
"section_notes": section_notes,
|
||
"items": items_out,
|
||
})
|
||
return sessions_out
|
||
|
||
|
||
def _plan_context(
|
||
plan: dict[str, Any], *, week: int | None = None, progression_ids: set[str] | None = None
|
||
) -> dict[str, Any]:
|
||
stages = plan.get("stages") if isinstance(plan.get("stages"), dict) else {}
|
||
if progression_ids:
|
||
stages = {key: value for key, value in stages.items() if key in progression_ids}
|
||
phase = _phase_for_week(plan, week) if week else None
|
||
return {
|
||
"plan_id": plan.get("plan_id"),
|
||
"published_revision": plan.get("published_revision"),
|
||
"schema_version": plan.get("schema_version"),
|
||
"contract_version": plan.get("contract_version"),
|
||
"title": plan.get("title") or plan.get("name"),
|
||
"subtitle": plan.get("subtitle") or "",
|
||
"weeks": plan.get("weeks"),
|
||
"selected_week": week,
|
||
"selected_phase": {
|
||
"name": _plain_text(phase.get("name") or ""),
|
||
"instructions": _plain_text(phase.get("params") or ""),
|
||
} if phase else None,
|
||
"training_format": deepcopy(plan.get("training_format") or {}),
|
||
"guidance": {
|
||
"goals": _plain_text(plan.get("front", {}).get("goals_box") or ""),
|
||
"goals_note": _plain_text(plan.get("front", {}).get("goals_note") or ""),
|
||
"session_method": _plain_text(plan.get("front", {}).get("session_how_body") or ""),
|
||
"timer_note": _plain_text(plan.get("front", {}).get("timer_note") or ""),
|
||
"reminders": [_plain_text(item) for item in plan.get("front", {}).get("reminders", [])],
|
||
},
|
||
"days": [
|
||
{
|
||
"day": day.get("num"),
|
||
"focus": _plain_text(day.get("focus") or ""),
|
||
"light": bool(day.get("light")),
|
||
"note": _plain_text(day.get("note") or ""),
|
||
"warmup": [_plain_text(item) for item in day.get("warmup", {}).get("items", [])],
|
||
"cooldown": [_plain_text(item) for item in day.get("cooldown", {}).get("items", [])],
|
||
"stretch": [_plain_text(item) for item in day.get("stretch", {}).get("items", [])],
|
||
"blocks": [
|
||
{
|
||
"label": _plain_text(rotation.get("label") or ""),
|
||
"exercises": [
|
||
{
|
||
"id": exercise.get("id") or "",
|
||
"exercise_id": exercise.get("exercise_id") or "",
|
||
"name": _plain_text(exercise.get("name") or ""),
|
||
"progression_id": exercise.get("progression_id") or "",
|
||
"movement_cluster": exercise.get("movement_cluster") or "",
|
||
"cue": _plain_text(exercise.get("cue") or ""),
|
||
}
|
||
for exercise in rotation.get("exercises", []) if isinstance(exercise, dict)
|
||
],
|
||
}
|
||
for rotation in day.get("rotations", []) if isinstance(rotation, dict)
|
||
],
|
||
}
|
||
for day in plan.get("days", []) if isinstance(day, dict)
|
||
],
|
||
"phases": [
|
||
{
|
||
"name": _plain_text(item.get("name") or ""),
|
||
"instructions": _plain_text(item.get("params") or ""),
|
||
"start_week": item.get("start_week"),
|
||
"end_week": item.get("end_week"),
|
||
}
|
||
for item in plan.get("phases", []) if isinstance(item, dict)
|
||
],
|
||
"progression_stages": stages,
|
||
"exercise_catalog": deepcopy(plan.get("exercise_catalog") or {}),
|
||
}
|
||
|
||
|
||
def _latest_state_before(records: list[dict[str, Any]], week: int) -> list[dict[str, Any]]:
|
||
latest: dict[str, dict[str, Any]] = {}
|
||
for session in records:
|
||
if int(session.get("week") or 0) >= week:
|
||
continue
|
||
for item in session.get("items", []):
|
||
progression_id = str(item.get("progression_id") or "")
|
||
if not progression_id:
|
||
continue
|
||
latest[progression_id] = {
|
||
"progression_id": progression_id,
|
||
"exercise": item.get("exercise") or progression_id,
|
||
"progression": item.get("progression") or "",
|
||
"result": item.get("result") or "",
|
||
"note": item.get("note") or "",
|
||
"session_key": session.get("session_key") or "",
|
||
}
|
||
return list(latest.values())
|
||
|
||
|
||
def _plan_bodyweight_kg(plan: dict[str, Any]) -> float | None:
|
||
"""Liest ein optional im Plantext genanntes Körpergewicht, ohne es zu raten."""
|
||
candidates = [
|
||
plan.get("subtitle"), plan.get("title"),
|
||
(plan.get("meta") or {}).get("subtitle") if isinstance(plan.get("meta"), dict) else "",
|
||
]
|
||
for candidate in candidates:
|
||
for value in re.findall(r"(?<!\d)(\d{2,3}(?:[.,]\d+)?)\s*kg\b", _plain_text(candidate or ""), flags=re.I):
|
||
number = float(value.replace(",", "."))
|
||
if 35 <= number <= 300:
|
||
return number
|
||
return None
|
||
|
||
|
||
def _public_equivalence_guide(plan: dict[str, Any]) -> dict[str, Any]:
|
||
tables = []
|
||
for table in _EQUIVALENCE_TABLES:
|
||
mode = table["mode"]
|
||
unit = "Referenz-Reps" if mode == "reps" else "Referenzsekunden"
|
||
variants = []
|
||
for variant in table.get("variants", []):
|
||
factor = float(variant.get("factor") or 1)
|
||
if mode == "reps":
|
||
if factor <= 1:
|
||
needed = round(1 / factor, 1) if factor else 0
|
||
example = f"{needed:g} {'Rep' if needed == 1 else 'Reps'} ≈ 1 Referenz-Rep"
|
||
else:
|
||
example = f"1 Rep = {factor:g} Referenz-Reps"
|
||
else:
|
||
example = f"20 s = {round(20 * factor, 1):g} {unit}"
|
||
variants.append({
|
||
"label": variant.get("label") or "Variante",
|
||
"factor": factor,
|
||
"example": example,
|
||
})
|
||
tables.append({
|
||
"id": table["id"], "label": table["label"], "mode": mode,
|
||
"reference": table["reference"], "variants": variants,
|
||
})
|
||
plan_catalog = []
|
||
catalog = plan.get("exercise_catalog") if isinstance(plan.get("exercise_catalog"), dict) else {}
|
||
seen_catalog: set[tuple[str, str]] = set()
|
||
for entry in catalog.values():
|
||
if not isinstance(entry, dict):
|
||
continue
|
||
progression_id = str(entry.get("progression_id") or "")
|
||
cluster_id = str(entry.get("movement_cluster") or "general")
|
||
key = (progression_id, cluster_id)
|
||
if key in seen_catalog:
|
||
continue
|
||
seen_catalog.add(key)
|
||
variants = []
|
||
for variant in entry.get("variants", []) if isinstance(entry.get("variants"), list) else []:
|
||
if not isinstance(variant, dict):
|
||
continue
|
||
variants.append({
|
||
"id": str(variant.get("id") or ""),
|
||
"name": _plain_text(variant.get("name") or ""),
|
||
"factor": float(variant.get("factor") or 1.0),
|
||
"movement_cluster": str(variant.get("movement_cluster") or cluster_id),
|
||
"result_schema": deepcopy(variant.get("result_schema") if isinstance(variant.get("result_schema"), dict) else {}),
|
||
})
|
||
plan_catalog.append({
|
||
"exercise_id": str(entry.get("id") or ""),
|
||
"name": _plain_text(entry.get("name") or progression_id),
|
||
"progression_id": progression_id,
|
||
"movement_cluster": cluster_id,
|
||
"movement_label": _plain_text(entry.get("movement_label") or cluster_id),
|
||
"variants": variants,
|
||
})
|
||
plan_catalog.sort(key=lambda row: (row["movement_label"], row["name"]))
|
||
bodyweight = _plan_bodyweight_kg(plan)
|
||
return {
|
||
"version": PROGRESSION_NORMALIZATION_VERSION,
|
||
"title": "Varianten-Cluster und Referenzwerte",
|
||
"bodyweight_kg": bodyweight,
|
||
"method": (
|
||
"Wiederholungen werden innerhalb desselben Bewegungsmusters mit einem Variantenfaktor in Referenz-Reps umgerechnet. "
|
||
"Isometrische Übungen bleiben getrennt und werden als Referenzsekunden ausgewertet. Externe Last wird bei reinen "
|
||
"Gewichtsübungen als kg·Reps bzw. kg·s gerechnet. Bei Bodyweight-Übungen wird Zusatzlast nur dann relativ berücksichtigt, "
|
||
"wenn das Körpergewicht ausdrücklich im Plan steht."
|
||
),
|
||
"caveat": (
|
||
"Die Faktoren sind transparente Analyseheuristiken, keine exakten physiologischen Gleichungen und keine Vorgabe für die "
|
||
"Trainingssteuerung. Technik, ROM, Tempo und Nähe zum Versagen bleiben zusätzlich entscheidend."
|
||
),
|
||
"tables": tables,
|
||
"plan_catalog": plan_catalog,
|
||
"sources": deepcopy(_EQUIVALENCE_SOURCES),
|
||
}
|
||
|
||
|
||
def _extract_training_measure(text: Any) -> dict[str, Any] | None:
|
||
"""Extrahiert Messwerte bevorzugt aus ``result_data``, sonst aus Freitext.
|
||
|
||
Beispiele: ``8/7/6``, ``3x8``, ``6x20 s``, ``4 kg, 6/6/5 je Seite``.
|
||
Die Rohangabe bleibt zusätzlich erhalten; unklare Felder werden nicht erfunden.
|
||
"""
|
||
if isinstance(text, dict):
|
||
structured = text.get("result_data") if isinstance(text.get("result_data"), dict) else text
|
||
data = _sanitize_result_data(structured)
|
||
if data:
|
||
if data["laterality"] == "unilateral" and data["sides_mode"] == "separate":
|
||
left = [value for value in data["left_values"] if value is not None]
|
||
right = [value for value in data["right_values"] if value is not None]
|
||
values = left + right
|
||
total = sum(values)
|
||
per_side = False
|
||
else:
|
||
values = [value for value in data["values"] if value is not None]
|
||
per_side = data["laterality"] == "unilateral"
|
||
total = sum(values) * (2 if per_side else 1)
|
||
if values:
|
||
measure_mode = "seconds" if data["mode"] == "minutes" else data["mode"]
|
||
if data["mode"] == "minutes":
|
||
values = [round(value * 60, 3) for value in values]
|
||
total = total * 60
|
||
return {
|
||
"mode": measure_mode, "total": round(total, 3), "values": values,
|
||
"sets": data["sets"], "weight_kg": data.get("weight_kg"),
|
||
"per_side": per_side, "raw": _format_result_data(data), "structured": True,
|
||
}
|
||
text = text.get("result") or ""
|
||
|
||
raw = _plain_text(text or "").strip()
|
||
if not raw:
|
||
return None
|
||
low = raw.casefold().replace("×", "x")
|
||
kg_values = [float(v.replace(",", ".")) for v in re.findall(r"(\d+(?:[.,]\d+)?)\s*kg\b", low)]
|
||
weight_kg = max(kg_values) if kg_values else None
|
||
per_side = bool(re.search(r"\b(?:je|pro)\s+seite\b|/\s*seite\b", low))
|
||
side_multiplier = 2 if per_side else 1
|
||
|
||
# Dauer ist eindeutig, sobald eine Sekunden-/Minuten-Einheit vorkommt.
|
||
has_seconds = bool(re.search(r"\b(?:s|sek\.?|sekunden|min\.?|minuten)\b", low))
|
||
if has_seconds:
|
||
work = re.sub(r"\d+(?:[.,]\d+)?\s*kg\b", " ", low)
|
||
set_duration = re.search(
|
||
r"(\d+(?:[.,]\d+)?)\s*x\s*(\d+(?:[.,]\d+)?)\s*(s|sek\.?|sekunden|min\.?|minuten)\b",
|
||
work,
|
||
)
|
||
if set_duration:
|
||
sets = float(set_duration.group(1).replace(",", "."))
|
||
duration = float(set_duration.group(2).replace(",", "."))
|
||
if set_duration.group(3).startswith("min"):
|
||
duration *= 60
|
||
total = sets * duration * side_multiplier
|
||
values = [duration] * max(1, int(round(sets)))
|
||
return {
|
||
"mode": "seconds", "total": round(total, 3), "values": values,
|
||
"sets": sets, "weight_kg": weight_kg, "per_side": per_side, "raw": raw,
|
||
}
|
||
durations: list[float] = []
|
||
for number, unit in re.findall(r"(\d+(?:[.,]\d+)?)\s*(s|sek\.?|sekunden|min\.?|minuten)\b", work):
|
||
value = float(number.replace(",", "."))
|
||
if unit.startswith("min"):
|
||
value *= 60
|
||
durations.append(value)
|
||
if durations:
|
||
return {
|
||
"mode": "seconds", "total": round(sum(durations) * side_multiplier, 3),
|
||
"values": durations, "sets": len(durations), "weight_kg": weight_kg,
|
||
"per_side": per_side, "raw": raw,
|
||
}
|
||
# Formate wie „20/20/18 s“: Einheit einmal am Ende.
|
||
numbers = [float(v.replace(",", ".")) for v in re.findall(r"\d+(?:[.,]\d+)?", work)]
|
||
if numbers:
|
||
return {
|
||
"mode": "seconds", "total": round(sum(numbers) * side_multiplier, 3),
|
||
"values": numbers, "sets": len(numbers), "weight_kg": weight_kg,
|
||
"per_side": per_side, "raw": raw,
|
||
}
|
||
return None
|
||
|
||
# Reps: Gewichtsangaben werden vor dem Auslesen entfernt, damit 10 kg nicht
|
||
# versehentlich als zehn Wiederholungen zählen.
|
||
work = re.sub(r"\d+(?:[.,]\d+)?\s*kg\b", " ", low)
|
||
set_reps = re.search(r"(\d+(?:[.,]\d+)?)\s*x\s*(\d+(?:[.,]\d+)?)", work)
|
||
if set_reps:
|
||
sets = float(set_reps.group(1).replace(",", "."))
|
||
reps = float(set_reps.group(2).replace(",", "."))
|
||
total = sets * reps * side_multiplier
|
||
return {
|
||
"mode": "reps", "total": round(total, 3), "values": [reps] * max(1, int(round(sets))),
|
||
"sets": sets, "weight_kg": weight_kg, "per_side": per_side, "raw": raw,
|
||
}
|
||
numbers = [float(v.replace(",", ".")) for v in re.findall(r"\d+(?:[.,]\d+)?", work)]
|
||
if not numbers:
|
||
return None
|
||
return {
|
||
"mode": "reps", "total": round(sum(numbers) * side_multiplier, 3),
|
||
"values": numbers, "sets": len(numbers), "weight_kg": weight_kg,
|
||
"per_side": per_side, "raw": raw,
|
||
}
|
||
|
||
|
||
def _numeric_measure(text: Any) -> dict[str, Any] | None:
|
||
"""Kompatible Kurzform für bestehende Aufrufer und Tests."""
|
||
measure = _extract_training_measure(text)
|
||
if not measure:
|
||
return None
|
||
unit = "Sekunden" if measure["mode"] == "seconds" else "Reps"
|
||
return {"value": measure["total"], "unit": unit, "values": measure["values"]}
|
||
|
||
|
||
def _catalog_entry(plan: dict[str, Any], item: dict[str, Any]) -> dict[str, Any] | None:
|
||
catalog = plan.get("exercise_catalog") if isinstance(plan.get("exercise_catalog"), dict) else {}
|
||
exercise_id = str(item.get("exercise_id") or "")
|
||
if exercise_id and isinstance(catalog.get(exercise_id), dict):
|
||
return catalog[exercise_id]
|
||
progression_id = str(item.get("progression_id") or "")
|
||
for entry in catalog.values():
|
||
if isinstance(entry, dict) and progression_id and str(entry.get("progression_id") or "") == progression_id:
|
||
return entry
|
||
return None
|
||
|
||
|
||
def _explicit_cluster_spec(plan: dict[str, Any], item: dict[str, Any], measure: dict[str, Any]) -> dict[str, Any] | None:
|
||
entry = _catalog_entry(plan, item) or {}
|
||
cluster_id = str(item.get("movement_cluster") or entry.get("movement_cluster") or "")
|
||
name = _plain_text(item.get("exercise") or item.get("exercise_name") or entry.get("name") or "")
|
||
text = f"{name} {_plain_text(item.get('progression') or '')}".casefold()
|
||
seconds = measure.get("mode") == "seconds"
|
||
if not cluster_id or cluster_id == "general":
|
||
return None
|
||
|
||
if cluster_id == "horizontal_push":
|
||
return {"id": "push_support_hold" if seconds else "push_up", "label": "Push-up Startposition / High Plank" if seconds else "Push-up", "mode": measure["mode"], "priority": 100}
|
||
if cluster_id == "horizontal_pull":
|
||
external = bool(measure.get("weight_kg")) or any(token in text for token in ("kb", "kettlebell", "rucksack", "hantel"))
|
||
return {"id": "loaded_row" if external else "inverted_row", "label": "Rudern" if external else "Inverted Row", "mode": measure["mode"], "priority": 92 if not external else 84, "external_load": external}
|
||
if cluster_id == "vertical_pull":
|
||
return {"id": "active_hang" if seconds else "pull_up", "label": "Active Hang" if seconds else "Pull-up", "mode": measure["mode"], "priority": 100}
|
||
if cluster_id == "vertical_push":
|
||
if "dip" in text:
|
||
return {"id": "dip_support_hold" if seconds else "dip", "label": "Dip Support Hold" if seconds else "Dip", "mode": measure["mode"], "priority": 82}
|
||
if "pike" in text:
|
||
return {"id": "pike_hold" if seconds else "pike_press", "label": "Pike Hold" if seconds else "Pike Push-up", "mode": measure["mode"], "priority": 78}
|
||
return {"id": "loaded_press", "label": "Loaded Press", "mode": measure["mode"], "priority": 72, "external_load": True}
|
||
if cluster_id == "knee_dominant":
|
||
if seconds:
|
||
if "wall" in text or "wandsitz" in text:
|
||
return {"id": "wall_sit", "label": "Wall Sit", "mode": "seconds", "priority": 75}
|
||
return {"id": "deep_squat_hold", "label": "Deep Squat Hold", "mode": "seconds", "priority": 90}
|
||
if "lunge" in text or "ausfallschritt" in text:
|
||
return {"id": "lunge", "label": "Ausfallschritt", "mode": "reps", "priority": 78}
|
||
return {"id": "squat", "label": "Squat", "mode": "reps", "priority": 100}
|
||
if cluster_id == "hip_hinge":
|
||
if "bridge" in text:
|
||
return {"id": "bridge_hold" if seconds else "bridge", "label": "Glute Bridge Hold" if seconds else "Glute Bridge", "mode": measure["mode"], "priority": 74}
|
||
if "swing" in text:
|
||
return {"id": "swing", "label": "Kettlebell Swing", "mode": measure["mode"], "priority": 72, "external_load": True}
|
||
return {"id": "hinge", "label": "Hinge / Deadlift", "mode": measure["mode"], "priority": 88, "external_load": True}
|
||
if cluster_id == "core":
|
||
if seconds:
|
||
if "side" in text or "seit" in text:
|
||
return {"id": "side_plank", "label": "Side Plank", "mode": "seconds", "priority": 82}
|
||
if "hollow" in text:
|
||
return {"id": "hollow_hold", "label": "Hollow Hold", "mode": "seconds", "priority": 82}
|
||
return {"id": "plank", "label": name or "Core Hold", "mode": "seconds", "priority": 100}
|
||
if "dead bug" in text:
|
||
return {"id": "dead_bug", "label": "Dead Bug", "mode": "reps", "priority": 66}
|
||
if "beinheben" in text or "leg raise" in text:
|
||
return {"id": "leg_raise", "label": "Beinheben", "mode": "reps", "priority": 70}
|
||
if cluster_id == "elbow_flexion":
|
||
return {"id": "curl", "label": "Bizeps-Curl", "mode": measure["mode"], "priority": 52, "external_load": True}
|
||
if cluster_id == "calf":
|
||
return {"id": "calf_raise", "label": "Wadenheben", "mode": measure["mode"], "priority": 58}
|
||
if cluster_id == "loaded_mobility":
|
||
return {"id": "grip_hold" if seconds else f"exercise:{_slug(name)}", "label": name or "Kettlebell-Kontrolle", "mode": measure["mode"], "priority": 55, "external_load": True}
|
||
return None
|
||
|
||
|
||
def _cluster_spec(plan: dict[str, Any], item: dict[str, Any], measure: dict[str, Any]) -> dict[str, Any]:
|
||
explicit = _explicit_cluster_spec(plan, item, measure)
|
||
if explicit:
|
||
return explicit
|
||
name = _plain_text(item.get("exercise") or item.get("exercise_name") or "")
|
||
progression = _plain_text(item.get("progression") or "")
|
||
pid = str(item.get("progression_id") or "")
|
||
text = f"{name} {progression} {pid}".casefold()
|
||
seconds = measure.get("mode") == "seconds"
|
||
|
||
def found(*patterns: str) -> bool:
|
||
return any(re.search(pattern, text, flags=re.I) for pattern in patterns)
|
||
|
||
if seconds:
|
||
if found(r"side plank", r"seit.*plank", r"\bsp\b"):
|
||
return {"id": "side_plank", "label": "Side Plank", "mode": "seconds", "priority": 82}
|
||
if found(r"hollow", r"\bholl\b"):
|
||
return {"id": "hollow_hold", "label": "Hollow Hold", "mode": "seconds", "priority": 82}
|
||
if found(r"wall sit", r"wandsitz", r"\bwall\b"):
|
||
return {"id": "wall_sit", "label": "Wall Sit", "mode": "seconds", "priority": 75}
|
||
if found(r"deep squat", r"squat hold", r"tiefe.*kniebeuge.*halt"):
|
||
return {"id": "deep_squat_hold", "label": "Deep Squat Hold", "mode": "seconds", "priority": 90}
|
||
if found(r"active hang", r"hang", r"hängen"):
|
||
return {"id": "active_hang", "label": "Active Hang", "mode": "seconds", "priority": 92}
|
||
if found(r"push.?up", r"liegestütz") and found(r"plank", r"hold", r"startposition"):
|
||
return {"id": "push_support_hold", "label": "Push-up Startposition / High Plank", "mode": "seconds", "priority": 90}
|
||
if found(r"pike"):
|
||
return {"id": "pike_hold", "label": "Pike Hold", "mode": "seconds", "priority": 76}
|
||
if found(r"dip") and found(r"support", r"hold", r"stütz"):
|
||
return {"id": "dip_support_hold", "label": "Dip Support Hold", "mode": "seconds", "priority": 78}
|
||
if found(r"plank"):
|
||
return {"id": "plank", "label": "Plank", "mode": "seconds", "priority": 100}
|
||
if found(r"superman", r"y-t-w", r"ytw"):
|
||
return {"id": "back_hold", "label": "Superman / Y-T-W Hold", "mode": "seconds", "priority": 68}
|
||
if found(r"glute bridge", r"bridge"):
|
||
return {"id": "bridge_hold", "label": "Glute Bridge Hold", "mode": "seconds", "priority": 72}
|
||
if found(r"bottoms.?up", r"carry", r"griff"):
|
||
return {"id": "grip_hold", "label": "Griff- / Carry-Hold", "mode": "seconds", "priority": 62}
|
||
fallback = pid or _slug(name) or "isometric"
|
||
return {"id": f"iso:{fallback}", "label": name or "Isometrischer Hold", "mode": "seconds", "priority": 45}
|
||
|
||
if found(r"push.?up", r"liegestütz") and not found(r"pike"):
|
||
return {"id": "push_up", "label": "Push-up", "mode": "reps", "priority": 100}
|
||
if found(r"dip"):
|
||
return {"id": "dip", "label": "Dip", "mode": "reps", "priority": 82}
|
||
if found(r"pike"):
|
||
return {"id": "pike_press", "label": "Pike Push-up", "mode": "reps", "priority": 78}
|
||
if found(r"pull.?up", r"klimmzug"):
|
||
return {"id": "pull_up", "label": "Pull-up", "mode": "reps", "priority": 100}
|
||
if found(r"inverted row", r"body row", r"bent leg row", r"straight leg row"):
|
||
return {"id": "inverted_row", "label": "Inverted Row", "mode": "reps", "priority": 92}
|
||
if found(r"squat", r"kniebeuge") and not found(r"wall sit"):
|
||
return {"id": "squat", "label": "Squat", "mode": "reps", "priority": 100}
|
||
if found(r"ausfallschritt", r"lunge"):
|
||
return {"id": "lunge", "label": "Ausfallschritt", "mode": "reps", "priority": 78}
|
||
if found(r"rdl", r"deadlift", r"hip hinge", r"kreuzheben"):
|
||
return {"id": "hinge", "label": "Hinge / Deadlift", "mode": "reps", "priority": 88, "external_load": True}
|
||
if found(r"glute bridge", r"bridge"):
|
||
return {"id": "bridge", "label": "Glute Bridge", "mode": "reps", "priority": 74}
|
||
if found(r"wadenheben", r"calf"):
|
||
return {"id": "calf_raise", "label": "Wadenheben", "mode": "reps", "priority": 58}
|
||
if found(r"rudern.*eng", r"row.*eng"):
|
||
return {"id": "row_lat", "label": "Einarm-Rudern eng", "mode": "reps", "priority": 82, "external_load": True}
|
||
if found(r"rudern.*breit", r"row.*breit"):
|
||
return {"id": "row_upper", "label": "Einarm-Rudern breit", "mode": "reps", "priority": 78, "external_load": True}
|
||
if found(r"rudern", r"\brow\b"):
|
||
return {"id": "loaded_row", "label": "Rudern", "mode": "reps", "priority": 84, "external_load": True}
|
||
if found(r"curl", r"bizeps"):
|
||
return {"id": "curl", "label": "Bizeps-Curl", "mode": "reps", "priority": 52, "external_load": True}
|
||
if found(r"seitheben", r"lateral raise"):
|
||
return {"id": "lateral_raise", "label": "Seitheben", "mode": "reps", "priority": 50, "external_load": True}
|
||
if found(r"floor press", r"overhead press", r"schulterdrücken"):
|
||
return {"id": "loaded_press", "label": "Loaded Press", "mode": "reps", "priority": 72, "external_load": True}
|
||
if found(r"swing"):
|
||
return {"id": "swing", "label": "Kettlebell Swing", "mode": "reps", "priority": 72, "external_load": True}
|
||
if found(r"mountain climber"):
|
||
return {"id": "mountain_climber", "label": "Mountain Climbers", "mode": "reps", "priority": 52}
|
||
if found(r"dead bug"):
|
||
return {"id": "dead_bug", "label": "Dead Bug", "mode": "reps", "priority": 66}
|
||
if found(r"beinheben", r"leg raise"):
|
||
return {"id": "leg_raise", "label": "Beinheben", "mode": "reps", "priority": 70}
|
||
if found(r"bicycle"):
|
||
return {"id": "bicycle", "label": "Bicycle Crunch", "mode": "reps", "priority": 48}
|
||
if found(r"russian twist"):
|
||
return {"id": "russian_twist", "label": "Russian Twist", "mode": "reps", "priority": 48}
|
||
fallback = pid or _slug(name) or "exercise"
|
||
return {"id": f"exercise:{fallback}", "label": name or fallback, "mode": "reps", "priority": 40}
|
||
|
||
def _equivalence_table(cluster_id: str) -> dict[str, Any] | None:
|
||
return next((table for table in _EQUIVALENCE_TABLES if table.get("id") == cluster_id), None)
|
||
|
||
|
||
def _stage_factor(plan: dict[str, Any], item: dict[str, Any], mode: str) -> float:
|
||
pid = str(item.get("progression_id") or "")
|
||
progression = _slug(item.get("progression") or "")
|
||
step_id = str(item.get("progression_step_id") or "")
|
||
stage = plan.get("stages", {}).get(pid) if isinstance(plan.get("stages"), dict) else None
|
||
steps = stage.get("steps") if isinstance(stage, dict) and isinstance(stage.get("steps"), list) else []
|
||
if not steps:
|
||
return 1.0
|
||
index = None
|
||
selected: dict[str, Any] | None = None
|
||
for i, raw_step in enumerate(steps):
|
||
step = raw_step if isinstance(raw_step, dict) else {"name": str(raw_step or "")}
|
||
normalized = _slug(step.get("name") or "")
|
||
if (step_id and str(step.get("id") or "") == step_id) or (progression and (progression == normalized or progression in normalized or normalized in progression)):
|
||
index = i
|
||
selected = step
|
||
break
|
||
if selected:
|
||
try:
|
||
explicit = float(selected.get("factor"))
|
||
except (TypeError, ValueError):
|
||
explicit = 0.0
|
||
if explicit > 0:
|
||
return round(explicit, 3)
|
||
if index is None or len(steps) <= 1:
|
||
return 1.0
|
||
if mode == "seconds":
|
||
factors = [0.65 + (0.50 * i / (len(steps) - 1)) for i in range(len(steps))]
|
||
else:
|
||
factors = [0.25 * (4 ** (i / (len(steps) - 1))) for i in range(len(steps))]
|
||
return round(factors[index], 3)
|
||
|
||
def _variation_factor(plan: dict[str, Any], item: dict[str, Any], cluster: dict[str, Any]) -> tuple[float, str]:
|
||
entry = _catalog_entry(plan, item)
|
||
step_id = str(item.get("progression_step_id") or "")
|
||
progression = _slug(item.get("progression") or "")
|
||
if entry:
|
||
variants = entry.get("variants") if isinstance(entry.get("variants"), list) else []
|
||
for variant in variants:
|
||
if not isinstance(variant, dict):
|
||
continue
|
||
variant_name = _slug(variant.get("name") or "")
|
||
if (step_id and str(variant.get("id") or "") == step_id) or (progression and variant_name and (progression == variant_name or progression in variant_name or variant_name in progression)):
|
||
try:
|
||
factor = float(variant.get("factor") or 1.0)
|
||
except (TypeError, ValueError):
|
||
factor = 1.0
|
||
return max(0.001, factor), _plain_text(variant.get("name") or item.get("progression") or "Variante")
|
||
if cluster.get("external_load"):
|
||
return 1.0, _plain_text(item.get("progression") or "Last aus Ergebnisfeld")
|
||
table = _equivalence_table(cluster["id"])
|
||
text = f"{_plain_text(item.get('progression') or '')} {_plain_text(item.get('exercise') or item.get('exercise_name') or '')}".casefold()
|
||
matches: list[tuple[float, str]] = []
|
||
if table:
|
||
for variant in table.get("variants", []):
|
||
if any(re.search(pattern, text, flags=re.I) for pattern in variant.get("patterns", [])):
|
||
matches.append((float(variant.get("factor") or 1), str(variant.get("label") or "Variante")))
|
||
if matches:
|
||
return max(matches, key=lambda entry: entry[0])
|
||
factor = _stage_factor(plan, item, cluster["mode"])
|
||
return factor, _plain_text(item.get("progression") or "nicht näher bezeichnet")
|
||
|
||
def _normalized_point(plan: dict[str, Any], item: dict[str, Any], measure: dict[str, Any], cluster: dict[str, Any]) -> dict[str, Any]:
|
||
factor, variant = _variation_factor(plan, item, cluster)
|
||
total = float(measure.get("total") or 0)
|
||
weight_kg = measure.get("weight_kg")
|
||
bodyweight_kg = _plan_bodyweight_kg(plan)
|
||
load_modifier = 1.0
|
||
unit = "Referenzsekunden" if measure["mode"] == "seconds" else "Referenz-Reps"
|
||
|
||
if cluster.get("external_load") and weight_kg:
|
||
value = total * float(weight_kg) * factor
|
||
unit = "kg·s" if measure["mode"] == "seconds" else "kg·Reps"
|
||
else:
|
||
# Zusatzlast bei Bodyweight-Bewegungen wird nur mit explizitem
|
||
# Körpergewicht verwendet. Ohne diese Information bleibt sie sichtbar,
|
||
# fließt aber nicht über eine erfundene Annahme ein.
|
||
if weight_kg and bodyweight_kg and cluster["id"] in {
|
||
"push_up", "dip", "pike_press", "pull_up", "inverted_row", "squat",
|
||
"lunge", "plank", "side_plank", "deep_squat_hold",
|
||
}:
|
||
load_modifier = 1 + float(weight_kg) / bodyweight_kg
|
||
value = total * factor * load_modifier
|
||
return {
|
||
"value": round(value, 2), "unit": unit, "factor": round(factor, 3),
|
||
"variant": variant, "raw_total": round(total, 2), "raw": measure.get("raw") or "",
|
||
"weight_kg": weight_kg, "load_modifier": round(load_modifier, 3),
|
||
}
|
||
|
||
|
||
def _cluster_progression_series(plan: dict[str, Any], records: list[dict[str, Any]], limit: int = 10) -> list[dict[str, Any]]:
|
||
sessions = sorted(records, key=lambda row: (int(row.get("week") or 0), int(row.get("day") or 0), str(row.get("session_key") or "")))
|
||
groups: dict[tuple[str, str], dict[str, Any]] = {}
|
||
for session in sessions:
|
||
session_points: dict[tuple[str, str], dict[str, Any]] = {}
|
||
for item in session.get("items", []):
|
||
measure = _extract_training_measure(item)
|
||
if not measure:
|
||
continue
|
||
cluster = _cluster_spec(plan, item, measure)
|
||
point = _normalized_point(plan, item, measure, cluster)
|
||
key = (cluster["id"], point["unit"])
|
||
bucket = session_points.setdefault(key, {
|
||
"cluster": cluster, "value": 0.0, "raw": [], "factors": [], "variants": [], "weights": [],
|
||
})
|
||
bucket["value"] += point["value"]
|
||
bucket["raw"].append(point["raw"])
|
||
bucket["factors"].append(point["factor"])
|
||
bucket["variants"].append(point["variant"])
|
||
if point.get("weight_kg") is not None:
|
||
bucket["weights"].append(point["weight_kg"])
|
||
for key, bucket in session_points.items():
|
||
cluster = bucket["cluster"]
|
||
group = groups.setdefault(key, {
|
||
"id": cluster["id"], "label": cluster["label"], "mode": cluster["mode"],
|
||
"unit": key[1], "priority": int(cluster.get("priority") or 40),
|
||
"reference": (_equivalence_table(cluster["id"]) or {}).get("reference", "planinterne Stufenheuristik"),
|
||
"points": [],
|
||
})
|
||
label = f"W{session.get('week')}/T{session.get('day')}"
|
||
group["points"].append({
|
||
"label": label, "week": session.get("week"), "day": session.get("day"),
|
||
"session_key": session.get("session_key"), "value": round(bucket["value"], 2),
|
||
"raw": " + ".join(bucket["raw"]),
|
||
"variant": " / ".join(dict.fromkeys(v for v in bucket["variants"] if v)),
|
||
"factor": round(max(bucket["factors"] or [1]), 3),
|
||
"weight_kg": max(bucket["weights"]) if bucket["weights"] else None,
|
||
})
|
||
series = []
|
||
for group in groups.values():
|
||
points = group["points"]
|
||
if not points:
|
||
continue
|
||
first = float(points[0]["value"] or 0)
|
||
last = float(points[-1]["value"] or 0)
|
||
change_pct = round((last - first) / first * 100, 1) if first else None
|
||
group["data_points"] = len(points)
|
||
group["first_value"] = first
|
||
group["last_value"] = last
|
||
group["change_pct"] = change_pct
|
||
group["latest_variant"] = points[-1].get("variant") or ""
|
||
group["importance"] = group["priority"] + min(len(points), 8) * 6
|
||
series.append(group)
|
||
series.sort(key=lambda row: (row["importance"], row["data_points"], row["label"]), reverse=True)
|
||
return series[:limit]
|
||
|
||
|
||
def _local_metrics(plan: dict[str, Any], records: list[dict[str, Any]], week: int | None = None) -> dict[str, Any]:
|
||
selected = [item for item in records if week is None or item.get("week") == week]
|
||
completed = sum(1 for item in selected if item.get("status") == "completed")
|
||
planned = len(plan.get("days") or []) if week is not None else max(1, len({item.get("week") for item in selected})) * len(plan.get("days") or [])
|
||
progression_ids = {
|
||
str(item.get("progression_id") or "")
|
||
for session in selected for item in session.get("items", [])
|
||
if str(item.get("progression_id") or "")
|
||
}
|
||
clusters = _cluster_progression_series(plan, selected, limit=50)
|
||
return {
|
||
"sessions_with_data": len(selected),
|
||
"completed_sessions": completed,
|
||
"planned_sessions": planned,
|
||
"completion_rate_pct": round(completed / planned * 100) if planned else 0,
|
||
"documented_exercises": len(progression_ids),
|
||
"measurable_progression_clusters": len(clusters),
|
||
}
|
||
|
||
|
||
def _flow_for_progression(plan: dict[str, Any], progression_id: str, exercise: str, current: str) -> dict[str, Any] | None:
|
||
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 []
|
||
if not steps:
|
||
return None
|
||
normalized_current = _slug(current)
|
||
current_index = None
|
||
for index, step in enumerate(steps):
|
||
step_name = step.get("name", "") if isinstance(step, dict) else step
|
||
normalized_step = _slug(step_name)
|
||
if normalized_current and (normalized_current == normalized_step or normalized_current in normalized_step or normalized_step in normalized_current):
|
||
current_index = index
|
||
break
|
||
nodes = []
|
||
for index, step in enumerate(steps):
|
||
step_name = step.get("name", "") if isinstance(step, dict) else step
|
||
status = ""
|
||
if current_index is not None:
|
||
if index < current_index:
|
||
status = "done"
|
||
elif index == current_index:
|
||
status = "current"
|
||
elif index == current_index + 1:
|
||
status = "next"
|
||
nodes.append({"id": str(step.get("id") if isinstance(step, dict) else f"n{index + 1}"), "label": _plain_text(step_name), "status": status})
|
||
return {
|
||
"title": exercise,
|
||
"nodes": nodes,
|
||
"edges": [{"from": nodes[i]["id"], "to": nodes[i + 1]["id"]} for i in range(len(nodes) - 1)],
|
||
"insight": f"Dokumentierter Stand: {current}" if current else "Noch keine eindeutige Stufe dokumentiert.",
|
||
}
|
||
|
||
|
||
def _local_visuals(plan: dict[str, Any], records: list[dict[str, Any]], *, overall: bool) -> dict[str, Any]:
|
||
normalized = _cluster_progression_series(plan, records, limit=6)
|
||
charts = []
|
||
for group in normalized:
|
||
if len(group["points"]) < 2:
|
||
continue
|
||
labels = [point["label"] if overall else f"T{point['day']}" for point in group["points"]]
|
||
values = [point["value"] for point in group["points"]]
|
||
first_variant = group["points"][0].get("variant") or "unbekannt"
|
||
last_variant = group["points"][-1].get("variant") or "unbekannt"
|
||
charts.append({
|
||
"title": f"{group['label']} · normalisierter Verlauf",
|
||
"type": "line", "unit": group["unit"], "labels": labels,
|
||
"series": [{"name": group["label"], "values": values}],
|
||
"insight": (
|
||
f"{first_variant} → {last_variant}. Werte sind über Variantenfaktoren vergleichbar; "
|
||
f"Rohdaten bleiben im Analysearchiv erhalten."
|
||
),
|
||
})
|
||
latest: dict[str, dict[str, Any]] = {}
|
||
for session in records:
|
||
for item in session.get("items", []):
|
||
progression_id = str(item.get("progression_id") or "")
|
||
if progression_id:
|
||
latest[progression_id] = item
|
||
flows = []
|
||
for progression_id, item in latest.items():
|
||
flow = _flow_for_progression(
|
||
plan, progression_id, str(item.get("exercise") or progression_id), str(item.get("progression") or "")
|
||
)
|
||
if flow:
|
||
flows.append(flow)
|
||
if len(flows) >= 4:
|
||
break
|
||
return {
|
||
"charts": charts, "flowcharts": flows,
|
||
"normalization_version": PROGRESSION_NORMALIZATION_VERSION,
|
||
"cluster_count": len(normalized),
|
||
}
|
||
|
||
|
||
def _week_dataset(plan: dict[str, Any], tracker: dict[str, Any], week: int) -> dict[str, Any]:
|
||
records = _session_records(plan, tracker)
|
||
sessions = [item for item in records if item.get("week") == week]
|
||
if not sessions:
|
||
raise ValueError(f"Woche {week} enthält noch keine befüllten Sessiondaten.")
|
||
progression_ids = {
|
||
str(item.get("progression_id") or "")
|
||
for session in sessions for item in session.get("items", [])
|
||
if str(item.get("progression_id") or "")
|
||
}
|
||
profile = tracker.get("profile") if isinstance(tracker.get("profile"), dict) else {}
|
||
week_state = (tracker.get("week_statuses") or {}).get(str(week), {}) if isinstance(tracker.get("week_statuses"), dict) else {}
|
||
return {
|
||
"analysis_type": "week",
|
||
"analysis_kind": "final" if week_state.get("status") == "closed" else "interim",
|
||
"week_status": deepcopy(week_state),
|
||
"prompt_version": PROMPT_VERSION,
|
||
"analysis_schema_version": ANALYSIS_SCHEMA_VERSION,
|
||
"normalization_version": PROGRESSION_NORMALIZATION_VERSION,
|
||
"model": os.environ.get("OPENAI_MODEL", "gpt-5.5").strip() or "gpt-5.5",
|
||
"week": week,
|
||
"plan": _plan_context(plan, week=week, progression_ids=progression_ids),
|
||
"profile": {
|
||
"display_name": _plain_text(profile.get("display_name") or ""),
|
||
"start_date": str(profile.get("start_date") or ""),
|
||
"plan_notes": _plain_text(profile.get("plan_notes") or ""),
|
||
},
|
||
"previous_end_state": _latest_state_before(records, week),
|
||
"local_metrics": _local_metrics(plan, records, week),
|
||
"progression_normalization": {
|
||
"version": PROGRESSION_NORMALIZATION_VERSION,
|
||
"clusters": _cluster_progression_series(plan, sessions, limit=10),
|
||
},
|
||
"sessions": sessions,
|
||
}
|
||
|
||
|
||
def _compact_week_result(record: dict[str, Any]) -> dict[str, Any]:
|
||
result = record.get("result") if isinstance(record.get("result"), dict) else {}
|
||
return {
|
||
"week": record.get("week"),
|
||
"source_hash": record.get("source_hash"),
|
||
"headline": result.get("headline") or "",
|
||
"summary": result.get("summary") or result.get("overview", {}).get("summary", ""),
|
||
"metrics": result.get("metrics", [])[:4] if isinstance(result.get("metrics"), list) else [],
|
||
"exercise_updates": result.get("exercise_updates", [])[:6] if isinstance(result.get("exercise_updates"), list) else [],
|
||
"plan_adjustments": result.get("plan_adjustments", [])[:3] if isinstance(result.get("plan_adjustments"), list) else [],
|
||
"warnings": result.get("warnings", [])[:2] if isinstance(result.get("warnings"), list) else [],
|
||
}
|
||
|
||
|
||
def _overall_dataset(plan: dict[str, Any], tracker: dict[str, Any]) -> dict[str, Any]:
|
||
records = _session_records(plan, tracker)
|
||
cache = _read_analysis_cache(plan)
|
||
weeks = sorted({int(item.get("week") or 0) for item in records if int(item.get("week") or 0) > 0})
|
||
summaries = []
|
||
for week in weeks:
|
||
record = cache["weeks"].get(str(week))
|
||
if not isinstance(record, dict):
|
||
raise ValueError(f"Für Woche {week} fehlt eine aktuelle Wochenanalyse.")
|
||
summaries.append(_compact_week_result(record))
|
||
if not summaries:
|
||
raise ValueError("Noch keine Wochenanalyse für eine Gesamtanalyse vorhanden.")
|
||
weekly_coverage = []
|
||
for week in weeks:
|
||
metrics = _local_metrics(plan, records, week)
|
||
weekly_coverage.append({"week": week, **metrics})
|
||
return {
|
||
"analysis_type": "overall",
|
||
"prompt_version": PROMPT_VERSION,
|
||
"analysis_schema_version": ANALYSIS_SCHEMA_VERSION,
|
||
"normalization_version": PROGRESSION_NORMALIZATION_VERSION,
|
||
"model": os.environ.get("OPENAI_MODEL", "gpt-5.5").strip() or "gpt-5.5",
|
||
"week_statuses": deepcopy(tracker.get("week_statuses") or {}),
|
||
"plan": _plan_context(plan),
|
||
"weekly_analyses": summaries,
|
||
"local_aggregates": {
|
||
"coverage_by_week": weekly_coverage,
|
||
"overall": _local_metrics(plan, records),
|
||
"current_progressions": _latest_state_before(records, max(weeks) + 1),
|
||
"progression_normalization": {
|
||
"version": PROGRESSION_NORMALIZATION_VERSION,
|
||
"clusters": _cluster_progression_series(plan, records, limit=10),
|
||
},
|
||
},
|
||
}
|
||
|
||
|
||
def _meaningful_weeks(plan: dict[str, Any], tracker: dict[str, Any]) -> list[int]:
|
||
return sorted({int(item.get("week") or 0) for item in _session_records(plan, tracker) if int(item.get("week") or 0) > 0})
|
||
|
||
|
||
def _analysis_catalog(plan: dict[str, Any], tracker: dict[str, Any]) -> dict[str, Any]:
|
||
cache = _read_analysis_cache(plan)
|
||
week_entries = []
|
||
all_current = True
|
||
for week in _meaningful_weeks(plan, tracker):
|
||
dataset = _week_dataset(plan, tracker, week)
|
||
source_hash = _stable_hash(dataset)
|
||
record = cache["weeks"].get(str(week))
|
||
if not isinstance(record, dict):
|
||
status = "missing"
|
||
elif record.get("source_hash") != source_hash:
|
||
status = "stale"
|
||
else:
|
||
status = "current"
|
||
if status != "current":
|
||
all_current = False
|
||
week_entries.append({
|
||
"week": week, "status": status,
|
||
"week_status": dataset.get("week_status", {}).get("status", "open"),
|
||
"analysis_kind": dataset.get("analysis_kind", "interim"),
|
||
"sessions": len(dataset.get("sessions", [])),
|
||
"created_at": record.get("created_at") if isinstance(record, dict) else "",
|
||
"record_id": record.get("id") if isinstance(record, dict) else "",
|
||
})
|
||
overall_record = cache.get("overall")
|
||
if not week_entries:
|
||
overall_status = "missing"
|
||
overall_hash = ""
|
||
elif not all_current:
|
||
overall_status = "needs_weeks"
|
||
overall_hash = ""
|
||
else:
|
||
overall_dataset = _overall_dataset(plan, tracker)
|
||
overall_hash = _stable_hash(overall_dataset)
|
||
if not isinstance(overall_record, dict):
|
||
overall_status = "missing"
|
||
elif overall_record.get("source_hash") != overall_hash:
|
||
overall_status = "stale"
|
||
else:
|
||
overall_status = "current"
|
||
return {
|
||
"weeks": week_entries,
|
||
"overall": {
|
||
"status": overall_status,
|
||
"weeks": [entry["week"] for entry in week_entries],
|
||
"created_at": overall_record.get("created_at") if isinstance(overall_record, dict) else "",
|
||
"record_id": overall_record.get("id") if isinstance(overall_record, dict) else "",
|
||
"pending_weeks": [entry["week"] for entry in week_entries if entry["status"] != "current"],
|
||
},
|
||
}
|
||
|
||
|
||
def _call_ai(messages: list[dict[str, str]], model: str) -> Any:
|
||
from core import ai as core_ai
|
||
return core_ai.chat_json(messages, model=model)
|
||
|
||
|
||
def _violates_fixed_interval(text: Any) -> bool:
|
||
low = _plain_text(text).casefold()
|
||
return bool(re.search(
|
||
r"(?:haltezeit|arbeitszeit|intervall|hold).{0,30}(?:verläng|erhöh|länger)|(?:länger|mehr)\s*(?:als\s*)?\d*\s*(?:s|sek)",
|
||
low,
|
||
))
|
||
|
||
|
||
def _compact_ai_result(result: Any, training_format: dict[str, Any]) -> dict[str, Any]:
|
||
if not isinstance(result, dict):
|
||
raise ValueError("Die KI-Antwort besitzt nicht das erwartete JSON-Format")
|
||
clean = _sanitize(result)
|
||
if not isinstance(clean, dict):
|
||
raise ValueError("Ungültige KI-Antwort")
|
||
clean["metrics"] = clean.get("metrics", [])[:4] if isinstance(clean.get("metrics"), list) else []
|
||
clean["exercise_updates"] = clean.get("exercise_updates", [])[:6] if isinstance(clean.get("exercise_updates"), list) else []
|
||
clean["plan_adjustments"] = clean.get("plan_adjustments", [])[:3] if isinstance(clean.get("plan_adjustments"), list) else []
|
||
for index, adjustment in enumerate(clean["plan_adjustments"]):
|
||
if isinstance(adjustment, dict):
|
||
adjustment.setdefault("id", f"proposal-{index+1}")
|
||
adjustment.setdefault("action", "review")
|
||
clean["warnings"] = clean.get("warnings", [])[:2] if isinstance(clean.get("warnings"), list) else []
|
||
if training_format.get("fixed_interval"):
|
||
replacement = (
|
||
"Arbeitsintervall unverändert lassen; nur die im Plan erlaubte Übungsstufe, das Gewicht, "
|
||
"den kontrollierten Bewegungsumfang oder die technische Qualität anpassen."
|
||
)
|
||
for exercise in clean["exercise_updates"]:
|
||
if isinstance(exercise, dict) and _violates_fixed_interval(exercise.get("next_action")):
|
||
exercise["next_action"] = replacement
|
||
for adjustment in clean["plan_adjustments"]:
|
||
if isinstance(adjustment, dict):
|
||
combined = " ".join(str(adjustment.get(field) or "") for field in ("suggested_change", "manual_step"))
|
||
if _violates_fixed_interval(combined):
|
||
adjustment["suggested_change"] = replacement
|
||
adjustment["manual_step"] = "Im Trainingsplan-Modul nur Progressionsstufe oder Last anpassen; Tabata-Timer unverändert lassen."
|
||
return clean
|
||
|
||
|
||
def _run_progress_analysis(dataset: dict[str, Any], scope: str) -> dict[str, Any]:
|
||
model = os.environ.get("OPENAI_MODEL", "gpt-5.5").strip() or "gpt-5.5"
|
||
schema = {
|
||
"headline": "Kurzer Titel",
|
||
"summary": "Kompakte Einordnung in höchstens 4 Sätzen",
|
||
"data_quality": "Ein kurzer Satz zur Aussagekraft",
|
||
"metrics": [{"label": "Kennzahl", "value": "Wert", "detail": "kurze Einordnung"}],
|
||
"exercise_updates": [{
|
||
"name": "Übungscluster", "trend": "up|stable|down|unclear",
|
||
"current_level": "dokumentierter Stand", "evidence": "knapper Beleg aus Referenzwerten und Rohdaten",
|
||
"next_action": "nächster planverträglicher Schritt", "criterion": "messbares Kriterium",
|
||
}],
|
||
"plan_adjustments": [{
|
||
"id": "stabile kurze Vorschlags-ID",
|
||
"action": "change_progression|change_load|change_variant|keep|review",
|
||
"target": "Woche/Tag/Übung oder Block",
|
||
"target_exercise_id": "ID aus plan.days.blocks.exercises, wenn eindeutig",
|
||
"target_progression_id": "Progressions-ID, wenn eindeutig",
|
||
"target_step_id": "aktuelle oder vorgeschlagene Stufen-ID, wenn eindeutig",
|
||
"suggested_change": "konkreter Vorschlag",
|
||
"reason": "kurze Begründung", "manual_step": "was im Trainingsplan-Modul manuell zu ändern wäre",
|
||
"condition": "nur wenn/sonst beibehalten",
|
||
}],
|
||
"warnings": ["nur echte Auffälligkeiten"],
|
||
"conclusion": "ein knapper Abschlusssatz",
|
||
}
|
||
target = (
|
||
"Bewerte genau diese Planwoche und formuliere höchstens drei konkrete, optionale Anpassungen für die nächste Woche."
|
||
if scope == "week" else
|
||
"Bewerte die Entwicklung über die vorhandenen Wochen und formuliere höchstens drei konkrete, optionale Anpassungen für die nächste Woche beziehungsweise den nächsten Planabschnitt."
|
||
)
|
||
messages = [
|
||
{
|
||
"role": "system",
|
||
"content": (
|
||
"Du analysierst ein Trainingsprotokoll auf Deutsch. Der übergebene Trainingsplan ist bindend und muss die Empfehlungen bestimmen. "
|
||
"Verändere niemals eigenmächtig Sessionformat, Arbeitsintervall, Pausen, Rundenzahl, Satzanzahl oder Progressionslogik. "
|
||
"Bei Tabata oder anderen festen Intervallen darfst du insbesondere keine längeren Holds oder längere Arbeitszeiten empfehlen. "
|
||
"Progression erfolgt dann nur planverträglich, etwa über die vorgesehene schwierigere Variante, Gewicht, ROM, Tempo oder Technik. "
|
||
"Nutze ausschließlich die Daten im Input, erfinde keine Leistungen und gib keine medizinische Diagnose. "
|
||
"Eine Session mit Status stopped wurde vorzeitig beendet und darf nicht als abgeschlossen gewertet werden; vorhandene Teildaten dürfen vorsichtig berücksichtigt werden. "
|
||
"Nutze die deterministisch berechneten progression_normalization-Cluster als maßgebliche Zeitreihe und rechne die Faktoren nicht selbst neu. "
|
||
"Referenz-Reps vergleichen Varianten derselben dynamischen Bewegung; Referenzsekunden vergleichen ausschließlich isometrische Varianten. "
|
||
"kg·Reps und kg·s sind Lastvolumen und dürfen nicht mit Bodyweight-Referenzwerten vermischt werden. "
|
||
"Wähle die wichtigsten Cluster nach Planrelevanz und Datenpunkten. Bei mindestens vier brauchbaren Clustern sollst du vier bis sechs davon knapp bewerten. "
|
||
"Die Analyse soll trotzdem kompakt bleiben: maximal 4 Kennzahlen, 6 kurze Übungsupdates, 3 Plananpassungen und 2 Warnungen. "
|
||
"Plananpassungen sind nur Vorschläge zur manuellen Übernahme im Trainingsplan-Modul; behaupte nie, den Plan geändert zu haben. "
|
||
"Nutze für Vorschläge nach Möglichkeit die im Datensatz vorhandenen stabilen exercise_id-, progression_id- und step_id-Werte. "
|
||
+ target + " Antworte ausschließlich als JSON im vorgegebenen Schema."
|
||
),
|
||
},
|
||
{
|
||
"role": "user",
|
||
"content": "Gewünschtes JSON-Schema:\n" + json.dumps(schema, ensure_ascii=False)
|
||
+ "\n\nAnalyse-Datensatz:\n" + json.dumps(dataset, ensure_ascii=False),
|
||
},
|
||
]
|
||
result = _call_ai(messages, model)
|
||
training_format = dataset.get("plan", {}).get("training_format", {}) if isinstance(dataset.get("plan"), dict) else {}
|
||
return {
|
||
"model": model,
|
||
"request": {
|
||
"scope": scope,
|
||
"schema": schema,
|
||
"messages": messages,
|
||
"dataset": deepcopy(dataset),
|
||
},
|
||
"result": _compact_ai_result(result, training_format),
|
||
}
|
||
|
||
|
||
def _week_record(plan: dict[str, Any], tracker: dict[str, Any], week: int) -> dict[str, Any]:
|
||
dataset = _week_dataset(plan, tracker, week)
|
||
source_hash = _stable_hash(dataset)
|
||
output = _run_progress_analysis(dataset, "week")
|
||
created_at = _utc_now()
|
||
return {
|
||
"id": f"week-{week:02d}",
|
||
"type": "week", "week": week, "created_at": created_at,
|
||
"analysis_kind": dataset.get("analysis_kind", "interim"),
|
||
"prompt_version": PROMPT_VERSION, "analysis_schema_version": ANALYSIS_SCHEMA_VERSION,
|
||
"normalization_version": PROGRESSION_NORMALIZATION_VERSION,
|
||
"model": output["model"], "source_hash": source_hash,
|
||
"sessions_considered": len(dataset["sessions"]),
|
||
"request": output["request"],
|
||
"response": output["result"],
|
||
"visuals": _local_visuals(plan, dataset["sessions"], overall=False),
|
||
}
|
||
|
||
|
||
def _overall_record(plan: dict[str, Any], tracker: dict[str, Any]) -> dict[str, Any]:
|
||
dataset = _overall_dataset(plan, tracker)
|
||
source_hash = _stable_hash(dataset)
|
||
output = _run_progress_analysis(dataset, "overall")
|
||
records = _session_records(plan, tracker)
|
||
weeks = sorted({item.get("week") for item in records})
|
||
created_at = _utc_now()
|
||
return {
|
||
"id": "overall",
|
||
"type": "overall", "created_at": created_at,
|
||
"prompt_version": PROMPT_VERSION, "analysis_schema_version": ANALYSIS_SCHEMA_VERSION,
|
||
"normalization_version": PROGRESSION_NORMALIZATION_VERSION,
|
||
"model": output["model"], "source_hash": source_hash,
|
||
"sessions_considered": len(records), "weeks_considered": weeks,
|
||
"request": output["request"],
|
||
"response": output["result"],
|
||
"visuals": _local_visuals(plan, records, overall=True),
|
||
}
|
||
|
||
|
||
def _job_state(plan: dict[str, Any], job_id: str, **changes: Any) -> dict[str, Any]:
|
||
state = _read_analysis_state(plan)
|
||
if state.get("job_id") != job_id:
|
||
return state
|
||
now = datetime.now(timezone.utc)
|
||
state.update(changes)
|
||
state["heartbeat_at"] = now.isoformat(timespec="seconds")
|
||
state["lease_expires_at"] = (now + timedelta(seconds=_ANALYSIS_TIMEOUT_SECONDS)).isoformat(timespec="seconds")
|
||
return _write_analysis_state(plan, state)
|
||
|
||
|
||
def _analysis_heartbeat(plan_filename: str, job_id: str, stop_event: threading.Event) -> None:
|
||
while not stop_event.wait(_ANALYSIS_HEARTBEAT_SECONDS):
|
||
try:
|
||
path = _validated_plan_path(plan_filename)
|
||
if path is None:
|
||
return
|
||
plan = _normalize_plan(path)
|
||
state = _read_analysis_state(plan)
|
||
if state.get("job_id") != job_id or state.get("status") != "running":
|
||
return
|
||
_job_state(plan, job_id)
|
||
except Exception:
|
||
# Der Hauptworker entscheidet über Erfolg oder Fehler. Ein einzelner
|
||
# Heartbeat-Fehler darf den API-Aufruf nicht abbrechen.
|
||
pass
|
||
|
||
|
||
def _analysis_worker(plan_filename: str, job_id: str, scope: str, week: int | None) -> None:
|
||
heartbeat_stop = threading.Event()
|
||
heartbeat = threading.Thread(
|
||
target=_analysis_heartbeat,
|
||
args=(plan_filename, job_id, heartbeat_stop),
|
||
name=f"trainingstracker-heartbeat-{job_id}", daemon=True,
|
||
)
|
||
heartbeat.start()
|
||
try:
|
||
path = _validated_plan_path(plan_filename)
|
||
if path is None:
|
||
raise ValueError("Trainingsplan nicht mehr vorhanden.")
|
||
plan = _normalize_plan(path)
|
||
tracker = _load_tracker(plan)
|
||
if scope == "week":
|
||
assert week is not None
|
||
_job_state(plan, job_id, message=f"Woche {week} wird analysiert …", step=0, total_steps=1)
|
||
record = _week_record(plan, tracker, week)
|
||
_archive_analysis_record(plan, record)
|
||
selection = f"week:{week}"
|
||
else:
|
||
catalog = _analysis_catalog(plan, tracker)
|
||
pending = list(catalog.get("overall", {}).get("pending_weeks", []))
|
||
total = len(pending) + 1
|
||
for index, pending_week in enumerate(pending, start=1):
|
||
_job_state(
|
||
plan, job_id, message=f"Woche {pending_week} wird aktualisiert …",
|
||
step=index - 1, total_steps=total,
|
||
)
|
||
tracker = _load_tracker(plan)
|
||
record = _week_record(plan, tracker, pending_week)
|
||
_archive_analysis_record(plan, record)
|
||
_job_state(plan, job_id, message="Gesamtanalyse wird erstellt …", step=max(0, total - 1), total_steps=total)
|
||
tracker = _load_tracker(plan)
|
||
record = _overall_record(plan, tracker)
|
||
_archive_analysis_record(plan, record)
|
||
selection = "overall"
|
||
finished_total = 1 if scope == "week" else total
|
||
_job_state(
|
||
plan, job_id, status="done", message="Analyse abgeschlossen.",
|
||
selection=selection, step=finished_total, total_steps=finished_total,
|
||
finished_at=_utc_now(), error="",
|
||
)
|
||
except Exception as exc: # noqa: BLE001
|
||
try:
|
||
path = _validated_plan_path(plan_filename)
|
||
if path is not None:
|
||
plan = _normalize_plan(path)
|
||
_job_state(
|
||
plan, job_id, status="error", message="Analyse fehlgeschlagen.",
|
||
error=str(exc), finished_at=_utc_now(),
|
||
)
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
heartbeat_stop.set()
|
||
with _ANALYSIS_GUARD:
|
||
_ANALYSIS_THREADS.pop(plan_filename, None)
|
||
|
||
|
||
def _start_analysis(plan: dict[str, Any], tracker: dict[str, Any], scope: str, week: int | None) -> dict[str, Any]:
|
||
catalog = _analysis_catalog(plan, tracker)
|
||
if scope == "week":
|
||
entry = next((item for item in catalog["weeks"] if item["week"] == week), None)
|
||
if entry is None:
|
||
raise ValueError(f"Woche {week} enthält noch keine befüllten Sessiondaten.")
|
||
if entry["status"] == "current":
|
||
return {"cached": True, "selection": f"week:{week}"}
|
||
label = f"Woche {week}"
|
||
total_steps = 1
|
||
elif scope == "overall":
|
||
if not catalog["weeks"]:
|
||
raise ValueError("Noch keine befüllten Sessiondaten für eine Analyse vorhanden.")
|
||
if catalog["overall"]["status"] == "current":
|
||
return {"cached": True, "selection": "overall"}
|
||
pending = catalog["overall"].get("pending_weeks", [])
|
||
label = "Gesamtanalyse"
|
||
total_steps = len(pending) + 1
|
||
else:
|
||
raise ValueError("Unbekannter Analysebereich.")
|
||
|
||
with _analysis_file_lock(plan["source_file"]):
|
||
current_state = _read_analysis_state(plan)
|
||
if current_state.get("status") == "running":
|
||
raise RuntimeError("Es läuft bereits eine Progressionsanalyse.")
|
||
job_id = datetime.now(timezone.utc).strftime("job-%Y%m%dT%H%M%S%fZ")
|
||
state = {
|
||
"status": "running", "job_id": job_id, "scope": scope, "week": week,
|
||
"label": label, "message": "Analyse wird vorbereitet …", "step": 0,
|
||
"total_steps": total_steps, "started_at": _utc_now(), "heartbeat_at": _utc_now(),
|
||
"lease_expires_at": (datetime.now(timezone.utc) + timedelta(seconds=_ANALYSIS_TIMEOUT_SECONDS)).isoformat(timespec="seconds"),
|
||
"owner_pid": os.getpid(), "error": "",
|
||
}
|
||
_write_analysis_state(plan, state)
|
||
thread = threading.Thread(
|
||
target=_analysis_worker,
|
||
args=(plan["source_file"], job_id, scope, week),
|
||
name=f"trainingstracker-{job_id}", daemon=True,
|
||
)
|
||
with _ANALYSIS_GUARD:
|
||
_ANALYSIS_THREADS[plan["source_file"]] = thread
|
||
thread.start()
|
||
return {"cached": False, "job": state}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# HTTP
|
||
# ---------------------------------------------------------------------------
|
||
@app.after_request
|
||
def no_cache(response):
|
||
if request.path.startswith("/api/"):
|
||
response.headers["Cache-Control"] = "no-store"
|
||
return response
|
||
|
||
|
||
@app.route("/")
|
||
def index():
|
||
index_path = STATIC_DIR / "index.html"
|
||
source = index_path.read_text(encoding="utf-8")
|
||
base = request.script_root.rstrip("/")
|
||
source = source.replace("__APP_BASE__", base)
|
||
return Response(source, mimetype="text/html")
|
||
|
||
|
||
@app.route("/api/health")
|
||
def api_health():
|
||
return jsonify({
|
||
"ok": True,
|
||
"plans_dir": str(PLANS_DIR),
|
||
"sessions_dir": str(SESSIONS_DIR),
|
||
"analyses_dir": str(ANALYSES_DIR),
|
||
"plans_dir_exists": PLANS_DIR.is_dir(),
|
||
"openai_configured": _openai_ready(),
|
||
"versions": {
|
||
"plan_schema": PLAN_SCHEMA_VERSION,
|
||
"training_contract": CONTRACT_VERSION,
|
||
"tracker_schema": TRACKER_SCHEMA_VERSION,
|
||
"result_data": RESULT_DATA_VERSION,
|
||
"prompt": PROMPT_VERSION,
|
||
"analysis_schema": ANALYSIS_SCHEMA_VERSION,
|
||
"normalization": PROGRESSION_NORMALIZATION_VERSION,
|
||
},
|
||
})
|
||
|
||
|
||
@app.route("/api/plans")
|
||
def api_plans():
|
||
settings = _read_settings()
|
||
selected = settings.get("selected_plan")
|
||
plans = []
|
||
for filename, path in _known_plan_files().items():
|
||
try:
|
||
plan = _normalize_plan(path)
|
||
tracker = _load_tracker(plan)
|
||
completed = sum(
|
||
1 for session in tracker.get("sessions", {}).values()
|
||
if isinstance(session, dict) and session.get("status") == "completed"
|
||
)
|
||
plans.append({
|
||
"id": filename,
|
||
"filename": filename,
|
||
"name": plan["name"],
|
||
"title": plan["title"],
|
||
"subtitle": plan["subtitle"],
|
||
"weeks": plan["weeks"],
|
||
"days": len(plan["days"]),
|
||
"tracked": _tracker_path(filename).exists(),
|
||
"start_date": tracker.get("profile", {}).get("start_date", ""),
|
||
"completed_sessions": completed,
|
||
"source_changed": bool(tracker.get("source_changed")),
|
||
"plan_id": plan.get("plan_id"),
|
||
"published_revision": plan.get("published_revision"),
|
||
"tracker_revision": tracker.get("revision", 1),
|
||
})
|
||
except Exception:
|
||
continue
|
||
if selected not in {item["id"] for item in plans}:
|
||
selected = plans[0]["id"] if plans else None
|
||
return jsonify({"plans": plans, "selected": selected})
|
||
|
||
|
||
@app.route("/api/plans/<path:plan_id>/select", methods=["POST"])
|
||
def api_select_plan(plan_id: str):
|
||
if _validated_plan_path(plan_id) is None:
|
||
return jsonify({"error": "Trainingsplan nicht gefunden"}), 404
|
||
_save_selected(plan_id)
|
||
return jsonify({"ok": True, "selected": plan_id})
|
||
|
||
|
||
@app.route("/api/plans/<path:plan_id>")
|
||
def api_plan(plan_id: str):
|
||
path = _validated_plan_path(plan_id)
|
||
if path is None:
|
||
return jsonify({"error": "Trainingsplan nicht gefunden"}), 404
|
||
try:
|
||
plan = _normalize_plan(path)
|
||
tracker = _load_tracker(plan)
|
||
# Nur für die UI zusammensetzen; diese Felder existieren nicht in der Sessiondatei.
|
||
tracker["analysis_cache"] = _read_analysis_cache(plan)
|
||
tracker["analysis_state"] = _read_analysis_state(plan)
|
||
except Exception as exc:
|
||
return jsonify({"error": f"Plan konnte nicht gelesen werden: {exc}"}), 400
|
||
return jsonify({"plan": plan, "tracker": tracker, "analysis_catalog": _analysis_catalog(plan, tracker), "equivalence_guide": _public_equivalence_guide(plan), "capabilities": {"openai": _openai_ready()}})
|
||
|
||
|
||
@app.route("/api/plans/<path:plan_id>/tracker", methods=["PUT"])
|
||
def api_save_tracker(plan_id: str):
|
||
path = _validated_plan_path(plan_id)
|
||
if path is None:
|
||
return jsonify({"error": "Trainingsplan nicht gefunden"}), 404
|
||
if request.content_length and request.content_length > 5 * 1024 * 1024:
|
||
return jsonify({"error": "Tracker-Daten sind zu groß"}), 413
|
||
try:
|
||
plan = _normalize_plan(path)
|
||
payload = request.get_json(force=True)
|
||
saved = _save_tracker(plan, payload)
|
||
except RuntimeError as exc:
|
||
if str(exc).startswith("revision_conflict:"):
|
||
current = str(exc).split(":", 1)[1]
|
||
return jsonify({"error": "Die Sessiondaten wurden in einem anderen Tab oder Gerät geändert.", "code": "revision_conflict", "current_revision": int(current)}), 409
|
||
return jsonify({"error": str(exc)}), 409
|
||
except (ValueError, TypeError) as exc:
|
||
return jsonify({"error": str(exc)}), 400
|
||
return jsonify({"ok": True, "updated_at": saved["updated_at"], "revision": saved.get("revision", 1), "analysis_catalog": _analysis_catalog(plan, saved), "analysis_state": _read_analysis_state(plan)})
|
||
|
||
|
||
@app.route("/api/plans/<path:plan_id>/tracker/session", methods=["PATCH"])
|
||
def api_patch_tracker_session(plan_id: str):
|
||
path = _validated_plan_path(plan_id)
|
||
if path is None:
|
||
return jsonify({"error": "Trainingsplan nicht gefunden"}), 404
|
||
try:
|
||
plan = _normalize_plan(path)
|
||
saved = _patch_tracker(plan, request.get_json(force=True))
|
||
except RuntimeError as exc:
|
||
if str(exc).startswith("revision_conflict:"):
|
||
current = str(exc).split(":", 1)[1]
|
||
return jsonify({"error": "Die Session wurde in einem anderen Tab oder Gerät geändert.", "code": "revision_conflict", "current_revision": int(current)}), 409
|
||
return jsonify({"error": str(exc)}), 409
|
||
except (ValueError, TypeError) as exc:
|
||
return jsonify({"error": str(exc)}), 400
|
||
return jsonify({
|
||
"ok": True, "updated_at": saved["updated_at"], "revision": saved.get("revision", 1),
|
||
"analysis_catalog": _analysis_catalog(plan, saved), "analysis_state": _read_analysis_state(plan),
|
||
})
|
||
|
||
|
||
@app.route("/api/plans/<path:plan_id>/analysis", methods=["POST"])
|
||
def api_progress_analysis(plan_id: str):
|
||
"""Startet auf expliziten Button-Druck eine Wochen- oder Gesamtanalyse."""
|
||
path = _validated_plan_path(plan_id)
|
||
if path is None:
|
||
return jsonify({"error": "Trainingsplan nicht gefunden"}), 404
|
||
if not _openai_ready():
|
||
return jsonify({"error": "OpenAI-Key oder Modell fehlt – siehe boehmitools-Einstellungen."}), 400
|
||
try:
|
||
payload = request.get_json(silent=True) or {}
|
||
scope = str(payload.get("scope") or "overall")
|
||
week = _as_int(payload.get("week"), 0) if scope == "week" else None
|
||
if scope == "week" and week <= 0:
|
||
raise ValueError("Eine gültige Woche muss ausgewählt werden.")
|
||
plan = _normalize_plan(path)
|
||
tracker = _load_tracker(plan)
|
||
outcome = _start_analysis(plan, tracker, scope, week)
|
||
latest = _load_tracker(plan)
|
||
catalog = _analysis_catalog(plan, latest)
|
||
analysis_state = _read_analysis_state(plan)
|
||
analysis_cache = _read_analysis_cache(plan)
|
||
except RuntimeError as exc:
|
||
return jsonify({"error": str(exc), "analysis_state": _read_analysis_state(plan)}), 409
|
||
except (ValueError, TypeError) as exc:
|
||
return jsonify({"error": str(exc)}), 400
|
||
if outcome.get("cached"):
|
||
return jsonify({
|
||
"ok": True, "cached": True, "selection": outcome.get("selection"),
|
||
"analysis_state": analysis_state,
|
||
"analysis_cache": analysis_cache,
|
||
"analysis_catalog": catalog,
|
||
})
|
||
return jsonify({
|
||
"ok": True, "cached": False, "job": outcome.get("job"),
|
||
"analysis_state": analysis_state,
|
||
"analysis_catalog": catalog,
|
||
}), 202
|
||
|
||
|
||
@app.route("/api/plans/<path:plan_id>/analysis/status")
|
||
def api_progress_analysis_status(plan_id: str):
|
||
path = _validated_plan_path(plan_id)
|
||
if path is None:
|
||
return jsonify({"error": "Trainingsplan nicht gefunden"}), 404
|
||
try:
|
||
plan = _normalize_plan(path)
|
||
tracker = _load_tracker(plan)
|
||
catalog = _analysis_catalog(plan, tracker)
|
||
except Exception as exc: # noqa: BLE001
|
||
return jsonify({"error": f"Analysestatus konnte nicht gelesen werden: {exc}"}), 400
|
||
return jsonify({
|
||
"analysis_state": _read_analysis_state(plan),
|
||
"analysis_cache": _read_analysis_cache(plan),
|
||
"analysis_catalog": catalog,
|
||
"updated_at": tracker.get("updated_at", ""),
|
||
})
|
||
|
||
|
||
@app.route("/api/plans/<path:plan_id>/tracker/export")
|
||
def api_export_tracker(plan_id: str):
|
||
path = _validated_plan_path(plan_id)
|
||
if path is None:
|
||
return jsonify({"error": "Trainingsplan nicht gefunden"}), 404
|
||
tracker_path = _tracker_path(path.name)
|
||
if not tracker_path.exists():
|
||
plan = _normalize_plan(path)
|
||
_atomic_json_write(tracker_path, _default_tracker(plan))
|
||
return send_file(
|
||
tracker_path,
|
||
mimetype="application/json",
|
||
as_attachment=True,
|
||
download_name=f"tracking-{path.name}",
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
app.run(
|
||
host=os.environ.get("HOST", "0.0.0.0"),
|
||
port=int(os.environ.get("PORT", "8081")),
|
||
debug=os.environ.get("DEBUG") == "1",
|
||
)
|