J4stMarkus/trainingsplan
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
135425c781efdb893d7c93788810e1587da379d9
trainingsplan/server.py
T
AI-Agent 135425c781
Image bauen und in die Registry schieben / build (push) Successful in 6s
Details
Wizard: User erst beim Abschliessen, Reihenfolge Sprache/Design/Modus, Reset im Einzelmodus 
- Benutzer wird erst beim 'Einrichtung abschliessen' angelegt (POST
  /api/register im wizardFinish); bis dahin frei umentscheidbar, im Modus-
  Schritt Hinweis + Live-Validierung (Finish deaktiviert bis ausgefuellt)
- Reihenfolge: Sprache -> Darstellung (Theme/OLED/Akzent) -> Modus
  (Einzel/Mehrbenutzer); Onboarding bleibt Sprache -> Darstellung
- Sprach-Schritt: Repo-Link + Kurztext zum Starten (nutzt den Raum)
- Wizard-Modal: min-height statt fixer Hoehe -> Modus-Schritt mit Formular
  wird nicht mehr abgeschnitten
- Server-Reset auch im Einzelmodus moeglich (nur Bestaetigungstext,
  kein Passwort noetig); Reset-Sektion erscheint dann entsprechend
- Tests: 167 Checks gruen (39+18+52+32+26)
2026-10-02 19:57:59 +00:00

1660 lines
62 KiB
Python
Raw Blame History

#!/usr/bin/env python3
"""
Trainingsplan - selbstgehosteter Trainingsverlaufs-Tracker.
Backend aus der Python-Standardbibliothek (http.server) + SQLite.
Alle fachlichen Einstellungen (Uebungen, Saetze, Reps, Gewichte, Maschinen,
Erklaerungen, Cardio-Felder) liegen in config.json - nicht im Code.
Mehrbenutzer:
- Solange KEIN Benutzer existiert, laeuft die App im offenen Einzelmodus
(user_id = 0). Der erste Benutzer registriert sich und uebernimmt dabei
vorhandene Daten (Migration).
- Danach ist fuer alle /api/*-Endpunkte (ausser /api/meta, /api/login,
/api/register) ein Bearer-Token noetig. Daten sind strikt pro Benutzer.
API:
GET / statische Datei index.html
GET /styles.css, /app.js statische Dateien
GET /api/meta {today, now, auth_required, users}
GET /api/config normalisierte Konfiguration
GET /api/settings Einstellungen des Benutzers
PUT /api/settings Teil-Update
GET /api/sessions Filter: activity, status, from, to
POST /api/sessions Session starten {activity, day}
PATCH /api/sessions/<id> Teil-Update: {meta, sets}
POST /api/sessions/<id>/autofill Saetze auf letztes Workout setzen
DELETE /api/sessions/<id> Session samt Saetzen loeschen
GET /api/workouts abgeschlossene Workouts, uebungen gruppiert
GET /api/reset-values letzte abgeschlossene Messung eines Sets
GET /api/stats?exercise= Steigerung je Termin
GET /api/export?format=json|csv kompletter Stand des Benutzers
POST /api/import {format, data}
POST /api/register erster Benutzer (nur wenn keine existieren)
POST /api/login {username, password} -> {token}
POST /api/logout Token invalidieren
GET /api/me {username, is_owner}
GET /api/users Benutzerliste (nur Owner)
POST /api/users Benutzer anlegen (nur Owner)
DELETE /api/users/<id> Benutzer + Daten loeschen (nur Owner)
Umgebung:
PORT Port (Default 8080)
TRAINING_STATE_DIR Verzeichnis fuer state.db (Default ./data)
"""
import csv
import hashlib
import hmac
import io
import json
import os
import secrets
import sqlite3
import sys
from datetime import date, datetime
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import parse_qs, urlparse
BASE = os.path.dirname(os.path.abspath(__file__))
STATE_DIR = os.environ.get("TRAINING_STATE_DIR", os.path.join(BASE, "data"))
DB = os.path.join(STATE_DIR, "state.db")
CONFIG_FILE = os.path.join(BASE, "config.json")
STATIC_FILES = {
"/": ("index.html", "text/html; charset=utf-8"),
"/index.html": ("index.html", "text/html; charset=utf-8"),
"/styles.css": ("styles.css", "text/css; charset=utf-8"),
"/app.js": ("app.js", "application/javascript; charset=utf-8"),
}
MAX_BODY = 8 * 1024 * 1024
OPEN_USER = 0 # user_id ohne Anmeldung (offener Modus / Alt-Daten)
PBKDF2_ITERATIONS = 200_000
CONFIG = {}
DEFAULT_SETTINGS = {
"theme": "dark", # dark | system | light
"oled": False, # reines Schwarz fuer den dunklen Modus
"accent": "white", # Schluessel aus der Palette (Frontend-CSS)
"autoReset": False, # letztes Workout der Kategorie beim Session-Start laden
"planPrefill": False, # Plan-Gewichte (startKg) beim Session-Start vorbefuellen
"showTips": True, # Uebungs-Hinweise ("4 Sätze · 8-10 reps · Start 18 kg")
"language": "de", # de | en
"setupDone": False, # Ersteinrichtungs-Wizard abgeschlossen (serverweit)
"onboarded": False, # persoenlicher Wizard (Sprache/Akzent) abgeschlossen
"weightUnit": "kg", # kg | lb (Anzeige-Einheit fuer Gewichte)
"startTab": "today", # Standard-Ansicht beim Start
}
# ---------------------------------------------------------------------------
# Konfiguration: normalisieren, damit das Frontend nie ueber rohe Struktur
# stolpert und der Nicht-Programmierer nur config.json pflegt.
# ---------------------------------------------------------------------------
def _num(v, default=None):
try:
return int(v)
except (TypeError, ValueError):
return default
def _norm_templates(tpls):
"""Cardio-Feld-Templates: {key, label, unit, kind, step, optional}."""
if isinstance(tpls, dict):
tpls = tpls.get("session") or tpls.get("set") or []
out = []
for t in tpls or []:
if not isinstance(t, dict):
continue
key = t.get("key")
if not key:
continue
out.append({
"key": key,
"label": t.get("label") or key,
"unit": t.get("unit") or "",
"kind": t.get("kind") or "number",
"step": t.get("step"),
"optional": bool(t.get("optional")),
})
return out
def _norm_sets(ex):
"""sets als Zahl (uniform) oder als Liste individueller Saetze."""
start = ex.get("startKg") or ""
reps = ex.get("reps") or ""
raw = ex.get("sets")
if isinstance(raw, int) or (isinstance(raw, str) and raw.strip().isdigit()):
count = max(1, _num(raw, 1))
return {"count": count, "weight": start, "reps": reps}
if isinstance(raw, list):
sets = []
for s in raw[:100]:
if not isinstance(s, dict):
continue
sets.append({
"weight": s.get("weight") if s.get("weight") is not None else start,
"reps": s.get("reps") if s.get("reps") is not None else reps,
})
return {"list": sets}
return {"count": 1, "weight": start, "reps": reps}
def _norm_cycle(cycle):
if not isinstance(cycle, dict):
return {"name": "Trainingsplan", "days": []}
days = []
for d in cycle.get("days") or []:
if not isinstance(d, dict):
continue
title = d.get("title") or d.get("key") or "Tag"
key = d.get("key") or title
exercises = []
for x in d.get("exercises") or []:
if not isinstance(x, dict) or not x.get("name"):
continue
name = x["name"]
exercises.append({
"name": name,
"key": f"{key}/{name}",
"sets": _norm_sets(x),
"target": x.get("reps") or "",
"startKg": x.get("startKg") or "",
"info": x.get("info") or "",
"tags": list(x.get("tags") or []),
})
days.append({"key": key, "title": title, "exercises": exercises})
return {"name": cycle.get("name") or "Trainingsplan", "days": days}
def normalize_config(raw):
cfg = raw if isinstance(raw, dict) else {}
profile = cfg.get("profile") or {}
weight_unit = profile.get("weightUnit") or "kg"
reps_label = profile.get("repsLabel") or "reps"
out = {
"profile": {
"name": profile.get("name") or "",
"weightUnit": weight_unit,
"repsLabel": reps_label,
},
"activities": [],
}
seen = set()
for a in cfg.get("activities") or []:
if not isinstance(a, dict):
continue
aid = str(a.get("id") or "").strip()
if not aid or aid in seen:
continue
seen.add(aid)
atype = a.get("type") or "strength"
item = {
"id": aid,
"name": a.get("name") or aid,
"type": atype,
"active": bool(a.get("active", True)),
"displayOrder": int(a.get("displayOrder") or 100),
"weightUnit": a.get("weightUnit") or weight_unit,
"repsLabel": a.get("repsLabel") or reps_label,
}
if atype == "cardio":
item["fieldTemplates"] = _norm_templates(a.get("fieldTemplates"))
else:
item["cycle"] = _norm_cycle(a.get("cycle"))
out["activities"].append(item)
out["activities"].sort(key=lambda x: x["displayOrder"])
return out
def load_config():
global CONFIG
if os.path.exists(CONFIG_FILE):
try:
with open(CONFIG_FILE, encoding="utf-8") as f:
CONFIG = normalize_config(json.load(f))
return
except Exception as exc:
print(f"WARNUNG: config.json konnte nicht gelesen werden ({exc}). "
f"Server laeuft mit leerer Konfiguration.", file=sys.stderr)
CONFIG = normalize_config({})
def cardio_keys():
"""Alle Cardio-Feld-Schluessel aus der Konfiguration (fuer CSV-Spalten)."""
keys = []
for a in CONFIG.get("activities", []):
if a.get("type") == "cardio":
for t in a.get("fieldTemplates", []):
if t.get("key") and t["key"] not in keys:
keys.append(t["key"])
return keys
def find_activity_config(aid):
for a in CONFIG.get("activities", []):
if a.get("id") == aid:
return a
return None
def find_exercise_config(exercise):
"""Konfig-Uebung zu einem Uebungsschluessel (Tag/Name) oder None."""
for a in CONFIG.get("activities", []):
if a.get("type") != "strength":
continue
for d in a.get("cycle", {}).get("days", []):
for x in d.get("exercises", []):
if x.get("key") == exercise:
return x
return None
# ---------------------------------------------------------------------------
# SQLite
# ---------------------------------------------------------------------------
def db():
con = sqlite3.connect(DB, timeout=30)
con.row_factory = sqlite3.Row
con.execute("PRAGMA foreign_keys=ON")
con.execute("PRAGMA busy_timeout=5000")
return con
def init_db():
os.makedirs(STATE_DIR, exist_ok=True)
con = db()
con.execute("PRAGMA journal_mode=WAL")
con.executescript("""
CREATE TABLE IF NOT EXISTS sessions(
id INTEGER PRIMARY KEY AUTOINCREMENT,
activity TEXT NOT NULL,
day TEXT NOT NULL DEFAULT '',
date TEXT NOT NULL,
start_time TEXT NOT NULL DEFAULT '',
end_time TEXT NOT NULL DEFAULT '',
note TEXT NOT NULL DEFAULT '',
fields TEXT NOT NULL DEFAULT '{}',
status TEXT NOT NULL DEFAULT 'active',
created_at TEXT NOT NULL DEFAULT '',
user_id INTEGER NOT NULL DEFAULT 0
);
CREATE TABLE IF NOT EXISTS entries(
id INTEGER PRIMARY KEY AUTOINCREMENT,
session_id INTEGER NOT NULL,
exercise TEXT NOT NULL,
sort INTEGER NOT NULL DEFAULT 0,
weight TEXT NOT NULL DEFAULT '',
reps TEXT NOT NULL DEFAULT '',
done INTEGER NOT NULL DEFAULT 0
);
CREATE INDEX IF NOT EXISTS idx_entries_session ON entries(session_id);
CREATE INDEX IF NOT EXISTS idx_sessions_status ON sessions(status);
CREATE TABLE IF NOT EXISTS settings(
user_id INTEGER NOT NULL DEFAULT 0,
key TEXT NOT NULL,
value TEXT NOT NULL,
PRIMARY KEY(user_id, key)
);
CREATE TABLE IF NOT EXISTS users(
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE,
password_hash TEXT NOT NULL,
salt TEXT NOT NULL,
created_at TEXT NOT NULL DEFAULT ''
);
CREATE TABLE IF NOT EXISTS tokens(
token_hash TEXT PRIMARY KEY,
user_id INTEGER NOT NULL,
created_at TEXT NOT NULL DEFAULT ''
);
""")
# Migration: aeltere DBs ohne user_id bekommen den offenen Modus
cols = [r[1] for r in con.execute("PRAGMA table_info(sessions)").fetchall()]
if "user_id" not in cols:
con.execute("ALTER TABLE sessions ADD COLUMN user_id INTEGER NOT NULL DEFAULT 0")
cols = [r[1] for r in con.execute("PRAGMA table_info(settings)").fetchall()]
if "user_id" not in cols:
# alte settings-Tabelle (key, value) -> neue (user_id, key, value)
con.execute("ALTER TABLE settings RENAME TO settings_legacy")
con.execute("""CREATE TABLE settings(
user_id INTEGER NOT NULL DEFAULT 0,
key TEXT NOT NULL,
value TEXT NOT NULL,
PRIMARY KEY(user_id, key))""")
con.execute("INSERT INTO settings(user_id,key,value) "
"SELECT 0, key, value FROM settings_legacy")
con.execute("DROP TABLE settings_legacy")
con.commit()
con.close()
def _clean_str(v):
return str(v).strip() if v is not None else ""
def _parse_num(v):
"""'18' | '18,5' | '18.5' -> float, sonst None (beide Dezimaltrennzeichen)."""
s = _clean_str(v).replace(",", ".")
if not s:
return None
try:
return float(s)
except ValueError:
return None
def _user_clause(user_id, alias=""):
"""WHERE-Teil fuer Benutzerscoping. OPEN_USER=0 (offener Modus/Alt-Daten)."""
col = f"{alias}user_id"
return f"{col} = ?", [user_id]
def user_count():
con = db()
row = con.execute("SELECT COUNT(*) AS c FROM users").fetchone()
con.close()
return row["c"]
def entries_for_session(sid, con=None):
own = con is None
if own:
con = db()
try:
rows = con.execute(
"SELECT id,exercise,sort,weight,reps,done FROM entries "
"WHERE session_id=? ORDER BY sort", (sid,)).fetchall()
return [dict(r) for r in rows]
finally:
if own:
con.close()
def fetch_sessions(user_id, activity=None, status=None,
from_d=None, to_d=None, limit=None):
sql = "SELECT * FROM sessions WHERE 1=1"
params = []
clause, extra = _user_clause(user_id)
sql += f" AND {clause}"
params += extra
if activity:
sql += " AND activity=?"
params.append(activity)
if status:
sql += " AND status=?"
params.append(status)
if from_d:
sql += " AND date>=?"
params.append(from_d)
if to_d:
sql += " AND date<=?"
params.append(to_d)
sql += " ORDER BY date DESC, id DESC"
if limit:
sql += " LIMIT ?"
params.append(limit)
con = db()
rows = con.execute(sql, params).fetchall()
ids = [r["id"] for r in rows]
entries = {}
if ids:
ph = ",".join("?" * len(ids))
for e in con.execute(
f"SELECT * FROM entries WHERE session_id IN ({ph}) "
"ORDER BY session_id, sort", ids).fetchall():
entries.setdefault(e["session_id"], []).append(dict(e))
con.close()
out = []
for r in rows:
d = dict(r)
d["fields"] = json.loads(d.get("fields") or "{}")
d["entries"] = entries.get(d["id"], [])
out.append(d)
return out
def fetch_session(user_id, sid):
con = db()
clause, extra = _user_clause(user_id)
r = con.execute(f"SELECT * FROM sessions WHERE id=? AND {clause}",
[sid] + extra).fetchone()
con.close()
if not r:
return None
d = dict(r)
d["fields"] = json.loads(d.get("fields") or "{}")
d["entries"] = entries_for_session(sid)
return d
def create_session(user_id, data):
activity = _clean_str(data.get("activity"))
if not activity:
raise ValueError("activity erforderlich")
sdate = _clean_str(data.get("date")) or date.today().isoformat()
day = _clean_str(data.get("day"))
start = _clean_str(data.get("start_time")) or datetime.now().strftime("%H:%M")
con = db()
cur = con.execute(
"INSERT INTO sessions(activity,day,date,start_time,status,created_at,user_id) "
"VALUES(?,?,?,?,?,?,?)",
(activity, day, sdate, start, "active",
datetime.now().isoformat(timespec="seconds"), user_id))
con.commit()
sid = cur.lastrowid
con.close()
# Optionale Vorbefuellung: letztes Workout und/oder Plan-Gewichte
settings = get_settings(user_id)
if settings.get("autoReset") or settings.get("planPrefill"):
act = find_activity_config(activity)
if act and act.get("type") == "strength":
fill_session_start(user_id, sid)
return fetch_session(user_id, sid)
def _reconcile_sets(con, sid, sets):
"""Set-Zustand einer Uebung aktualisieren, ohne andere Uebungen anzufassen.
Reconcile ist pro Uebung (exercise) abgegrenzt:
- Saetze werden ueber (session_id, exercise, sort) geupsert -> stabile IDs.
- Nur Saetze genau dieser Uebung, die im Payload fehlen, werden geloescht.
Eine Aenderung in Uebung A beruehrt Uebung B strukturell nie.
"""
by_ex = {}
for s in sets:
if not isinstance(s, dict):
continue
ex = _clean_str(s.get("exercise"))
if not ex:
continue
by_ex.setdefault(ex, []).append(s)
for ex, items in by_ex.items():
for i, s in enumerate(items):
sort = _num(s.get("sort"), i)
weight = _clean_str(s.get("weight"))
reps = _clean_str(s.get("reps"))
done = 1 if s.get("done") else 0
row = con.execute(
"SELECT id FROM entries WHERE session_id=? AND exercise=? AND sort=?",
(sid, ex, sort)).fetchone()
if row:
con.execute(
"UPDATE entries SET weight=?,reps=?,done=? WHERE id=?",
(weight, reps, done, row["id"]))
else:
con.execute(
"INSERT INTO entries(session_id,exercise,sort,weight,reps,done) "
"VALUES(?,?,?,?,?,?)",
(sid, ex, sort, weight, reps, done))
sorts = [_num(s.get("sort"), i) for i, s in enumerate(items)]
if sorts:
ph = ",".join("?" * len(sorts))
con.execute(
f"DELETE FROM entries WHERE session_id=? AND exercise=? "
f"AND sort NOT IN ({ph})",
[sid, ex] + sorts)
def update_session(user_id, sid, data):
con = db()
clause, extra = _user_clause(user_id)
row = con.execute(f"SELECT * FROM sessions WHERE id=? AND {clause}",
[sid] + extra).fetchone()
if not row:
con.close()
return None
meta = data.get("meta") or {}
if isinstance(meta, dict) and meta:
changes = []
params = []
for col in ("date", "day", "start_time", "end_time", "note", "status"):
if col in meta:
changes.append(f"{col}=?")
params.append(_clean_str(meta[col]))
if "fields" in meta and isinstance(meta["fields"], dict):
changes.append("fields=?")
params.append(json.dumps(meta["fields"], ensure_ascii=False))
if changes:
params.append(sid)
con.execute(f"UPDATE sessions SET {','.join(changes)} WHERE id=?", params)
# Abschliessen ohne explizite Endzeit: Serverzeit eintragen.
if meta.get("status") == "done" and not row["end_time"]:
con.execute("UPDATE sessions SET end_time=? WHERE id=?",
(datetime.now().strftime("%H:%M"), sid))
if "sets" in data and isinstance(data["sets"], list):
_reconcile_sets(con, sid, data["sets"])
con.commit()
con.close()
return fetch_session(user_id, sid)
def delete_session(user_id, sid):
con = db()
clause, extra = _user_clause(user_id)
row = con.execute(f"SELECT id FROM sessions WHERE id=? AND {clause}",
[sid] + extra).fetchone()
if not row:
con.close()
return 0
con.execute("DELETE FROM entries WHERE session_id=?", (sid,))
cur = con.execute("DELETE FROM sessions WHERE id=?", (sid,))
con.commit()
deleted = cur.rowcount
con.close()
return deleted
# ---------------------------------------------------------------------------
# Einstellungen
# ---------------------------------------------------------------------------
def get_settings(user_id=OPEN_USER):
"""Vollstaendige Einstellungen, typ-korrekt.
Schichtung: Defaults < Server-Standard (user_id=0, gesetzt im
Ersteinrichtungs-Wizard) < eigene Werte des Benutzers.
"""
con = db()
out = dict(DEFAULT_SETTINGS)
def apply(uid):
for r in con.execute("SELECT key,value FROM settings WHERE user_id=?",
(uid,)).fetchall():
k, v = r["key"], r["value"]
if k not in out:
continue
if isinstance(out[k], bool):
out[k] = str(v).lower() in ("1", "true", "yes", "on")
else:
out[k] = str(v)
if user_id != OPEN_USER:
apply(OPEN_USER) # Server-Standard (Sprache/Akzent aus dem Wizard)
apply(user_id)
con.close()
return out
def update_settings(user_id, patch):
if not isinstance(patch, dict):
raise ValueError("patch muss ein Objekt sein")
con = db()
bool_keys = ("autoReset", "showTips", "setupDone", "onboarded", "oled", "planPrefill")
for k, v in patch.items():
if k not in DEFAULT_SETTINGS:
continue
if k in bool_keys:
v = bool(v)
else:
v = str(v)
con.execute(
"INSERT INTO settings(user_id,key,value) VALUES(?,?,?) "
"ON CONFLICT(user_id,key) DO UPDATE SET value=excluded.value",
(user_id, k, json.dumps(v) if isinstance(v, bool) else v))
# Der erste Benutzer (Owner) setzt Sprache/Akzent auch als Server-Standard
if user_id == 1:
for k in ("language", "accent"):
if k in patch:
con.execute(
"INSERT INTO settings(user_id,key,value) VALUES(?,?,?) "
"ON CONFLICT(user_id,key) DO UPDATE SET value=excluded.value",
(OPEN_USER, k, patch[k]))
con.commit()
con.close()
return get_settings(user_id)
# ---------------------------------------------------------------------------
# Benutzer & Authentifizierung
# ---------------------------------------------------------------------------
def _hash_pw(password, salt=None):
if salt is None:
salt = secrets.token_hex(16)
dk = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"),
salt.encode("ascii"), PBKDF2_ITERATIONS)
return salt, dk.hex()
def _verify_pw(password, salt, expected):
_, dk = _hash_pw(password, salt)
return hmac.compare_digest(dk, expected)
def create_user(username, password):
username = _clean_str(username)
password = _clean_str(password)
if len(username) < 2:
raise ValueError("Benutzername zu kurz (mindestens 2 Zeichen)")
if len(password) < 4:
raise ValueError("Passwort zu kurz (mindestens 4 Zeichen)")
salt, pw_hash = _hash_pw(password)
con = db()
try:
cur = con.execute(
"INSERT INTO users(username,password_hash,salt,created_at) "
"VALUES(?,?,?,?)",
(username, pw_hash, salt,
datetime.now().isoformat(timespec="seconds")))
uid = cur.lastrowid
except sqlite3.IntegrityError:
con.close()
raise ValueError("Benutzername ist bereits vergeben")
# Erster Benutzer uebernimmt die Daten aus dem offenen Modus:
# - Sessions wandern zum Owner (Alt-Daten gehoeren ihm),
# - Settings (Sprache/Akzent aus dem Wizard) werden KOPIERT, damit der
# Server-Standard (user_id=0) fuer weitere Benutzer erhalten bleibt.
if uid == 1:
con.execute("UPDATE sessions SET user_id=? WHERE user_id=?",
(uid, OPEN_USER))
con.execute(
"INSERT INTO settings(user_id,key,value) "
"SELECT ?, key, value FROM settings WHERE user_id=?",
(uid, OPEN_USER))
token = _issue_token(con, uid)
con.commit()
con.close()
return uid, username, token
def _issue_token(con, user_id):
token = secrets.token_hex(32)
con.execute(
"INSERT INTO tokens(token_hash,user_id,created_at) VALUES(?,?,?)",
(hashlib.sha256(token.encode("ascii")).hexdigest(), user_id,
datetime.now().isoformat(timespec="seconds")))
return token
def login_user(username, password):
con = db()
row = con.execute("SELECT * FROM users WHERE username=?",
(_clean_str(username),)).fetchone()
if not row:
con.close()
return None
if not _verify_pw(password, row["salt"], row["password_hash"]):
con.close()
return None
token = _issue_token(con, row["id"])
con.commit()
con.close()
return row["id"], row["username"], token
def logout_token(token):
h = hashlib.sha256(token.encode("ascii")).hexdigest()
con = db()
con.execute("DELETE FROM tokens WHERE token_hash=?", (h,))
con.commit()
con.close()
def resolve_user(token):
if not token:
return None
h = hashlib.sha256(token.encode("ascii")).hexdigest()
con = db()
row = con.execute("SELECT user_id FROM tokens WHERE token_hash=?",
(h,)).fetchone()
con.close()
return row["user_id"] if row else None
def list_users():
con = db()
rows = con.execute("SELECT id,username,created_at FROM users ORDER BY id"
).fetchall()
con.close()
return [dict(r) for r in rows]
def delete_user(uid):
con = db()
sid_rows = con.execute("SELECT id FROM sessions WHERE user_id=?",
(uid,)).fetchall()
for r in sid_rows:
con.execute("DELETE FROM entries WHERE session_id=?", (r["id"],))
con.execute("DELETE FROM sessions WHERE user_id=?", (uid,))
con.execute("DELETE FROM settings WHERE user_id=?", (uid,))
con.execute("DELETE FROM tokens WHERE user_id=?", (uid,))
cur = con.execute("DELETE FROM users WHERE id=?", (uid,))
con.commit()
deleted = cur.rowcount
con.close()
return deleted
def change_own_password(uid, current, new):
"""Eigenes Passwort aendern. True bei Erfolg, False bei falschem altem."""
new = _clean_str(new)
if len(new) < 4:
raise ValueError("Passwort zu kurz (mindestens 4 Zeichen)")
con = db()
row = con.execute("SELECT * FROM users WHERE id=?", (uid,)).fetchone()
if not row or not _verify_pw(current, row["salt"], row["password_hash"]):
con.close()
return False
salt, pw_hash = _hash_pw(new)
con.execute("UPDATE users SET password_hash=?, salt=? WHERE id=?",
(pw_hash, salt, uid))
con.commit()
con.close()
return True
def set_user_password(uid, new):
"""Passwort eines Benutzers setzen/zuruecksetzen (Owner)."""
new = _clean_str(new)
if len(new) < 4:
raise ValueError("Passwort zu kurz (mindestens 4 Zeichen)")
salt, pw_hash = _hash_pw(new)
con = db()
row = con.execute("SELECT id FROM users WHERE id=?", (uid,)).fetchone()
if not row:
con.close()
return False
con.execute("UPDATE users SET password_hash=?, salt=? WHERE id=?",
(pw_hash, salt, uid))
# Sicherheitshalber alte Tokens des Benutzers invalidieren
con.execute("DELETE FROM tokens WHERE user_id=?", (uid,))
con.commit()
con.close()
return True
def reset_server():
"""Kompletter Reset: alle Sessions, Settings, Tokens und Benutzer loeschen."""
con = db()
for t in ("entries", "sessions", "settings", "tokens", "users"):
con.execute(f"DELETE FROM {t}")
con.execute("DELETE FROM sqlite_sequence")
con.commit()
con.close()
# ---------------------------------------------------------------------------
# Auto-Reset: letztes Workout einer Kategorie laden
# ---------------------------------------------------------------------------
def last_workout_sets(user_id, exercise, exclude_session=None):
"""Gefuellte Saetze der letzten abgeschlossenen Session einer Uebung.
Liefert nur Saetze mit Gewicht, in Original-Reihenfolge. Die (heutige)
in-Arbeit-Session zaehlt nie - nur status='done'.
"""
con = db()
clause, extra = _user_clause(user_id, alias="s.")
row = con.execute(
f"SELECT e.session_id FROM entries e "
f"JOIN sessions s ON s.id=e.session_id "
f"WHERE e.exercise=? AND s.status='done' AND e.weight!='' "
f"AND {clause} AND s.id!=? "
f"ORDER BY s.date DESC, s.id DESC LIMIT 1",
[exercise] + extra + [exclude_session or -1]).fetchone()
if not row:
con.close()
return []
sets = con.execute(
"SELECT weight,reps FROM entries WHERE session_id=? AND exercise=? "
"AND weight!='' ORDER BY sort",
(row["session_id"], exercise)).fetchall()
con.close()
return [{"weight": s["weight"], "reps": s["reps"]} for s in sets]
def _replace_sets(con, sid, ex_key, sets):
"""Saetze einer Uebung komplett ersetzen (done=0)."""
con.execute("DELETE FROM entries WHERE session_id=? AND exercise=?",
(sid, ex_key))
for i, s in enumerate(sets):
con.execute(
"INSERT INTO entries(session_id,exercise,sort,weight,reps,done) "
"VALUES(?,?,?,?,?,0)",
(sid, ex_key, i, s.get("weight", ""), s.get("reps", "")))
def fill_session_start(user_id, sid):
"""Frische Session beim Start vorbefuellen (pro Uebung):
- autoReset: letztes abgeschlossenes Workout der Uebung (nur gefuellte
Saetze); hat Vorrang.
- planPrefill: Plan-Gewichte (startKg) aus der Config, Anzahl wie im Plan.
Uebungen ohne Quelle bleiben leer.
"""
con = db()
sess = con.execute("SELECT * FROM sessions WHERE id=?", (sid,)).fetchone()
if not sess:
con.close()
return
exercises = []
act = find_activity_config(sess["activity"])
if act and act.get("type") == "strength":
for d in act.get("cycle", {}).get("days", []):
if d.get("key") == sess["day"]:
exercises = d.get("exercises", [])
break
settings = get_settings(user_id)
auto = settings.get("autoReset")
plan = settings.get("planPrefill")
if not auto and not plan:
con.close()
return
for ex in exercises:
ex_key = ex.get("key")
if auto:
sets = last_workout_sets(user_id, ex_key, exclude_session=sid)
if sets:
_replace_sets(con, sid, ex_key, sets)
continue
if plan:
cfg = ex.get("sets") or {}
count = len(cfg.get("list", [])) if cfg.get("list") else cfg.get("count", 0)
start = ex.get("startKg") or ""
if count:
_replace_sets(con, sid, ex_key,
[{"weight": start, "reps": ""}] * count)
con.commit()
con.close()
def autofill_session(user_id, sid, exercises=None):
"""Setze die Saetze der Session auf das letzte abgeschlossene Workout.
exercises: Liste von Uebungs-Schluesseln. None = alle Uebungen des
Sessions-Tags (aus der Config) plus alle, die bereits Saetze haben.
Nur Uebungen mit Verlauf werden ersetzt; die Abhak-Status werden
zurueckgesetzt (frische Session).
"""
con = db()
clause, extra = _user_clause(user_id)
sess = con.execute(f"SELECT * FROM sessions WHERE id=? AND {clause}",
[sid] + extra).fetchone()
if not sess:
con.close()
return None
if exercises is None:
exercises = []
act = find_activity_config(sess["activity"])
if act and act.get("type") == "strength":
for d in act.get("cycle", {}).get("days", []):
if d.get("key") == sess["day"]:
exercises = [ex["key"] for ex in d.get("exercises", [])]
break
for r in con.execute(
"SELECT DISTINCT exercise FROM entries WHERE session_id=?",
(sid,)).fetchall():
if r["exercise"] not in exercises:
exercises.append(r["exercise"])
filled = {}
for ex in exercises:
sets = last_workout_sets(user_id, ex, exclude_session=sid)
if sets:
filled[ex] = sets
for ex, sets in filled.items():
con.execute("DELETE FROM entries WHERE session_id=? AND exercise=?",
(sid, ex))
for i, s in enumerate(sets):
con.execute(
"INSERT INTO entries(session_id,exercise,sort,weight,reps,done) "
"VALUES(?,?,?,?,?,0)",
(sid, ex, i, s["weight"], s["reps"]))
con.commit()
con.close()
return fetch_session(user_id, sid)
def reset_values(user_id, exercise, set_idx, exclude_session=None):
"""Letztes abgeschlossenes Gewicht/Reps eines Sets.
In-Arbeit-Sessions (status=active) zaehlen nie mit - nur abgeschlossene
Tage bis einschliesslich heute.
"""
today = date.today().isoformat()
con = db()
clause, extra = _user_clause(user_id, alias="s.")
row = con.execute(
f"SELECT e.weight, e.reps FROM entries e "
f"JOIN sessions s ON s.id=e.session_id "
f"WHERE e.exercise=? AND e.sort=? AND s.status='done' "
f"AND e.weight!='' AND s.date<=? AND {clause} AND s.id!=? "
f"ORDER BY s.date DESC, s.id DESC LIMIT 1",
[exercise, set_idx, today] + extra + [exclude_session or -1]).fetchone()
con.close()
if not row:
return {}
return {"weight": row["weight"], "reps": row["reps"]}
def stats_for_exercise(user_id, exercise):
"""Bestes Gewicht je Termin + Differenz zur ersten Messung."""
con = db()
clause, extra = _user_clause(user_id, alias="s.")
rows = con.execute(
f"SELECT s.date, e.weight, e.reps FROM entries e "
f"JOIN sessions s ON s.id=e.session_id "
f"WHERE e.exercise=? AND s.status='done' AND e.weight!='' "
f"AND {clause} "
f"ORDER BY s.date ASC, s.id ASC, e.sort ASC",
[exercise] + extra).fetchall()
con.close()
best = {}
reps = {}
for r in rows:
w = _parse_num(r["weight"])
if w is None:
continue
d = r["date"]
if d not in best or w > best[d]:
best[d] = w
reps[d] = r["reps"]
first = None
out = []
for d in sorted(best):
if first is None:
first = best[d]
out.append({
"date": d,
"weight": best[d],
"reps": reps[d],
"delta": round(best[d] - first, 2),
})
return out
# ---------------------------------------------------------------------------
# Workouts-API: abgeschlossene Sessions, Uebungen gruppiert
# ---------------------------------------------------------------------------
def fetch_workouts(user_id, activity=None, from_d=None, to_d=None, limit=None):
sessions = fetch_sessions(user_id, activity=activity, status="done",
from_d=from_d, to_d=to_d, limit=limit)
out = []
for s in sessions:
act = find_activity_config(s["activity"])
by_ex = {}
for e in s["entries"]:
by_ex.setdefault(e["exercise"], []).append(e)
exercises = []
for ex_key, sets in by_ex.items():
ex_cfg = find_exercise_config(ex_key)
exercises.append({
"exercise": ex_key,
"name": ex_cfg["name"] if ex_cfg else ex_key,
"sets": sorted(
[{"sort": x["sort"], "weight": x["weight"],
"reps": x["reps"], "done": x["done"]} for x in sets],
key=lambda x: x["sort"]),
})
duration = None
if s["start_time"] and s["end_time"]:
try:
h1, m1 = (int(x) for x in s["start_time"].split(":"))
h2, m2 = (int(x) for x in s["end_time"].split(":"))
duration = (h2 * 60 + m2 - (h1 * 60 + m1)) % (24 * 60)
except Exception:
duration = None
out.append({
"id": s["id"],
"activity": s["activity"],
"activity_name": act["name"] if act else s["activity"],
"day": s["day"],
"date": s["date"],
"start_time": s["start_time"],
"end_time": s["end_time"],
"duration_min": duration,
"note": s["note"],
"exercises": exercises,
"fields": s["fields"],
})
return out
# ---------------------------------------------------------------------------
# Export / Import
# ---------------------------------------------------------------------------
EXPORT_FORMAT = "trainingsplan-export-v2"
def export_json(user_id):
return {
"format": EXPORT_FORMAT,
"exported_at": datetime.now().isoformat(timespec="seconds"),
"config": CONFIG,
"sessions": fetch_sessions(user_id),
}
def export_csv(user_id):
keys = cardio_keys()
header = (["session_id", "activity", "day", "date", "start_time",
"end_time", "note", "status", "kind", "exercise", "set",
"weight", "reps", "done"] + keys)
bio = io.StringIO()
writer = csv.writer(bio)
writer.writerow(header)
for s in fetch_sessions(user_id):
base = [s["id"], s["activity"], s["day"], s["date"], s["start_time"],
s["end_time"], s["note"], s["status"]]
if s["entries"]:
for e in s["entries"]:
row = base + ["set", e["exercise"], e["sort"] + 1,
e["weight"], e["reps"], e["done"]]
row += [s["fields"].get(k, "") for k in keys]
writer.writerow(row)
else:
row = base + ["cardio", "", "", "", "", ""]
row += [s["fields"].get(k, "") for k in keys]
writer.writerow(row)
return "\ufeff" + bio.getvalue()
def import_json_doc(user_id, doc):
con = db()
created = updated = 0
demo_ids = {s.get("id") for s in doc.get("sessions") or []
if isinstance(s, dict) and isinstance(s.get("id"), int)}
next_free = None
def _next_free():
# Freie ID oberhalb aller vorhandenen UND aller Import-IDs, damit
# keine Kollision mit spaeteren Sessions desselben Imports entsteht.
nonlocal next_free
if next_free is None:
mdb = con.execute(
"SELECT COALESCE(MAX(id),0) AS m FROM sessions").fetchone()["m"]
mdoc = max(demo_ids) if demo_ids else 0
next_free = max(mdb, mdoc) + 1
return next_free
for s in doc.get("sessions") or []:
if not isinstance(s, dict):
continue
sid = s.get("id")
meta = (
_clean_str(s.get("activity")),
_clean_str(s.get("day")),
_clean_str(s.get("date")),
_clean_str(s.get("start_time")),
_clean_str(s.get("end_time")),
_clean_str(s.get("note")),
json.dumps(s.get("fields") or {}, ensure_ascii=False),
s.get("status") or "done",
)
clause, extra = _user_clause(user_id)
if sid and con.execute(f"SELECT id FROM sessions WHERE id=? AND {clause}",
[sid] + extra).fetchone():
con.execute(
"UPDATE sessions SET activity=?,day=?,date=?,start_time=?,"
"end_time=?,note=?,fields=?,status=? WHERE id=?",
meta + (sid,))
con.execute("DELETE FROM entries WHERE session_id=?", (sid,))
updated += 1
else:
# ID uebernehmen, wenn frei (Backup-Wiederherstellung); sonst
# freie ID oberhalb aller bekannten IDs (keine Kollision).
if sid and not con.execute("SELECT id FROM sessions WHERE id=?",
(sid,)).fetchone():
cur = con.execute(
"INSERT INTO sessions(id,activity,day,date,start_time,"
"end_time,note,fields,status,created_at,user_id) "
"VALUES(?,?,?,?,?,?,?,?,?,?,?)",
[sid] + list(meta) +
[datetime.now().isoformat(timespec="seconds"), user_id])
sid = cur.lastrowid
else:
nf = _next_free()
cur = con.execute(
"INSERT INTO sessions(id,activity,day,date,start_time,"
"end_time,note,fields,status,created_at,user_id) "
"VALUES(?,?,?,?,?,?,?,?,?,?,?)",
[nf] + list(meta) +
[datetime.now().isoformat(timespec="seconds"), user_id])
sid = nf
next_free += 1
created += 1
for i, e in enumerate(s.get("entries") or []):
if not isinstance(e, dict):
continue
con.execute(
"INSERT INTO entries(session_id,exercise,sort,weight,reps,done) "
"VALUES(?,?,?,?,?,?)",
(sid, _clean_str(e.get("exercise")), i,
_clean_str(e.get("weight")), _clean_str(e.get("reps")),
1 if e.get("done") else 0))
con.commit()
con.close()
return created, updated
def import_csv_text(user_id, text):
text = text.lstrip("\ufeff")
reader = csv.DictReader(io.StringIO(text))
keys = cardio_keys()
groups = {}
order = []
for r in reader:
sid = r.get("session_id") or ""
if sid.strip().isdigit():
key = "sid:" + sid
else:
key = "|".join([r.get("activity", ""), r.get("day", ""),
r.get("date", ""), r.get("start_time", ""),
r.get("end_time", ""), r.get("note", "")])
if key not in groups:
groups[key] = {"rows": [], "sid": sid if sid.strip().isdigit() else None}
order.append(key)
groups[key]["rows"].append(r)
con = db()
clause, extra = _user_clause(user_id)
created = updated = 0
demo_ids = set()
for r in reader:
sid = r.get("session_id") or ""
if sid.strip().isdigit():
demo_ids.add(int(sid))
next_free = None
def _next_free():
nonlocal next_free
if next_free is None:
mdb = con.execute(
"SELECT COALESCE(MAX(id),0) AS m FROM sessions").fetchone()["m"]
mdoc = max(demo_ids) if demo_ids else 0
next_free = max(mdb, mdoc) + 1
return next_free
for key in order:
g = groups[key]
rows = g["rows"]
first = rows[0]
fields = {}
for k in keys:
v = _clean_str(first.get(k))
if v:
fields[k] = v
meta = (_clean_str(first.get("activity")), _clean_str(first.get("day")),
_clean_str(first.get("date")), _clean_str(first.get("start_time")),
_clean_str(first.get("end_time")), _clean_str(first.get("note")),
json.dumps(fields, ensure_ascii=False),
_clean_str(first.get("status")) or "done")
sid = g["sid"]
if sid and con.execute(f"SELECT id FROM sessions WHERE id=? AND {clause}",
[sid] + extra).fetchone():
con.execute(
"UPDATE sessions SET activity=?,day=?,date=?,start_time=?,"
"end_time=?,note=?,fields=?,status=? WHERE id=?",
meta + (sid,))
con.execute("DELETE FROM entries WHERE session_id=?", (sid,))
updated += 1
else:
if sid and not con.execute("SELECT id FROM sessions WHERE id=?",
(sid,)).fetchone():
cur = con.execute(
"INSERT INTO sessions(id,activity,day,date,start_time,"
"end_time,note,fields,status,created_at,user_id) "
"VALUES(?,?,?,?,?,?,?,?,?,?,?)",
[sid] + list(meta) +
[datetime.now().isoformat(timespec="seconds"), user_id])
sid = cur.lastrowid
else:
nf = _next_free()
cur = con.execute(
"INSERT INTO sessions(id,activity,day,date,start_time,"
"end_time,note,fields,status,created_at,user_id) "
"VALUES(?,?,?,?,?,?,?,?,?,?,?)",
[nf] + list(meta) +
[datetime.now().isoformat(timespec="seconds"), user_id])
sid = nf
next_free += 1
created += 1
for r in rows:
if _clean_str(r.get("kind")) != "set":
continue
set_no = _num(r.get("set"), 1) or 1
con.execute(
"INSERT INTO entries(session_id,exercise,sort,weight,reps,done) "
"VALUES(?,?,?,?,?,?)",
(sid, _clean_str(r.get("exercise")), max(0, set_no - 1),
_clean_str(r.get("weight")), _clean_str(r.get("reps")),
1 if _clean_str(r.get("done")) in ("1", "true", "True") else 0))
con.commit()
con.close()
return created, updated
# ---------------------------------------------------------------------------
# HTTP-Handler
# ---------------------------------------------------------------------------
class Handler(BaseHTTPRequestHandler):
server_version = "Trainingsplan/3"
def _send(self, code, body, ctype):
b = body if isinstance(body, bytes) else body.encode("utf-8")
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(b)))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(b)
def _send_json(self, code, obj):
self._send(code, json.dumps(obj, ensure_ascii=False),
"application/json; charset=utf-8")
def _read_body(self):
length = self.headers.get("Content-Length")
length = int(length) if length and length.isdigit() else 0
if length <= 0:
return b""
if length > MAX_BODY:
raise ValueError("Body zu gross")
return self.rfile.read(length)
def _read_json(self):
raw = self._read_body()
try:
return json.loads(raw.decode("utf-8"))
except Exception:
return {}
def _bearer_token(self):
auth = self.headers.get("Authorization", "")
if auth.startswith("Bearer "):
return auth[7:].strip()
return None
def _require_user(self):
"""Gibt user_id zurueck oder sendet 401 und liefert None.
Ohne registrierten Benutzer gilt der offene Modus (user_id=0).
Sobald ein Benutzer existiert, ist fuer alle Daten ein Token noetig.
"""
if user_count() == 0:
return OPEN_USER
uid = resolve_user(self._bearer_token())
if uid is not None:
return uid
self._send(401, json.dumps({"error": "Anmeldung erforderlich"}),
"application/json; charset=utf-8")
return None
def _send_file(self, path):
fname, ctype = STATIC_FILES[path]
fpath = os.path.join(BASE, fname)
if not os.path.exists(fpath):
self._send(404, "nicht gefunden", "text/plain; charset=utf-8")
return
with open(fpath, "rb") as f:
self._send(200, f.read(), ctype)
def log_message(self, fmt, *args):
pass
# -- GET --------------------------------------------------------------
def do_GET(self):
parsed = urlparse(self.path)
path = parsed.path
q = parse_qs(parsed.query)
try:
if path in STATIC_FILES:
self._send_file(path)
elif path in ("/api/meta", "/api/meta/"):
now = datetime.now()
self._send_json(200, {
"today": date.today().isoformat(),
"now": now.strftime("%H:%M"),
"auth_required": user_count() > 0,
"users": user_count(),
"language": get_settings(OPEN_USER)["language"],
})
elif path in ("/api/me", "/api/me/"):
uid = self._require_user()
if uid is None:
return
if uid == OPEN_USER:
self._send_json(200, {"username": None, "is_owner": False})
else:
con = db()
row = con.execute("SELECT username FROM users WHERE id=?",
(uid,)).fetchone()
con.close()
self._send_json(200, {"username": row["username"],
"is_owner": uid == 1})
else:
uid = self._require_user()
if uid is None:
return
if path in ("/api/config", "/api/config/"):
self._send_json(200, CONFIG)
elif path in ("/api/settings", "/api/settings/"):
self._send_json(200, get_settings(uid))
elif path in ("/api/sessions", "/api/sessions/"):
self._send_json(200, fetch_sessions(
uid,
activity=q.get("activity", [None])[0],
status=q.get("status", [None])[0],
from_d=q.get("from", [None])[0],
to_d=q.get("to", [None])[0]))
elif path in ("/api/workouts", "/api/workouts/"):
limit = _num(q.get("limit", [None])[0], None)
self._send_json(200, fetch_workouts(
uid,
activity=q.get("activity", [None])[0],
from_d=q.get("from", [None])[0],
to_d=q.get("to", [None])[0],
limit=limit))
elif path in ("/api/reset-values", "/api/reset-values/"):
exercise = q.get("exercise", [""])[0]
set_idx = _num(q.get("set", ["0"])[0], 0) or 0
self._send_json(200, reset_values(uid, exercise, set_idx))
elif path in ("/api/stats", "/api/stats/"):
exercise = q.get("exercise", [""])[0]
self._send_json(200, stats_for_exercise(uid, exercise))
elif path in ("/api/users", "/api/users/"):
self.require_owner(uid, lambda: self._send_json(
200, list_users()))
elif path in ("/api/export", "/api/export/"):
self.do_export(uid, q)
elif path == "/favicon.ico":
self._send(204, b"", "image/x-icon")
else:
self._send(404, "nicht gefunden", "text/plain; charset=utf-8")
except Exception as exc:
print(f"Fehler in {path}: {exc}", file=sys.stderr)
self._send(500, json.dumps({"error": str(exc)}),
"application/json; charset=utf-8")
def require_owner(self, uid, then):
if uid == 1:
then()
else:
self._send(403, json.dumps({"error": "nur der erste Benutzer"}),
"application/json; charset=utf-8")
def do_export(self, uid, q):
fmt = q.get("format", ["json"])[0]
if fmt == "csv":
body = export_csv(uid)
ctype = "text/csv; charset=utf-8"
fname = "trainingsplan.csv"
else:
body = json.dumps(export_json(uid), ensure_ascii=False, indent=1)
ctype = "application/json; charset=utf-8"
fname = "trainingsplan-backup.json"
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Disposition",
f'attachment; filename="{fname}"')
self.send_header("Content-Length", str(len(body.encode("utf-8"))))
self.send_header("Cache-Control", "no-store")
self.end_headers()
self.wfile.write(body.encode("utf-8"))
# -- POST -------------------------------------------------------------
def do_POST(self):
path = urlparse(self.path).path
try:
if path in ("/api/register", "/api/register/"):
self.do_register()
elif path in ("/api/login", "/api/login/"):
self.do_login()
elif path in ("/api/logout", "/api/logout/"):
logout_token(self._bearer_token() or "")
self._send_json(200, {"ok": True})
elif path in ("/api/password", "/api/password/"):
uid = self._require_user()
if uid is None:
return
data = self._read_json()
if change_own_password(uid, data.get("current_password", ""),
data.get("new_password", "")):
self._send_json(200, {"ok": True})
else:
self._send(400, json.dumps(
{"error": "Aktuelles Passwort ist falsch"}),
"application/json; charset=utf-8")
elif path in ("/api/reset", "/api/reset/"):
self.do_reset()
else:
uid = self._require_user()
if uid is None:
return
prefix = "/api/sessions/"
if path in ("/api/sessions", "/api/sessions/"):
sess = create_session(uid, self._read_json())
self._send_json(200, sess)
elif path.startswith(prefix) and path.endswith("/autofill"):
rid = path[len(prefix):].rsplit("/", 1)[0]
if not rid.isdigit():
self._send(400, json.dumps({"error": "ungueltige id"}),
"application/json; charset=utf-8")
return
sess = autofill_session(uid, int(rid))
if sess is None:
self._send(404, json.dumps(
{"error": "session nicht gefunden"}),
"application/json; charset=utf-8")
else:
self._send_json(200, sess)
elif path in ("/api/users", "/api/users/"):
self.require_owner(uid, lambda: self.do_create_user())
elif path in ("/api/import", "/api/import/"):
self.do_import(uid)
else:
self._send(404, "nicht gefunden", "text/plain; charset=utf-8")
except ValueError as exc:
self._send(400, json.dumps({"error": str(exc)}),
"application/json; charset=utf-8")
except Exception as exc:
print(f"Fehler in {path}: {exc}", file=sys.stderr)
self._send(500, json.dumps({"error": str(exc)}),
"application/json; charset=utf-8")
def do_reset(self):
uid = self._require_user()
if uid is None:
return
data = self._read_json()
if uid != OPEN_USER:
# Mehrbenutzermodus: nur der Owner, mit Passwort
if uid != 1:
self._send(403, json.dumps({"error": "nur der erste Benutzer"}),
"application/json; charset=utf-8")
return
con = db()
row = con.execute("SELECT * FROM users WHERE id=?",
(uid,)).fetchone()
con.close()
if not row or not _verify_pw(data.get("password", ""),
row["salt"], row["password_hash"]):
self._send(400, json.dumps(
{"error": "Aktuelles Passwort ist falsch"}),
"application/json; charset=utf-8")
return
# Einzelmodus (offen): kein Passwort noetig, nur Bestaetigungstext
if _clean_str(data.get("confirm")) != "RESET":
self._send(400, json.dumps(
{"error": "Bestätigungstext ist falsch"}),
"application/json; charset=utf-8")
return
reset_server()
self._send_json(200, {"ok": True})
def do_register(self):
if user_count() > 0:
self._send(403, json.dumps(
{"error": "Registrierung geschlossen, es existiert bereits ein Benutzer"}),
"application/json; charset=utf-8")
return
data = self._read_json()
uid, username, token = create_user(
data.get("username", ""), data.get("password", ""))
self._send_json(200, {"token": token, "username": username, "id": uid})
def do_login(self):
data = self._read_json()
result = login_user(data.get("username", ""), data.get("password", ""))
if result is None:
self._send(401, json.dumps({"error": "Benutzername oder Passwort falsch"}),
"application/json; charset=utf-8")
return
uid, username, token = result
self._send_json(200, {"token": token, "username": username, "id": uid})
def do_create_user(self):
data = self._read_json()
uid, username, _ = create_user(data.get("username", ""),
data.get("password", ""))
self._send_json(200, {"id": uid, "username": username})
# -- PATCH / PUT ------------------------------------------------------
def do_PATCH(self):
path = urlparse(self.path).path
try:
uid = self._require_user()
if uid is None:
return
prefix = "/api/sessions/"
if path.startswith(prefix):
rid = path[len(prefix):].split("/")[0]
if not rid.isdigit():
self._send(400, json.dumps({"error": "ungueltige id"}),
"application/json; charset=utf-8")
return
sess = update_session(uid, int(rid), self._read_json())
if sess is None:
self._send(404, json.dumps(
{"error": "session nicht gefunden"}),
"application/json; charset=utf-8")
else:
self._send_json(200, sess)
else:
self._send(404, "nicht gefunden", "text/plain; charset=utf-8")
except Exception as exc:
print(f"Fehler in {path}: {exc}", file=sys.stderr)
self._send(500, json.dumps({"error": str(exc)}),
"application/json; charset=utf-8")
def do_PUT(self):
path = urlparse(self.path).path
try:
if path in ("/api/settings", "/api/settings/"):
uid = self._require_user()
if uid is None:
return
self._send_json(200, update_settings(uid, self._read_json()))
elif path.startswith("/api/users/") and path.endswith("/password"):
uid = self._require_user()
if uid is None:
return
if uid != 1:
self._send(403, json.dumps(
{"error": "nur der erste Benutzer"}),
"application/json; charset=utf-8")
return
rid = path[len("/api/users/"):].rsplit("/", 1)[0]
if not rid.isdigit():
self._send(400, json.dumps({"error": "ungueltige id"}),
"application/json; charset=utf-8")
return
if set_user_password(int(rid),
self._read_json().get("password", "")):
self._send_json(200, {"ok": True})
else:
self._send(404, json.dumps({"error": "benutzer nicht gefunden"}),
"application/json; charset=utf-8")
else:
self._send(404, "nicht gefunden", "text/plain; charset=utf-8")
except ValueError as exc:
self._send(400, json.dumps({"error": str(exc)}),
"application/json; charset=utf-8")
except Exception as exc:
print(f"Fehler in {path}: {exc}", file=sys.stderr)
self._send(500, json.dumps({"error": str(exc)}),
"application/json; charset=utf-8")
# -- DELETE ------------------------------------------------------------
def do_DELETE(self):
path = urlparse(self.path).path
try:
uid = self._require_user()
if uid is None:
return
prefix = "/api/sessions/"
if path.startswith(prefix):
rid = path[len(prefix):].split("/")[0]
if not rid.isdigit():
self._send(400, json.dumps({"error": "ungueltige id"}),
"application/json; charset=utf-8")
return
deleted = delete_session(uid, int(rid))
self._send_json(200, {"ok": True, "deleted": deleted})
elif path.startswith("/api/users/"):
if uid != 1:
self._send(403, json.dumps(
{"error": "nur der erste Benutzer"}),
"application/json; charset=utf-8")
return
rid = path[len("/api/users/"):].split("/")[0]
if not rid.isdigit() or int(rid) == uid:
self._send(400, json.dumps(
{"error": "ungueltige id oder eigener Account"}),
"application/json; charset=utf-8")
return
deleted = delete_user(int(rid))
self._send_json(200, {"ok": True, "deleted": deleted})
else:
self._send(404, "nicht gefunden", "text/plain; charset=utf-8")
except Exception as exc:
print(f"Fehler in {path}: {exc}", file=sys.stderr)
self._send(500, json.dumps({"error": str(exc)}),
"application/json; charset=utf-8")
def do_import(self, uid):
data = self._read_json()
fmt = data.get("format", "")
raw = data.get("data", "")
try:
if fmt == EXPORT_FORMAT:
doc = json.loads(raw)
created, updated = import_json_doc(uid, doc)
elif fmt == "csv":
created, updated = import_csv_text(uid, raw)
else:
self._send(400, json.dumps(
{"error": f"unbekanntes format: {fmt}"}),
"application/json; charset=utf-8")
return
self._send_json(200, {"ok": True, "created": created,
"updated": updated})
except Exception as exc:
self._send(400, json.dumps({"error": str(exc)}),
"application/json; charset=utf-8")
def main():
port = int(os.environ.get("PORT", "8080"))
if len(sys.argv) > 1 and sys.argv[1].isdigit():
port = int(sys.argv[1])
load_config()
init_db()
print(f"Trainingsplan auf Port {port}, DB: {DB}")
ThreadingHTTPServer(("0.0.0.0", port), Handler).serve_forever()
if __name__ == "__main__":
main()
Reference in new issue View Git Blame Copy Permalink