# -*- coding: utf-8 -*- """Backend fuer einen schlanken Trainingsplan-Editor.""" from __future__ import annotations import io import json import os import re from copy import deepcopy from pathlib import Path from typing import Any, Callable from flask import Flask, Response, jsonify, request, send_file from schema_contract import default_config, normalize_training_config, stable_id, validate_training_config from storage import atomic_json_write, utc_now BASE_DIR = Path(__file__).resolve().parent def resolve_data_dir() -> Path: configured = os.environ.get("DATA_DIR", "").strip() if configured: return Path(configured).expanduser().resolve() suite_data = os.environ.get("BOEHMITOOLS_DATA", "").strip() for candidate in ( Path("/home/michael/boehmitools/runtime/training/data/trainingsplan"), Path("/home/michael/runtime/training/data/trainingsplan"), ): if any((candidate / "plans").glob("*.json")): return candidate.resolve() if suite_data: return (Path(suite_data).expanduser() / "trainingsplan").resolve() return (BASE_DIR / "data").resolve() DATA_DIR = resolve_data_dir() PLANS_DIR = DATA_DIR / "plans" ACTIVE_FILE = DATA_DIR / "active.txt" LEGACY_CONFIG = DATA_DIR / "config.json" BACKUPS_DIR = DATA_DIR / "backups" STATIC_DIR = BASE_DIR / "static" PLANS_DIR.mkdir(parents=True, exist_ok=True) BACKUPS_DIR.mkdir(parents=True, exist_ok=True) app = Flask(__name__, static_folder=None) def slugify(name: Any) -> str: text = str(name or "").strip().lower() text = text.replace("ä", "ae").replace("ö", "oe").replace("ü", "ue").replace("ß", "ss") return re.sub(r"[^a-z0-9]+", "-", text).strip("-") or "trainingsplan" def plan_path(pid: str) -> Path: return PLANS_DIR / f"{pid}.json" def unique_id(base: str) -> str: stem, index = base, 2 while plan_path(stem).exists(): stem = f"{base}-{index}" index += 1 return stem def _payload() -> dict[str, Any]: body = request.get_json(silent=True) return body if isinstance(body, dict) else {} def _normalize_wrapper(pid: str, raw: Any) -> dict[str, Any]: if not isinstance(raw, dict): raw = {} name = str(raw.get("name") or pid) published_source = raw.get("config") if isinstance(raw.get("config"), dict) else raw published = normalize_training_config(published_source, plan_id=str(raw.get("plan_id") or pid)) draft_source = raw.get("draft") if isinstance(raw.get("draft"), dict) else published draft = normalize_training_config(draft_source, plan_id=published.get("plan_id") or pid) revision = max(1, int(raw.get("revision") or 1)) published_revision = max(1, int(raw.get("published_revision") or 1)) return { "id": pid, "plan_id": str(raw.get("plan_id") or published.get("plan_id") or pid), "name": name, "revision": revision, "published_revision": published_revision, "created_at": str(raw.get("created_at") or utc_now()), "updated_at": str(raw.get("updated_at") or utc_now()), "published_at": str(raw.get("published_at") or raw.get("updated_at") or utc_now()), "config": published, "draft": draft, } def read_plan(pid: str) -> dict[str, Any]: with plan_path(pid).open("r", encoding="utf-8") as handle: return _normalize_wrapper(pid, json.load(handle)) def write_plan(pid: str, wrapper: dict[str, Any], *, backup: bool = True) -> dict[str, Any]: normalized = _normalize_wrapper(pid, wrapper) normalized["updated_at"] = utc_now() atomic_json_write(plan_path(pid), {k: v for k, v in normalized.items() if k != "id"}, BACKUPS_DIR if backup else None) return normalized def create_wrapper(pid: str, name: str, cfg: dict[str, Any] | None = None) -> dict[str, Any]: plan = normalize_training_config(cfg or default_config(), plan_id=pid) plan["plan_id"] = pid plan["meta"]["title"] = str(name or plan["meta"].get("title") or "Neuer Trainingsplan") now = utc_now() return { "id": pid, "plan_id": pid, "name": plan["meta"]["title"], "revision": 1, "published_revision": 1, "created_at": now, "updated_at": now, "published_at": now, "config": deepcopy(plan), "draft": deepcopy(plan), } def list_plan_ids() -> list[str]: ids = [] for path in PLANS_DIR.glob("*.json"): try: raw = json.loads(path.read_text(encoding="utf-8")) except Exception: raw = {} source = raw.get("config") if isinstance(raw, dict) and isinstance(raw.get("config"), dict) else raw if isinstance(source, dict) and source.get("type") == "recipe": continue ids.append(path.stem) return sorted(ids) def editable_plan_exists(pid: str) -> bool: return pid in list_plan_ids() def get_active() -> str | None: pid = ACTIVE_FILE.read_text(encoding="utf-8").strip() if ACTIVE_FILE.exists() else None ids = list_plan_ids() return pid if pid in ids else (ids[0] if ids else None) def set_active(pid: str) -> None: ACTIVE_FILE.parent.mkdir(parents=True, exist_ok=True) ACTIVE_FILE.write_text(pid, encoding="utf-8") def ensure_seed() -> None: if not list_plan_ids(): try: raw = json.loads(LEGACY_CONFIG.read_text(encoding="utf-8")) if LEGACY_CONFIG.exists() else default_config() except Exception: raw = default_config() write_plan("trainingsplan", create_wrapper("trainingsplan", "Trainingsplan", raw), backup=False) if get_active() is None and list_plan_ids(): set_active(list_plan_ids()[0]) def public_wrapper(wrapper: dict[str, Any], *, draft: bool = True) -> dict[str, Any]: return { "id": wrapper["id"], "plan_id": wrapper["plan_id"], "name": wrapper["name"], "revision": wrapper["revision"], "published_revision": wrapper["published_revision"], "updated_at": wrapper["updated_at"], "published_at": wrapper["published_at"], "has_unpublished_changes": wrapper["draft"] != wrapper["config"], "config": deepcopy(wrapper["draft"] if draft else wrapper["config"]), } def _save_draft(pid: str, wrapper: dict[str, Any], config: dict[str, Any]) -> dict[str, Any]: meta = config.get("meta") if isinstance(config.get("meta"), dict) else {} title = str(meta.get("title") or "").strip() if title: wrapper["name"] = title wrapper["draft"] = normalize_training_config(config, plan_id=wrapper["plan_id"]) wrapper["draft"]["meta"]["title"] = str(wrapper["name"]) wrapper["revision"] += 1 return write_plan(pid, wrapper) def _mutate_draft(pid: str, mutator: Callable[[dict[str, Any]], Any]) -> tuple[dict[str, Any], Any]: if not editable_plan_exists(pid): raise FileNotFoundError(pid) wrapper = read_plan(pid) draft = deepcopy(wrapper["draft"]) result = mutator(draft) if result is None or result is False or (isinstance(result, str) and result in {"day", "exercise", "item", "used"}): return wrapper, result saved = _save_draft(pid, wrapper, draft) return saved, result def _find_by_id(items: list[dict[str, Any]], item_id: str) -> dict[str, Any] | None: return next((item for item in items if str(item.get("id")) == item_id), None) def _replace_list_item(items: list[dict[str, Any]], item_id: str, replacement: dict[str, Any]) -> bool: for index, item in enumerate(items): if str(item.get("id")) == item_id: items[index] = replacement return True return False def _delete_list_item(items: list[dict[str, Any]], item_id: str) -> bool: before = len(items) items[:] = [item for item in items if str(item.get("id")) != item_id] return len(items) != before ensure_seed() @app.route("/") def index(): return Response((STATIC_DIR / "index.html").read_text(encoding="utf-8"), mimetype="text/html") @app.route("/api/status") def api_status(): return jsonify({ "name": "Trainingsplan-Backend", "status": "ok", "data_dir": str(DATA_DIR), "plans_dir": str(PLANS_DIR), "plans": list_plan_ids(), "active": get_active(), }) @app.route("/api/plans", methods=["GET"]) def api_plans(): active = get_active() plans = [] for pid in list_plan_ids(): try: plan = read_plan(pid) plans.append({ "id": pid, "name": plan["name"], "active": pid == active, "revision": plan["revision"], "published_revision": plan["published_revision"], "has_unpublished_changes": plan["draft"] != plan["config"], }) except Exception: continue return jsonify({"plans": plans, "active": active}) @app.route("/api/plans", methods=["POST"]) def api_plan_create(): body = _payload() name = str(body.get("name") or "Neuer Trainingsplan").strip() source = body.get("from") cfg = deepcopy(read_plan(str(source))["draft"]) if source and editable_plan_exists(str(source)) else default_config() pid = unique_id(slugify(name)) wrapper = create_wrapper(pid, name, cfg) write_plan(pid, wrapper, backup=False) set_active(pid) return jsonify(public_wrapper(wrapper)), 201 @app.route("/api/plans/", methods=["GET"]) def api_plan_get(pid: str): if not editable_plan_exists(pid): return jsonify({"error": "not found"}), 404 return jsonify(public_wrapper(read_plan(pid))) @app.route("/api/plans//select", methods=["POST"]) def api_plan_select(pid: str): if not editable_plan_exists(pid): return jsonify({"error": "not found"}), 404 set_active(pid) return jsonify(public_wrapper(read_plan(pid))) @app.route("/api/plans/", methods=["POST"]) def api_plan_save(pid: str): if not editable_plan_exists(pid): return jsonify({"error": "not found"}), 404 body = _payload() config = body.get("config") if isinstance(body.get("config"), dict) else body expected = body.get("expected_revision") if isinstance(body, dict) else None wrapper = read_plan(pid) if expected is not None and int(expected) != wrapper["revision"]: return jsonify({"error": "Der Plan wurde zwischenzeitlich geändert.", "code": "revision_conflict", "current_revision": wrapper["revision"]}), 409 saved = _save_draft(pid, wrapper, config) validation = validate_training_config(saved["draft"]) return jsonify({"ok": True, **public_wrapper(saved), "validation": {"errors": validation["errors"], "warnings": validation["warnings"]}}) @app.route("/api/plans//publish", methods=["POST"]) def api_plan_publish(pid: str): if not editable_plan_exists(pid): return jsonify({"error": "not found"}), 404 body = _payload() wrapper = read_plan(pid) expected = body.get("expected_revision") if expected is not None and int(expected) != wrapper["revision"]: return jsonify({"error": "Der Entwurf wurde zwischenzeitlich geändert.", "code": "revision_conflict", "current_revision": wrapper["revision"]}), 409 validation = validate_training_config(wrapper["draft"]) if validation["errors"]: return jsonify({"error": "Der Plan enthält Validierungsfehler.", "validation": {"errors": validation["errors"], "warnings": validation["warnings"]}}), 400 wrapper["config"] = deepcopy(validation["config"]) wrapper["draft"] = deepcopy(validation["config"]) wrapper["published_revision"] += 1 wrapper["revision"] += 1 wrapper["published_at"] = utc_now() saved = write_plan(pid, wrapper) return jsonify({"ok": True, **public_wrapper(saved, draft=False), "validation": {"errors": [], "warnings": validation["warnings"]}}) @app.route("/api/plans//rename", methods=["POST"]) def api_plan_rename(pid: str): if not editable_plan_exists(pid): return jsonify({"error": "not found"}), 404 name = str(_payload().get("name") or "").strip() if not name: return jsonify({"error": "name required"}), 400 wrapper = read_plan(pid) wrapper["name"] = name wrapper["draft"]["meta"]["title"] = name wrapper["config"]["meta"]["title"] = name wrapper["revision"] += 1 saved = write_plan(pid, wrapper) return jsonify({"id": pid, "name": name, "revision": saved["revision"]}) @app.route("/api/plans/", methods=["DELETE"]) def api_plan_delete(pid: str): ids = list_plan_ids() if pid not in ids: return jsonify({"error": "not found"}), 404 if len(ids) <= 1: return jsonify({"error": "letzter Plan kann nicht gelöscht werden"}), 400 atomic_json_write(BACKUPS_DIR / f"deleted-{pid}-{utc_now().replace(':', '')}.json", {k: v for k, v in read_plan(pid).items() if k != "id"}) plan_path(pid).unlink() if get_active() is None: set_active(list_plan_ids()[0]) return jsonify({"ok": True, "active": get_active()}) @app.route("/api/plans//export") def api_plan_export(pid: str): if not editable_plan_exists(pid): return jsonify({"error": "not found"}), 404 return send_file(io.BytesIO(plan_path(pid).read_bytes()), mimetype="application/json", as_attachment=True, download_name=f"{pid}.json") @app.route("/api/plans/import", methods=["POST"]) def api_plan_import(): raw = request.files["file"].read() if "file" in request.files else request.data if not raw: return jsonify({"error": "keine Datei"}), 400 try: obj = json.loads(raw.decode("utf-8")) except Exception as exc: return jsonify({"error": f"ungültiges JSON: {exc}"}), 400 cfg = obj.get("config", {}) if isinstance(obj, dict) and isinstance(obj.get("config"), dict) else (obj if isinstance(obj, dict) else {}) name = str(obj.get("name") or (cfg.get("meta") or {}).get("title") or "Importierter Trainingsplan") pid = unique_id(slugify(name)) wrapper = create_wrapper(pid, name, cfg) write_plan(pid, wrapper, backup=False) set_active(pid) return jsonify(public_wrapper(wrapper)), 201 @app.route("/api/plans//exercises", methods=["GET"]) def api_exercises_list(pid: str): if not editable_plan_exists(pid): return jsonify({"error": "not found"}), 404 return jsonify({"exercises": read_plan(pid)["draft"].get("exercises", [])}) @app.route("/api/plans//exercises", methods=["POST"]) def api_exercise_create(pid: str): body = _payload() name = str(body.get("name") or "").strip() if not name: return jsonify({"error": "name required"}), 400 def mutate(cfg: dict[str, Any]) -> dict[str, Any]: progressions = body.get("progressions") if isinstance(body.get("progressions"), list) else [] exercise = { "id": stable_id("exercise", cfg.get("plan_id"), len(cfg.get("exercises", [])), name), "name": name, "progressions": progressions, } current_progression_id = str(body.get("current_progression_id") or "").strip() if current_progression_id: exercise["current_progression_id"] = current_progression_id cfg.setdefault("exercises", []).append(exercise) return exercise try: saved, exercise = _mutate_draft(pid, mutate) except FileNotFoundError: return jsonify({"error": "not found"}), 404 return jsonify({"ok": True, "exercise": exercise, "revision": saved["revision"]}), 201 @app.route("/api/plans//exercises/", methods=["PATCH"]) def api_exercise_update(pid: str, exercise_id: str): body = _payload() def mutate(cfg: dict[str, Any]) -> dict[str, Any] | None: exercises = cfg.setdefault("exercises", []) exercise = _find_by_id(exercises, exercise_id) if not exercise: return None updated = deepcopy(exercise) for key in ("name", "current_progression_id"): if key in body: value = str(body.get(key) or "").strip() if value: updated[key] = value else: updated.pop(key, None) if isinstance(body.get("progressions"), list): updated["progressions"] = body["progressions"] _replace_list_item(exercises, exercise_id, updated) return updated try: saved, exercise = _mutate_draft(pid, mutate) except FileNotFoundError: return jsonify({"error": "not found"}), 404 if exercise is None: return jsonify({"error": "exercise not found"}), 404 return jsonify({"ok": True, "exercise": exercise, "revision": saved["revision"]}) @app.route("/api/plans//exercises/", methods=["DELETE"]) def api_exercise_delete(pid: str, exercise_id: str): force = request.args.get("force") in {"1", "true", "yes"} def mutate(cfg: dict[str, Any]) -> bool | str: used = any(item.get("exercise_id") == exercise_id for day in cfg.get("days", []) for item in day.get("exercises", [])) if used and not force: return "used" removed = _delete_list_item(cfg.setdefault("exercises", []), exercise_id) if force: for day in cfg.get("days", []): day["exercises"] = [item for item in day.get("exercises", []) if item.get("exercise_id") != exercise_id] return removed try: saved, result = _mutate_draft(pid, mutate) except FileNotFoundError: return jsonify({"error": "not found"}), 404 if result == "used": return jsonify({"error": "exercise is used by training days", "code": "exercise_in_use"}), 409 if not result: return jsonify({"error": "exercise not found"}), 404 return jsonify({"ok": True, "revision": saved["revision"]}) @app.route("/api/plans//days", methods=["GET"]) def api_days_list(pid: str): if not editable_plan_exists(pid): return jsonify({"error": "not found"}), 404 return jsonify({"days": read_plan(pid)["draft"].get("days", [])}) @app.route("/api/plans//days", methods=["POST"]) def api_day_create(pid: str): body = _payload() def mutate(cfg: dict[str, Any]) -> dict[str, Any]: days = cfg.setdefault("days", []) next_num = max([int(day.get("num") or 0) for day in days], default=0) + 1 num = int(body.get("num") or next_num) name = str(body.get("name") or body.get("focus") or f"Tag {num}").strip() day = { "id": stable_id("day", cfg.get("plan_id"), num, name), "num": num, "name": name, "warmup": {"items": [str(item).strip() for item in body.get("warmup", []) if str(item).strip()]} if isinstance(body.get("warmup"), list) else {"items": []}, "cooldown": {"items": [str(item).strip() for item in body.get("cooldown", []) if str(item).strip()]} if isinstance(body.get("cooldown"), list) else {"items": []}, "exercises": [], } days.append(day) return day try: saved, day = _mutate_draft(pid, mutate) except FileNotFoundError: return jsonify({"error": "not found"}), 404 return jsonify({"ok": True, "day": day, "revision": saved["revision"]}), 201 @app.route("/api/plans//days/", methods=["PATCH"]) def api_day_update(pid: str, day_id: str): body = _payload() def mutate(cfg: dict[str, Any]) -> dict[str, Any] | None: days = cfg.setdefault("days", []) day = _find_by_id(days, day_id) if not day: return None updated = deepcopy(day) for key in ("name",): if key in body: value = str(body.get(key) or "").strip() if value: updated[key] = value else: updated.pop(key, None) if "num" in body: updated["num"] = int(body.get("num") or updated.get("num") or 1) for key in ("warmup", "cooldown"): if isinstance(body.get(key), list): updated[key] = {"items": [str(item).strip() for item in body[key] if str(item).strip()]} elif isinstance(body.get(key), dict): updated[key] = body[key] _replace_list_item(days, day_id, updated) return updated try: saved, day = _mutate_draft(pid, mutate) except FileNotFoundError: return jsonify({"error": "not found"}), 404 if day is None: return jsonify({"error": "day not found"}), 404 return jsonify({"ok": True, "day": day, "revision": saved["revision"]}) @app.route("/api/plans//days/", methods=["DELETE"]) def api_day_delete(pid: str, day_id: str): def mutate(cfg: dict[str, Any]) -> bool: return _delete_list_item(cfg.setdefault("days", []), day_id) try: saved, removed = _mutate_draft(pid, mutate) except FileNotFoundError: return jsonify({"error": "not found"}), 404 if not removed: return jsonify({"error": "day not found"}), 404 return jsonify({"ok": True, "revision": saved["revision"]}) @app.route("/api/plans//days//exercises", methods=["POST"]) def api_day_exercise_add(pid: str, day_id: str): body = _payload() def mutate(cfg: dict[str, Any]) -> dict[str, Any] | str: day = _find_by_id(cfg.setdefault("days", []), day_id) if not day: return "day" exercise_id = str(body.get("exercise_id") or "").strip() if not exercise_id and str(body.get("name") or "").strip(): exercise = { "id": stable_id("exercise", cfg.get("plan_id"), len(cfg.get("exercises", [])), body.get("name")), "name": str(body.get("name")).strip(), } cfg.setdefault("exercises", []).append(exercise) exercise_id = exercise["id"] if not _find_by_id(cfg.get("exercises", []), exercise_id): return "exercise" item = { "id": stable_id("day-exercise", day_id, exercise_id, len(day.get("exercises", []))), "exercise_id": exercise_id, } day.setdefault("exercises", []).append(item) return item try: saved, item = _mutate_draft(pid, mutate) except FileNotFoundError: return jsonify({"error": "not found"}), 404 if item == "day": return jsonify({"error": "day not found"}), 404 if item == "exercise": return jsonify({"error": "exercise not found"}), 404 return jsonify({"ok": True, "day_exercise": item, "revision": saved["revision"]}), 201 @app.route("/api/plans//days//exercises/", methods=["PATCH"]) def api_day_exercise_update(pid: str, day_id: str, item_id: str): body = _payload() def mutate(cfg: dict[str, Any]) -> dict[str, Any] | str: day = _find_by_id(cfg.setdefault("days", []), day_id) if not day: return "day" item = _find_by_id(day.setdefault("exercises", []), item_id) if not item: return "item" updated = deepcopy(item) if "exercise_id" in body: exercise_id = str(body.get("exercise_id") or "").strip() if not _find_by_id(cfg.get("exercises", []), exercise_id): return "exercise" updated["exercise_id"] = exercise_id _replace_list_item(day["exercises"], item_id, updated) return updated try: saved, item = _mutate_draft(pid, mutate) except FileNotFoundError: return jsonify({"error": "not found"}), 404 if item == "day": return jsonify({"error": "day not found"}), 404 if item == "item": return jsonify({"error": "day exercise not found"}), 404 if item == "exercise": return jsonify({"error": "exercise not found"}), 404 return jsonify({"ok": True, "day_exercise": item, "revision": saved["revision"]}) @app.route("/api/plans//days//exercises/", methods=["DELETE"]) def api_day_exercise_delete(pid: str, day_id: str, item_id: str): def mutate(cfg: dict[str, Any]) -> bool | str: day = _find_by_id(cfg.setdefault("days", []), day_id) if not day: return "day" return _delete_list_item(day.setdefault("exercises", []), item_id) try: saved, result = _mutate_draft(pid, mutate) except FileNotFoundError: return jsonify({"error": "not found"}), 404 if result == "day": return jsonify({"error": "day not found"}), 404 if not result: return jsonify({"error": "day exercise not found"}), 404 return jsonify({"ok": True, "revision": saved["revision"]}) @app.route("/api/config", methods=["GET"]) def api_get_config(): pid = get_active() return jsonify(public_wrapper(read_plan(pid)) if pid else {"config": default_config()}) @app.route("/api/config", methods=["POST"]) def api_save_config(): pid = get_active() if not pid: return jsonify({"error": "kein aktiver Plan"}), 400 return api_plan_save(pid) @app.route("/api/default") def api_default(): return jsonify(default_config()) @app.route("/api/reset", methods=["POST"]) def api_reset(): pid = get_active() if not pid: return jsonify({"error": "kein aktiver Plan"}), 400 wrapper = read_plan(pid) wrapper["draft"] = normalize_training_config(default_config(), plan_id=wrapper["plan_id"]) wrapper["draft"]["meta"]["title"] = wrapper["name"] wrapper["revision"] += 1 saved = write_plan(pid, wrapper) return jsonify(public_wrapper(saved)) if __name__ == "__main__": app.run(host=os.environ.get("HOST", "0.0.0.0"), port=int(os.environ.get("PORT", "8080")), debug=False)