dbks/p42.f-w
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

P42 Dashboard: erster Stand

Browse Source 
FastAPI-Backend und React-Frontend mit Konten, Sendungsverfolgung,
Projekt-Artikeln, Vorrat & Einkauf mit Rezeptabgleich, Reise-Weltkarte,
Kanban-Projekten mit Priorität, Server-Monitoring und den Designs
Matrix und Blackwall. Betrieb per Docker hinter nginx (siehe DEPLOY.md).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
guyg1474andClaude Opus 5.5 committed 2026-10-14 14:08:11 +02:00
commit dd58b3d1f8
75 files changed
+11418
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

backend/app/__init__.py
View File
Whitespace-only changes.
backend/app/__main__.py
+14
View File
@@ -0,0 +1,14 @@
"""Start mit `python -m app` (im Ordner backend/)."""
import os
import uvicorn
from .config import PORT
if __name__ == "__main__":
uvicorn.run(
"app.main:app",
host=os.environ.get("HOST", "0.0.0.0"),
port=int(os.environ.get("PORT", PORT)),
reload=os.environ.get("RELOAD") == "1",
)
backend/app/auth.py
+288
View File
@@ -0,0 +1,288 @@
"""Konten, Anmeldung (Sitzungs-Cookie) und Benutzerverwaltung."""
import hashlib
import json
import re
import secrets
import time
from collections import defaultdict
from fastapi import APIRouter, Body, Depends, HTTPException, Request, Response
from .config import LINKS as CONFIG_LINKS
from .db import get_meta, now_iso, transaction
router = APIRouter()
SESSION_COOKIE = "p42_session"
SESSION_SECONDS = 30 * 24 * 60 * 60
USERNAME = re.compile(r"[A-Za-z0-9._-]{3,32}")
MIN_PASSWORD = 8
# ---------- Passwörter ----------
_SCRYPT = {"n": 2**14, "r": 8, "p": 1, "dklen": 32}
def hash_password(password: str) -> str:
salt = secrets.token_bytes(16)
digest = hashlib.scrypt(password.encode(), salt=salt, **_SCRYPT)
return f"scrypt${salt.hex()}${digest.hex()}"
def verify_password(password: str, stored: str) -> bool:
_, salt, digest = stored.split("$")
calc = hashlib.scrypt(password.encode(), salt=bytes.fromhex(salt), **_SCRYPT)
return secrets.compare_digest(calc.hex(), digest)
# Für unbekannte Benutzernamen trotzdem rechnen, damit die Antwortzeit nichts verrät
_DUMMY_HASH = hash_password(secrets.token_hex(8))
# ---------- Hilfsfunktionen ----------
def user_dict(row) -> dict:
return {
"id": row["id"],
"username": row["username"],
"displayName": row["display_name"],
"isAdmin": bool(row["is_admin"]),
"createdAt": row["created_at"],
}
def _hash_token(token: str) -> str:
return hashlib.sha256(token.encode()).hexdigest()
def _username(value) -> str:
v = value.strip() if isinstance(value, str) else ""
if not USERNAME.fullmatch(v):
raise HTTPException(400, "Benutzername: 3–32 Zeichen, nur Buchstaben, Zahlen, Punkt, Minus und Unterstrich.")
return v
def _display_name(value, fallback: str = "") -> str:
v = value.strip() if isinstance(value, str) else ""
v = v or fallback
if not v or len(v) > 60:
raise HTTPException(400, "Anzeigename fehlt oder ist zu lang (max. 60 Zeichen).")
return v
def _password(value) -> str:
if not isinstance(value, str) or len(value) < MIN_PASSWORD:
raise HTTPException(400, f"Das Passwort muss mindestens {MIN_PASSWORD} Zeichen lang sein.")
if len(value) > 200:
raise HTTPException(400, "Das Passwort ist zu lang.")
return value
def _start_session(response: Response, request: Request, user_id: int) -> None:
token = secrets.token_urlsafe(32)
with transaction() as conn:
conn.execute("DELETE FROM sessions WHERE expires_at < ?", (time.time(),))
conn.execute(
"INSERT INTO sessions (token_hash, user_id, expires_at) VALUES (?, ?, ?)",
(_hash_token(token), user_id, time.time() + SESSION_SECONDS),
)
https = request.headers.get("x-forwarded-proto", request.url.scheme) == "https"
response.set_cookie(
SESSION_COOKIE, token, max_age=SESSION_SECONDS, httponly=True, samesite="lax", secure=https, path="/",
)
def _create_user(conn, username: str, display_name: str, password: str, is_admin: bool) -> int:
if conn.execute("SELECT 1 FROM users WHERE username = ?", (username,)).fetchone():
raise HTTPException(400, "Diesen Benutzernamen gibt es schon.")
cur = conn.execute(
"INSERT INTO users (username, display_name, password_hash, is_admin, created_at) VALUES (?, ?, ?, ?, ?)",
(username, display_name, hash_password(password), int(is_admin), now_iso()),
)
return cur.lastrowid
# ---------- Abhängigkeiten für geschützte Routen ----------
def current_user(request: Request) -> dict:
token = request.cookies.get(SESSION_COOKIE)
if token:
with transaction() as conn:
row = conn.execute(
"""SELECT u.* FROM sessions s JOIN users u ON u.id = s.user_id
WHERE s.token_hash = ? AND s.expires_at > ?""",
(_hash_token(token), time.time()),
).fetchone()
if row:
return user_dict(row)
raise HTTPException(401, "Nicht angemeldet.")
def require_admin(user: dict = Depends(current_user)) -> dict:
if not user["isAdmin"]:
raise HTTPException(403, "Nur für Administratoren.")
return user
# ---------- Anmelde-Schutz: max. 5 Fehlversuche in 10 Minuten ----------
_failures: dict[tuple, list[float]] = defaultdict(list)
MAX_FAILURES, WINDOW = 5, 600
def _check_rate_limit(key: tuple) -> None:
now = time.time()
_failures[key] = [t for t in _failures[key] if now - t < WINDOW]
if len(_failures[key]) >= MAX_FAILURES:
raise HTTPException(429, "Zu viele Fehlversuche. Bitte in ein paar Minuten erneut versuchen.")
# ---------- Routen: Anmeldung ----------
@router.get("/api/auth/status")
def status(request: Request):
with transaction() as conn:
has_users = conn.execute("SELECT 1 FROM users LIMIT 1").fetchone() is not None
try:
user = current_user(request)
except HTTPException:
user = None
return {"setupRequired": not has_users, "user": user}
@router.post("/api/auth/setup")
def setup(request: Request, response: Response, data: dict = Body(...)):
"""Erstes Konto anlegen – wird Administrator und übernimmt die bisherigen Daten."""
with transaction() as conn:
if conn.execute("SELECT 1 FROM users LIMIT 1").fetchone():
raise HTTPException(403, "Es gibt bereits ein Konto.")
username = _username(data.get("username"))
display_name = _display_name(data.get("displayName"), username)
password = _password(data.get("password"))
with transaction() as conn:
if conn.execute("SELECT 1 FROM users LIMIT 1").fetchone():
raise HTTPException(403, "Es gibt bereits ein Konto.")
user_id = _create_user(conn, username, display_name, password, is_admin=True)
# Bisherige Projekte ohne Autor diesem Konto zuordnen
conn.execute("UPDATE projects SET created_by = ? WHERE created_by IS NULL", (user_id,))
conn.execute("UPDATE projects SET updated_by = ? WHERE updated_by IS NULL", (user_id,))
# Bisherige Links (links.json bzw. config.json) übernehmen
legacy = get_meta(conn, "legacy_links")
links = json.loads(legacy) if legacy else CONFIG_LINKS
for key, url in (links or {}).items():
if url:
conn.execute("INSERT OR REPLACE INTO user_links VALUES (?, ?, ?)", (user_id, key, url))
conn.execute("DELETE FROM meta WHERE key = 'legacy_links'")
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
_start_session(response, request, user_id)
return user_dict(row)
@router.post("/api/auth/login")
def login(request: Request, response: Response, data: dict = Body(...)):
username = str(data.get("username") or "").strip()
password = str(data.get("password") or "")
key = (request.client.host if request.client else "", username.lower())
_check_rate_limit(key)
with transaction() as conn:
row = conn.execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
if not verify_password(password, row["password_hash"] if row else _DUMMY_HASH) or not row:
_failures[key].append(time.time())
raise HTTPException(401, "Benutzername oder Passwort falsch.")
_failures.pop(key, None)
_start_session(response, request, row["id"])
return user_dict(row)
@router.post("/api/auth/logout")
def logout(request: Request, response: Response):
token = request.cookies.get(SESSION_COOKIE)
if token:
with transaction() as conn:
conn.execute("DELETE FROM sessions WHERE token_hash = ?", (_hash_token(token),))
response.delete_cookie(SESSION_COOKIE, path="/")
return {"ok": True}
@router.patch("/api/auth/me")
def update_me(request: Request, data: dict = Body(...), user: dict = Depends(current_user)):
"""Eigenen Anzeigenamen oder eigenes Passwort ändern."""
with transaction() as conn:
if "displayName" in data:
conn.execute("UPDATE users SET display_name = ? WHERE id = ?", (_display_name(data["displayName"]), user["id"]))
if data.get("newPassword"):
row = conn.execute("SELECT password_hash FROM users WHERE id = ?", (user["id"],)).fetchone()
if not verify_password(str(data.get("currentPassword") or ""), row["password_hash"]):
raise HTTPException(400, "Das aktuelle Passwort stimmt nicht.")
conn.execute("UPDATE users SET password_hash = ? WHERE id = ?", (hash_password(_password(data["newPassword"])), user["id"]))
# Andere Sitzungen (z. B. auf anderen Geräten) beenden
current = _hash_token(request.cookies.get(SESSION_COOKIE, ""))
conn.execute("DELETE FROM sessions WHERE user_id = ? AND token_hash != ?", (user["id"], current))
row = conn.execute("SELECT * FROM users WHERE id = ?", (user["id"],)).fetchone()
return user_dict(row)
# ---------- Routen: Benutzerverwaltung (nur Admins) ----------
@router.get("/api/users")
def list_users(_: dict = Depends(require_admin)):
with transaction() as conn:
rows = conn.execute("SELECT * FROM users ORDER BY display_name COLLATE NOCASE").fetchall()
return [user_dict(r) for r in rows]
@router.post("/api/users", status_code=201)
def create_user(data: dict = Body(...), _: dict = Depends(require_admin)):
username = _username(data.get("username"))
display_name = _display_name(data.get("displayName"), username)
password = _password(data.get("password"))
with transaction() as conn:
user_id = _create_user(conn, username, display_name, password, bool(data.get("isAdmin")))
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
return user_dict(row)
def _ensure_other_admin(conn, user_id: int) -> None:
others = conn.execute("SELECT COUNT(*) FROM users WHERE is_admin = 1 AND id != ?", (user_id,)).fetchone()[0]
if not others:
raise HTTPException(400, "Es muss mindestens ein Administrator übrig bleiben.")
@router.patch("/api/users/{user_id}")
def update_user(user_id: int, data: dict = Body(...), _: dict = Depends(require_admin)):
with transaction() as conn:
if not conn.execute("SELECT 1 FROM users WHERE id = ?", (user_id,)).fetchone():
raise HTTPException(404, "Nicht gefunden.")
if "username" in data:
username = _username(data["username"])
if conn.execute("SELECT 1 FROM users WHERE username = ? AND id != ?", (username, user_id)).fetchone():
raise HTTPException(400, "Diesen Benutzernamen gibt es schon.")
conn.execute("UPDATE users SET username = ? WHERE id = ?", (username, user_id))
if "displayName" in data:
conn.execute("UPDATE users SET display_name = ? WHERE id = ?", (_display_name(data["displayName"]), user_id))
if "isAdmin" in data:
if not data["isAdmin"]:
_ensure_other_admin(conn, user_id)
conn.execute("UPDATE users SET is_admin = ? WHERE id = ?", (int(bool(data["isAdmin"])), user_id))
if data.get("password"):
conn.execute("UPDATE users SET password_hash = ? WHERE id = ?", (hash_password(_password(data["password"])), user_id))
conn.execute("DELETE FROM sessions WHERE user_id = ?", (user_id,))
row = conn.execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
return user_dict(row)
@router.delete("/api/users/{user_id}")
def delete_user(user_id: int, admin: dict = Depends(require_admin)):
if user_id == admin["id"]:
raise HTTPException(400, "Das eigene Konto kann nicht gelöscht werden.")
with transaction() as conn:
row = conn.execute("SELECT is_admin FROM users WHERE id = ?", (user_id,)).fetchone()
if not row:
raise HTTPException(404, "Nicht gefunden.")
if row["is_admin"]:
_ensure_other_admin(conn, user_id)
conn.execute("DELETE FROM users WHERE id = ?", (user_id,))
return {"ok": True}
backend/app/config.py
+24
View File
@@ -0,0 +1,24 @@
"""Lädt config.json (oder config.example.json) aus dem Projektordner."""
import json
import os
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
DATA_DIR = Path(os.environ.get("P42_DATA_DIR") or ROOT / "data")
FRONTEND_DIST = ROOT / "frontend" / "dist"
def _load() -> dict:
path = ROOT / "config.json"
if not path.exists():
path = ROOT / "config.example.json"
return json.loads(path.read_text(encoding="utf-8"))
config = _load()
PORT = int(config.get("port", 3000))
LINKS = config.get("links", {})
TRACKING = config.get("tracking", {})
# Wikimedia verlangt eine Kontaktangabe im User-Agent; leer = keine Wikipedia-Texte
WIKIPEDIA_CONTACT = (config.get("reisen") or {}).get("wikipediaKontakt", "").strip()
backend/app/db.py
+337
View File
@@ -0,0 +1,337 @@
"""SQLite-Datenbank (data/p42.db): Schema, Verbindungen und Übernahme der alten JSON-Daten."""
import html
import json
import re
import sqlite3
from contextlib import contextmanager
from datetime import datetime, timezone
from .config import DATA_DIR
DB_FILE = DATA_DIR / "p42.db"
# Geteilte Vorrats- und Einkaufsliste; key = Name in casefold() für den Vergleich.
# Pro Produkt, Liste und Einheit eine Zeile ("Mehl 1 kg" und "Mehl 2 Stück" dürfen nebeneinander stehen).
HOUSEHOLD_TABLE = """(
id INTEGER PRIMARY KEY,
key TEXT NOT NULL,
name TEXT NOT NULL,
list TEXT NOT NULL CHECK (list IN ('vorrat', 'einkauf')),
quantity REAL NOT NULL CHECK (quantity > 0),
unit TEXT NOT NULL DEFAULT 'stk' CHECK (unit IN ('stk', 'kg', 'l')),
updated_at TEXT NOT NULL,
updated_by INTEGER REFERENCES users(id) ON DELETE SET NULL,
UNIQUE (key, list, unit)
)"""
# Kanban-Projekte (je ein Board), gemeinsam für alle Konten
KANBAN_PROJECTS_TABLE = """(
id INTEGER PRIMARY KEY,
name TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
position REAL NOT NULL,
completed_at TEXT,
created_at TEXT NOT NULL,
created_by INTEGER REFERENCES users(id) ON DELETE SET NULL,
updated_at TEXT NOT NULL,
updated_by INTEGER REFERENCES users(id) ON DELETE SET NULL
)"""
# Aufgaben eines Kanban-Projekts; position sortiert innerhalb einer Spalte
KANBAN_TASKS_TABLE = """(
id INTEGER PRIMARY KEY,
project_id INTEGER NOT NULL REFERENCES kanban_projects(id) ON DELETE CASCADE,
title TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL CHECK (status IN ('offen', 'arbeit', 'wartet', 'erledigt')),
assignee INTEGER REFERENCES users(id) ON DELETE SET NULL,
priority INTEGER NOT NULL DEFAULT 2 CHECK (priority IN (1, 2, 3)), -- 1 hoch, 2 mittel, 3 niedrig
position REAL NOT NULL,
created_at TEXT NOT NULL,
created_by INTEGER REFERENCES users(id) ON DELETE SET NULL,
updated_at TEXT NOT NULL,
updated_by INTEGER REFERENCES users(id) ON DELETE SET NULL
)"""
SCHEMA = f"""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
username TEXT NOT NULL UNIQUE COLLATE NOCASE,
display_name TEXT NOT NULL,
password_hash TEXT NOT NULL,
is_admin INTEGER NOT NULL DEFAULT 0,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS sessions (
token_hash TEXT PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
expires_at REAL NOT NULL
);
CREATE TABLE IF NOT EXISTS user_links (
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
key TEXT NOT NULL,
url TEXT NOT NULL,
PRIMARY KEY (user_id, key)
);
CREATE TABLE IF NOT EXISTS projects (
id TEXT PRIMARY KEY,
title TEXT NOT NULL,
description TEXT NOT NULL DEFAULT '',
status TEXT NOT NULL,
cover TEXT,
content TEXT NOT NULL DEFAULT '',
created_at TEXT NOT NULL,
created_by INTEGER REFERENCES users(id) ON DELETE SET NULL,
updated_at TEXT NOT NULL,
updated_by INTEGER REFERENCES users(id) ON DELETE SET NULL
);
CREATE TABLE IF NOT EXISTS monitors (
id INTEGER PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
name TEXT NOT NULL,
type TEXT NOT NULL,
target TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS household_items {HOUSEHOLD_TABLE};
-- Produktkatalog für die Vorschläge
CREATE TABLE IF NOT EXISTS products (
key TEXT PRIMARY KEY,
name TEXT NOT NULL,
uses INTEGER NOT NULL DEFAULT 0,
last_used TEXT NOT NULL
);
-- Gelernte Zuordnungen Rezeptzutat → Produkt (z. B. "margarine" → "Butter")
CREATE TABLE IF NOT EXISTS ingredient_aliases (
key TEXT PRIMARY KEY,
product TEXT NOT NULL
);
-- Reise-Karte pro Konto: besuchte Städte (GeoNames-ID) und von Hand markierte Länder (ISO-Code)
CREATE TABLE IF NOT EXISTS travel_cities (
id INTEGER PRIMARY KEY,
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
geoname_id INTEGER NOT NULL,
name TEXT NOT NULL,
admin1 TEXT NOT NULL DEFAULT '',
country TEXT NOT NULL,
lat REAL NOT NULL,
lon REAL NOT NULL,
population INTEGER,
elevation REAL,
timezone TEXT,
created_at TEXT NOT NULL,
UNIQUE (user_id, geoname_id)
);
CREATE TABLE IF NOT EXISTS travel_countries (
user_id INTEGER NOT NULL REFERENCES users(id) ON DELETE CASCADE,
code TEXT NOT NULL,
PRIMARY KEY (user_id, code)
);
-- Zwischenspeicher für Infos von Wikipedia/Wikidata
CREATE TABLE IF NOT EXISTS info_cache (
key TEXT PRIMARY KEY,
value TEXT NOT NULL,
fetched_at REAL NOT NULL
);
-- Kanban: Projekte (je ein Board) → Aufgaben; gemeinsam für alle Konten
CREATE TABLE IF NOT EXISTS kanban_projects {KANBAN_PROJECTS_TABLE};
-- Verknüpfung Kanban-Projekt ↔ Artikel aus "P42 Projekte"
CREATE TABLE IF NOT EXISTS kanban_project_links (
project_id INTEGER NOT NULL REFERENCES kanban_projects(id) ON DELETE CASCADE,
article_id TEXT NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
PRIMARY KEY (project_id, article_id)
);
CREATE TABLE IF NOT EXISTS kanban_tasks {KANBAN_TASKS_TABLE};
CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value TEXT NOT NULL
);
"""
def now_iso() -> str:
return datetime.now(timezone.utc).isoformat(timespec="milliseconds").replace("+00:00", "Z")
def connect() -> sqlite3.Connection:
conn = sqlite3.connect(DB_FILE, timeout=10)
conn.row_factory = sqlite3.Row
conn.execute("PRAGMA foreign_keys = ON")
return conn
@contextmanager
def transaction():
"""Verbindung für einen Arbeitsschritt; bei Erfolg wird gespeichert, bei Fehlern zurückgerollt."""
conn = connect()
try:
with conn:
yield conn
finally:
conn.close()
def get_meta(conn, key: str):
row = conn.execute("SELECT value FROM meta WHERE key = ?", (key,)).fetchone()
return row["value"] if row else None
# ---------- Übernahme der alten JSON-Dateien ----------
def _posts_to_html(posts: list) -> str:
"""Ganz alte Daten mit Update-Feed in Artikeltext umwandeln."""
parts = []
for post in reversed(posts):
date = datetime.fromisoformat(post["createdAt"].replace("Z", "+00:00")).strftime("%d.%m.%Y")
parts.append(f"<h3>{date}</h3>")
for para in filter(None, re.split(r"\n{2,}", post.get("text") or "")):
parts.append(f"<p>{html.escape(para).replace(chr(10), '<br>')}</p>")
parts += [f'<p><img src="/uploads/{img}" alt=""></p>' for img in post.get("images", [])]
return "".join(parts)
def _import_legacy(conn) -> None:
projects_json = DATA_DIR / "projects.json"
if projects_json.exists():
for p in json.loads(projects_json.read_text(encoding="utf-8")).get("projects", []):
conn.execute(
"""INSERT OR IGNORE INTO projects
(id, title, description, status, cover, content, created_at, updated_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
(
p["id"], p["title"], p.get("description", ""), p.get("status", "aktiv"), p.get("cover"),
p.get("content") or _posts_to_html(p.get("posts", [])), p["createdAt"], p["updatedAt"],
),
)
projects_json.rename(projects_json.with_name("projects.json.importiert"))
# Links aus der Zeit vor den Konten bekommt das erste Konto
links_json = DATA_DIR / "links.json"
if links_json.exists():
conn.execute("INSERT OR REPLACE INTO meta VALUES ('legacy_links', ?)", (links_json.read_text(encoding="utf-8"),))
links_json.rename(links_json.with_name("links.json.importiert"))
def _migrate(conn) -> None:
# Vorrat/Einkauf aus der Zeit vor den Einheiten: Tabelle mit Einheit und Kommazahlen neu aufbauen
columns = {r["name"] for r in conn.execute("PRAGMA table_info(household_items)")}
if "unit" in columns:
return
conn.execute("ALTER TABLE household_items RENAME TO household_items_alt")
conn.execute("CREATE TABLE household_items " + HOUSEHOLD_TABLE)
conn.execute(
"""INSERT INTO household_items (id, key, name, list, quantity, unit, updated_at, updated_by)
SELECT id, key, name, list, quantity, 'stk', updated_at, updated_by FROM household_items_alt"""
)
conn.execute("DROP TABLE household_items_alt")
def _migrate_travel(conn) -> None:
# Reise-Städte aus einer Vorversion ohne Höhe/Zeitzone: Spalten ergänzen
columns = {r["name"] for r in conn.execute("PRAGMA table_info(travel_cities)")}
for column, kind in (("elevation", "REAL"), ("timezone", "TEXT")):
if column not in columns:
conn.execute(f"ALTER TABLE travel_cities ADD COLUMN {column} {kind}")
def _migrate_kanban_categories(conn) -> None:
"""Kategorien wieder entfernt: kanban_projects ohne category_id neu aufbauen.
Aufgaben und Artikel-Links verweisen per ON DELETE CASCADE auf kanban_projects. Ein normales
DROP würde sie mitlöschen, deshalb das SQLite-Verfahren: Fremdschlüssel aus, neue Tabelle
anlegen, kopieren, alte löschen, umbenennen, Verweise prüfen.
"""
columns = {r["name"] for r in conn.execute("PRAGMA table_info(kanban_projects)")}
if "category_id" not in columns:
conn.execute("DROP TABLE IF EXISTS kanban_categories")
return
conn.commit()
conn.execute("PRAGMA foreign_keys = OFF") # wirkt nur außerhalb einer Transaktion
try:
conn.execute("BEGIN")
conn.execute("CREATE TABLE kanban_projects_neu " + KANBAN_PROJECTS_TABLE)
conn.execute(
"""INSERT INTO kanban_projects_neu (id, name, description, position, created_at, created_by, updated_at, updated_by)
SELECT id, name, description, position, created_at, created_by, updated_at, updated_by FROM kanban_projects"""
)
conn.execute("DROP TABLE kanban_projects")
conn.execute("ALTER TABLE kanban_projects_neu RENAME TO kanban_projects")
conn.execute("DROP TABLE kanban_categories")
if conn.execute("PRAGMA foreign_key_check").fetchall():
raise RuntimeError("Kanban-Migration: ungültige Verweise, Änderungen werden zurückgenommen.")
conn.commit()
except Exception:
conn.rollback()
raise
finally:
conn.execute("PRAGMA foreign_keys = ON")
def _migrate_kanban_priority(conn) -> None:
# Aufgaben ohne Priorität: Spalte ergänzen, bestehende Aufgaben werden "mittel"
columns = {r["name"] for r in conn.execute("PRAGMA table_info(kanban_tasks)")}
if "priority" not in columns:
conn.execute(
"ALTER TABLE kanban_tasks ADD COLUMN priority INTEGER NOT NULL DEFAULT 2 CHECK (priority IN (1, 2, 3))"
)
def _migrate_kanban_completed(conn) -> None:
# Projekte "abgeschlossen": Spalte ergänzen, falls sie fehlt
columns = {r["name"] for r in conn.execute("PRAGMA table_info(kanban_projects)")}
if "completed_at" not in columns:
conn.execute("ALTER TABLE kanban_projects ADD COLUMN completed_at TEXT")
def _migrate_kanban(conn) -> None:
# Aufgaben aus der Zeit vor den Projekten: in "Allgemeines Board" übernehmen
columns = {r["name"] for r in conn.execute("PRAGMA table_info(kanban_tasks)")}
if "project_id" in columns:
return
conn.execute("ALTER TABLE kanban_tasks RENAME TO kanban_tasks_alt")
conn.execute("CREATE TABLE kanban_tasks " + KANBAN_TASKS_TABLE)
if conn.execute("SELECT COUNT(*) FROM kanban_tasks_alt").fetchone()[0]:
now = now_iso()
project = conn.execute(
"INSERT INTO kanban_projects (name, position, created_at, updated_at) VALUES ('Allgemeines Board', 1, ?, ?)",
(now, now),
).lastrowid
conn.execute(
"""INSERT INTO kanban_tasks (id, project_id, title, description, status, assignee, position,
created_at, created_by, updated_at, updated_by)
SELECT id, ?, title, description, status, assignee, position, created_at, created_by, updated_at, updated_by
FROM kanban_tasks_alt""",
(project,),
)
conn.execute("DROP TABLE kanban_tasks_alt")
def init() -> None:
DATA_DIR.mkdir(parents=True, exist_ok=True)
with transaction() as conn:
conn.execute("PRAGMA journal_mode = WAL")
conn.executescript(SCHEMA)
_migrate(conn)
_migrate_travel(conn)
_migrate_kanban_categories(conn)
_migrate_kanban_completed(conn)
_migrate_kanban(conn)
_migrate_kanban_priority(conn)
_import_legacy(conn)
init()
backend/app/haushalt.py
+205
View File
@@ -0,0 +1,205 @@
"""Geteilte Vorrats- und Einkaufsliste mit Produktkatalog für einheitliche Namen.
Alle Konten sehen und bearbeiten dieselben Listen. Jedes Produkt steht pro Liste und Einheit
(Anzahl, kg, l) nur einmal; erneutes Hinzufügen in derselben Einheit erhöht die Menge. Neue Produktnamen landen im Katalog und werden als
Vorschläge angeboten – "milch" wird dabei automatisch zum bereits bekannten "Milch".
"""
import math
import re
from fastapi import APIRouter, Body, Depends, HTTPException
from .auth import current_user
from .db import now_iso, transaction
from .links import ValidationError
router = APIRouter()
LISTS = ("vorrat", "einkauf")
UNITS = ("stk", "kg", "l")
MAX_NAME = 60
MAX_QUANTITY = 999
MAX_PRODUCTS = 2000
def _item(row) -> dict:
return {
"id": row["id"],
"name": row["name"],
"list": row["list"],
"quantity": _out_quantity(row["quantity"], row["unit"]),
"unit": row["unit"],
"updatedAt": row["updated_at"],
"updatedBy": {"id": row["updated_by"], "name": row["updated_by_name"]} if row["updated_by"] else None,
}
def _out_quantity(value, unit: str):
return int(value) if unit == "stk" else round(value, 2)
def _unit(value) -> str:
if value is None:
return "stk"
if value not in UNITS:
raise ValidationError("Unbekannte Einheit.")
return value
def _quantity(value, unit: str = "stk"):
"""Anzahl: ganze Zahl ab 1. kg/l: Kommazahl ab 0,01 (zwei Nachkommastellen)."""
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
raise ValidationError("Menge fehlt.")
if unit == "stk":
if value != int(value) or not 1 <= value <= MAX_QUANTITY:
raise ValidationError(f"Anzahl muss eine ganze Zahl zwischen 1 und {MAX_QUANTITY} sein.")
return int(value)
value = round(value, 2)
if not 0 < value <= MAX_QUANTITY:
raise ValidationError(f"Menge muss zwischen 0,01 und {MAX_QUANTITY} liegen.")
return value
def _clean_name(value) -> str:
name = re.sub(r"\s+", " ", value).strip() if isinstance(value, str) else ""
if not name:
raise ValidationError("Produktname fehlt.")
if len(name) > MAX_NAME:
raise ValidationError(f"Produktname ist zu lang (max. {MAX_NAME} Zeichen).")
return name
def _key(name: str) -> str:
# casefold() statt COLLATE NOCASE: SQLite vergleicht nur ASCII ohne Groß/klein ("Äpfel" ≠ "äpfel")
return name.casefold()
_UMLAUTS = str.maketrans({"ä": "a", "ö": "o", "ü": "u", "ß": "ss"})
def _sort_key(name: str) -> str:
# Deutsche Sortierung: "Äpfel" gehört zu A, nicht hinter Z
return _key(name).translate(_UMLAUTS)
def _canonical_name(conn, name: str) -> str:
"""Schreibweise aus dem Katalog übernehmen bzw. neues Produkt aufnehmen."""
row = conn.execute("SELECT name FROM products WHERE key = ?", (_key(name),)).fetchone()
if row:
conn.execute("UPDATE products SET uses = uses + 1, last_used = ? WHERE key = ?", (now_iso(), _key(name)))
return row["name"]
if conn.execute("SELECT COUNT(*) FROM products").fetchone()[0] >= MAX_PRODUCTS:
raise ValidationError("Der Produktkatalog ist voll. Bitte alte Vorschläge löschen.")
# Neue Produkte mit großem Anfangsbuchstaben, damit "milch" und "Milch" nicht doppelt entstehen
name = name[0].upper() + name[1:]
conn.execute("INSERT INTO products (key, name, uses, last_used) VALUES (?, ?, 1, ?)", (_key(name), name, now_iso()))
return name
def _add(conn, name: str, list_: str, quantity, unit: str, user_id: int) -> None:
"""Produkt auf eine Liste setzen; steht es dort schon in derselben Einheit, wird die Menge erhöht."""
conn.execute(
"""INSERT INTO household_items (key, name, list, quantity, unit, updated_at, updated_by)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT (key, list, unit) DO UPDATE SET
quantity = MIN(ROUND(quantity + excluded.quantity, 2), ?),
updated_at = excluded.updated_at,
updated_by = excluded.updated_by""",
(_key(name), name, list_, quantity, unit, now_iso(), user_id, MAX_QUANTITY),
)
def _get_item(conn, item_id: int):
row = conn.execute("SELECT * FROM household_items WHERE id = ?", (item_id,)).fetchone()
if not row:
raise HTTPException(404, "Eintrag nicht gefunden – vielleicht hat ihn gerade jemand anderes geändert.")
return row
def _state(conn) -> dict:
items = conn.execute(
"""SELECT i.*, u.display_name AS updated_by_name FROM household_items i
LEFT JOIN users u ON u.id = i.updated_by"""
).fetchall()
# Häufig genutzte Produkte zuerst vorschlagen
products = conn.execute("SELECT name, uses FROM products").fetchall()
return {
"items": [_item(r) for r in sorted(items, key=lambda r: _sort_key(r["name"]))],
"products": [r["name"] for r in sorted(products, key=lambda r: (-r["uses"], _sort_key(r["name"])))],
}
@router.get("/api/haushalt")
def get_household(_: dict = Depends(current_user)):
with transaction() as conn:
return _state(conn)
@router.post("/api/haushalt/items", status_code=201)
def add_item(data: dict = Body(...), user: dict = Depends(current_user)):
list_ = data.get("list")
if list_ not in LISTS:
raise ValidationError("Unbekannte Liste.")
name = _clean_name(data.get("name"))
unit = _unit(data.get("unit"))
quantity = _quantity(data.get("quantity", 1), unit)
with transaction() as conn:
_add(conn, _canonical_name(conn, name), list_, quantity, unit, user["id"])
return _state(conn)
@router.patch("/api/haushalt/items/{item_id}")
def update_item(item_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
"""Menge und/oder Einheit ändern. Beim Einheitenwechsel wird mit einer bestehenden Zeile zusammengeführt."""
with transaction() as conn:
row = _get_item(conn, item_id)
unit = _unit(data.get("unit", row["unit"]))
if "quantity" in data:
quantity = _quantity(data["quantity"], unit)
else:
# Nur Einheit gewechselt: Zahl bleibt, bei "Anzahl" auf ganze Stück aufrunden
quantity = max(1, math.ceil(row["quantity"])) if unit == "stk" else row["quantity"]
other = conn.execute(
"SELECT id FROM household_items WHERE key = ? AND list = ? AND unit = ? AND id != ?",
(row["key"], row["list"], unit, item_id),
).fetchone()
if other:
conn.execute("DELETE FROM household_items WHERE id = ?", (item_id,))
_add(conn, row["name"], row["list"], quantity, unit, user["id"])
else:
conn.execute(
"UPDATE household_items SET quantity = ?, unit = ?, updated_at = ?, updated_by = ? WHERE id = ?",
(quantity, unit, now_iso(), user["id"], item_id),
)
return _state(conn)
@router.post("/api/haushalt/items/{item_id}/move")
def move_item(item_id: int, user: dict = Depends(current_user)):
"""Einkauf → Vorrat ("gekauft") bzw. Vorrat → Einkauf ("aufgebraucht"), mit Zusammenführen."""
with transaction() as conn:
row = _get_item(conn, item_id)
target = "vorrat" if row["list"] == "einkauf" else "einkauf"
# Aufgebraucht heißt: einmal nachkaufen (1 Stück, 1 kg, 1 l), egal wie viel vorher da war
quantity = row["quantity"] if target == "vorrat" else 1
conn.execute("DELETE FROM household_items WHERE id = ?", (item_id,))
_add(conn, row["name"], target, quantity, row["unit"], user["id"])
return _state(conn)
@router.delete("/api/haushalt/items/{item_id}")
def delete_item(item_id: int, _: dict = Depends(current_user)):
with transaction() as conn:
_get_item(conn, item_id)
conn.execute("DELETE FROM household_items WHERE id = ?", (item_id,))
return _state(conn)
@router.delete("/api/haushalt/produkte")
def delete_product(name: str, _: dict = Depends(current_user)):
"""Vorschlag aus dem Katalog entfernen (z. B. Tippfehler). Einträge auf den Listen bleiben."""
with transaction() as conn:
cur = conn.execute("DELETE FROM products WHERE key = ?", (_key(name),))
if not cur.rowcount:
raise HTTPException(404, "Vorschlag nicht gefunden.")
return _state(conn)
backend/app/kanban.py
+368
View File
@@ -0,0 +1,368 @@
"""Kanban: Projekte (je ein Board) → Aufgaben mit Status, Bearbeiter und Beschreibung.
Alles ist für alle Konten gemeinsam. Löschen dürfen nur, wer etwas angelegt hat, und Administratoren.
Projekte lassen sich als abgeschlossen markieren und mit Artikeln aus "P42 Projekte" verknüpfen.
`position` (Kommazahl) sortiert Projekte und Aufgaben innerhalb einer Spalte.
"""
import math
import re
from fastapi import APIRouter, Body, Depends, HTTPException
from .auth import current_user
from .db import now_iso, transaction
from .links import ValidationError
router = APIRouter()
STATUSES = ("offen", "arbeit", "wartet", "erledigt")
MAX_NAME = 80
MAX_TITLE = 120
MAX_DESCRIPTION = 5000
MAX_TASKS = 1000 # pro Projekt
MAX_PROJECTS = 300
MAX_LINKS = 20
def _person(user_id, name) -> dict | None:
return {"id": user_id, "name": name} if user_id is not None and name else None
def _may_delete(user: dict, created_by) -> bool:
return user["isAdmin"] or created_by == user["id"]
# ---------- Eingaben prüfen ----------
def _name(value, field="Name", max_len=MAX_NAME) -> str:
v = re.sub(r"\s+", " ", value).strip() if isinstance(value, str) else ""
if not v:
raise ValidationError(f"{field} fehlt.")
if len(v) > max_len:
raise ValidationError(f"{field} ist zu lang (max. {max_len} Zeichen).")
return v
def _description(value) -> str:
if value is None:
return ""
if not isinstance(value, str):
raise ValidationError("Ungültige Beschreibung.")
text = value.replace("\r\n", "\n").strip()
if len(text) > MAX_DESCRIPTION:
raise ValidationError(f"Beschreibung ist zu lang (max. {MAX_DESCRIPTION} Zeichen).")
return text
PRIORITIES = {"hoch": 1, "mittel": 2, "niedrig": 3}
PRIORITY_NAMES = {v: k for k, v in PRIORITIES.items()}
def _priority(value) -> int:
if value not in PRIORITIES:
raise ValidationError("Unbekannte Priorität.")
return PRIORITIES[value]
def _status(value) -> str:
if value not in STATUSES:
raise ValidationError("Unbekannter Status.")
return value
def _assignee(conn, value):
if value is None:
return None
if isinstance(value, bool) or not isinstance(value, int) or not conn.execute(
"SELECT 1 FROM users WHERE id = ?", (value,)
).fetchone():
raise ValidationError("Unbekannter Bearbeiter.")
return value
def _position(value) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value):
raise ValidationError("Ungültige Position.")
return float(value)
def _links(conn, value) -> list:
if value is None:
return []
if not isinstance(value, list) or len(value) > MAX_LINKS or not all(isinstance(v, str) for v in value):
raise ValidationError("Ungültige Artikel-Verknüpfung.")
ids = list(dict.fromkeys(value))
found = {r["id"] for r in conn.execute(
f"SELECT id FROM projects WHERE id IN ({','.join('?' * len(ids))})", ids
)} if ids else set()
if len(found) != len(ids):
raise ValidationError("Ein verknüpfter Artikel existiert nicht mehr.")
return ids
def _next(conn, table: str, where: str = "", args: tuple = ()) -> float:
# table/where kommen nur aus diesem Modul, nie aus Anfragen
row = conn.execute(f"SELECT MAX(position) FROM {table} {where}", args).fetchone()
return (row[0] or 0) + 1
# ---------- Ausgabe ----------
def _users(conn) -> list:
rows = conn.execute("SELECT id, display_name FROM users ORDER BY display_name COLLATE NOCASE").fetchall()
return [{"id": u["id"], "name": u["display_name"]} for u in rows]
def _project_links(conn) -> dict:
links = {}
for r in conn.execute(
"""SELECT l.project_id, p.id, p.title FROM kanban_project_links l
JOIN projects p ON p.id = l.article_id ORDER BY p.title COLLATE NOCASE"""
):
links.setdefault(r["project_id"], []).append({"id": r["id"], "title": r["title"]})
return links
def _projects(conn, project_id=None) -> list:
rows = conn.execute(
"""SELECT k.*, cu.display_name AS created_by_name, uu.display_name AS updated_by_name
FROM kanban_projects k
LEFT JOIN users cu ON cu.id = k.created_by
LEFT JOIN users uu ON uu.id = k.updated_by"""
+ (" WHERE k.id = ?" if project_id is not None else "")
+ " ORDER BY k.position",
(project_id,) if project_id is not None else (),
).fetchall()
counts = {}
for r in conn.execute("SELECT project_id, status, COUNT(*) AS n FROM kanban_tasks GROUP BY project_id, status"):
counts.setdefault(r["project_id"], dict.fromkeys(STATUSES, 0))[r["status"]] = r["n"]
links = _project_links(conn)
return [
{
"id": r["id"],
"name": r["name"],
"description": r["description"],
"position": r["position"],
"completedAt": r["completed_at"],
"links": links.get(r["id"], []),
"counts": counts.get(r["id"], dict.fromkeys(STATUSES, 0)),
"createdAt": r["created_at"],
"createdBy": _person(r["created_by"], r["created_by_name"]),
"updatedAt": r["updated_at"],
"updatedBy": _person(r["updated_by"], r["updated_by_name"]),
}
for r in rows
]
def _overview(conn) -> dict:
articles = conn.execute("SELECT id, title FROM projects ORDER BY title COLLATE NOCASE").fetchall()
return {
"projects": _projects(conn),
"articles": [{"id": a["id"], "title": a["title"]} for a in articles],
"users": _users(conn),
}
def _task(row) -> dict:
return {
"id": row["id"],
"projectId": row["project_id"],
"title": row["title"],
"description": row["description"],
"status": row["status"],
"assignee": row["assignee"],
"priority": PRIORITY_NAMES[row["priority"]],
"position": row["position"],
"createdAt": row["created_at"],
"createdBy": _person(row["created_by"], row["created_by_name"]),
"updatedAt": row["updated_at"],
"updatedBy": _person(row["updated_by"], row["updated_by_name"]),
}
def _board(conn, project_id: int) -> dict:
projects = _projects(conn, project_id)
if not projects:
raise HTTPException(404, "Kanban-Projekt nicht gefunden – vielleicht wurde es gerade gelöscht.")
project = projects[0]
tasks = conn.execute(
"""SELECT t.*, cu.display_name AS created_by_name, uu.display_name AS updated_by_name
FROM kanban_tasks t
LEFT JOIN users cu ON cu.id = t.created_by
LEFT JOIN users uu ON uu.id = t.updated_by
WHERE t.project_id = ? ORDER BY t.status, t.priority, t.position""",
(project_id,),
).fetchall()
articles = conn.execute("SELECT id, title FROM projects ORDER BY title COLLATE NOCASE").fetchall()
return {
"project": project,
"tasks": [_task(r) for r in tasks],
"users": _users(conn),
"articles": [{"id": a["id"], "title": a["title"]} for a in articles],
}
def _save_links(conn, project_id: int, links: list) -> None:
conn.execute("DELETE FROM kanban_project_links WHERE project_id = ?", (project_id,))
conn.executemany(
"INSERT INTO kanban_project_links (project_id, article_id) VALUES (?, ?)", [(project_id, a) for a in links]
)
# ---------- Übersicht ----------
@router.get("/api/kanban")
def get_overview(_: dict = Depends(current_user)):
with transaction() as conn:
return _overview(conn)
# ---------- Projekte ----------
@router.post("/api/kanban/projekte", status_code=201)
def create_project(data: dict = Body(...), user: dict = Depends(current_user)):
name = _name(data.get("name"))
description = _description(data.get("description"))
with transaction() as conn:
links = _links(conn, data.get("links"))
if conn.execute("SELECT COUNT(*) FROM kanban_projects").fetchone()[0] >= MAX_PROJECTS:
raise ValidationError(f"Maximal {MAX_PROJECTS} Kanban-Projekte.")
now = now_iso()
project_id = conn.execute(
"""INSERT INTO kanban_projects (name, description, position, created_at, created_by, updated_at, updated_by)
VALUES (?, ?, ?, ?, ?, ?, ?)""",
(name, description, _next(conn, "kanban_projects"), now, user["id"], now, user["id"]),
).lastrowid
_save_links(conn, project_id, links)
return {"id": project_id, **_overview(conn)}
@router.get("/api/kanban/projekte/{project_id}")
def get_board(project_id: int, _: dict = Depends(current_user)):
with transaction() as conn:
return _board(conn, project_id)
@router.patch("/api/kanban/projekte/{project_id}")
def update_project(project_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
with transaction() as conn:
if not conn.execute("SELECT 1 FROM kanban_projects WHERE id = ?", (project_id,)).fetchone():
raise HTTPException(404, "Kanban-Projekt nicht gefunden.")
values = {}
if "name" in data:
values["name"] = _name(data["name"])
if "description" in data:
values["description"] = _description(data["description"])
if "completed" in data:
if not isinstance(data["completed"], bool):
raise ValidationError("Ungültiger Wert für abgeschlossen.")
values["completed_at"] = now_iso() if data["completed"] else None
if "links" in data:
_save_links(conn, project_id, _links(conn, data["links"]))
values.update(updated_at=now_iso(), updated_by=user["id"])
columns = ", ".join(f"{k} = ?" for k in values) # Schlüssel stammen nur aus der Liste oben
conn.execute(f"UPDATE kanban_projects SET {columns} WHERE id = ?", (*values.values(), project_id))
return _board(conn, project_id)
@router.delete("/api/kanban/projekte/{project_id}")
def delete_project(project_id: int, user: dict = Depends(current_user)):
with transaction() as conn:
row = conn.execute("SELECT * FROM kanban_projects WHERE id = ?", (project_id,)).fetchone()
if not row:
raise HTTPException(404, "Kanban-Projekt nicht gefunden.")
if not _may_delete(user, row["created_by"]):
raise HTTPException(403, "Nur wer das Projekt angelegt hat oder ein Administrator darf es löschen.")
conn.execute("DELETE FROM kanban_projects WHERE id = ?", (project_id,)) # Aufgaben & Links per CASCADE
return _overview(conn)
# ---------- Aufgaben ----------
def _get_task(conn, task_id: int):
row = conn.execute("SELECT * FROM kanban_tasks WHERE id = ?", (task_id,)).fetchone()
if not row:
raise HTTPException(404, "Aufgabe nicht gefunden – vielleicht wurde sie gerade gelöscht.")
return row
def _end_of(conn, project_id: int, status: str) -> float:
return _next(conn, "kanban_tasks", "WHERE project_id = ? AND status = ?", (project_id, status))
@router.post("/api/kanban/projekte/{project_id}/aufgaben", status_code=201)
def create_task(project_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
title = _name(data.get("title"), "Titel", MAX_TITLE)
description = _description(data.get("description"))
status = _status(data.get("status", "offen"))
priority = _priority(data.get("priority", "mittel"))
with transaction() as conn:
if not conn.execute("SELECT 1 FROM kanban_projects WHERE id = ?", (project_id,)).fetchone():
raise HTTPException(404, "Kanban-Projekt nicht gefunden.")
if conn.execute("SELECT COUNT(*) FROM kanban_tasks WHERE project_id = ?", (project_id,)).fetchone()[0] >= MAX_TASKS:
raise ValidationError(f"Maximal {MAX_TASKS} Aufgaben pro Projekt. Bitte erledigte Aufgaben löschen.")
now = now_iso()
conn.execute(
"""INSERT INTO kanban_tasks (project_id, title, description, status, assignee, priority, position,
created_at, created_by, updated_at, updated_by)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(project_id, title, description, status, _assignee(conn, data.get("assignee")), priority,
_end_of(conn, project_id, status), now, user["id"], now, user["id"]),
)
return _board(conn, project_id)
@router.patch("/api/kanban/aufgaben/{task_id}")
def update_task(task_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
"""Felder einzeln ändern. Neuer Status oder neue Priorität ohne Position: ans Ende der Stufe."""
with transaction() as conn:
row = _get_task(conn, task_id)
values = {}
if "title" in data:
values["title"] = _name(data["title"], "Titel", MAX_TITLE)
if "description" in data:
values["description"] = _description(data["description"])
if "assignee" in data:
values["assignee"] = _assignee(conn, data["assignee"])
if "status" in data:
values["status"] = _status(data["status"])
if "priority" in data:
values["priority"] = _priority(data["priority"])
if "position" in data:
values["position"] = _position(data["position"])
elif values.get("status", row["status"]) != row["status"] or values.get("priority", row["priority"]) != row["priority"]:
# Neue Spalte oder neue Priorität: ans Ende (innerhalb der Stufe, da nach Priorität sortiert wird)
values["position"] = _end_of(conn, row["project_id"], values.get("status", row["status"]))
if values:
values.update(updated_at=now_iso(), updated_by=user["id"])
columns = ", ".join(f"{k} = ?" for k in values) # Schlüssel stammen nur aus der Liste oben
conn.execute(f"UPDATE kanban_tasks SET {columns} WHERE id = ?", (*values.values(), task_id))
return _board(conn, row["project_id"])
@router.delete("/api/kanban/aufgaben/{task_id}")
def delete_task(task_id: int, user: dict = Depends(current_user)):
with transaction() as conn:
row = _get_task(conn, task_id)
if not _may_delete(user, row["created_by"]):
raise HTTPException(403, "Nur wer die Aufgabe angelegt hat oder ein Administrator darf sie löschen.")
conn.execute("DELETE FROM kanban_tasks WHERE id = ?", (task_id,))
return _board(conn, row["project_id"])
# ---------- Für die Artikelseite ----------
def boards_for_article(conn, article_id: str) -> list:
rows = conn.execute(
"""SELECT k.id, k.name FROM kanban_project_links l JOIN kanban_projects k ON k.id = l.project_id
WHERE l.article_id = ? ORDER BY k.name COLLATE NOCASE""",
(article_id,),
).fetchall()
return [{"id": r["id"], "name": r["name"]} for r in rows]
backend/app/links.py
+64
View File
@@ -0,0 +1,64 @@
"""Schnellzugriff-Links (Jellyfin, Audiobookshelf, Homer) – pro Konto."""
import re
from urllib.parse import urlparse
from fastapi import APIRouter, Body, Depends
from .auth import current_user
from .db import transaction
router = APIRouter()
KEYS = ("jellyfin", "audiobookshelf", "homer")
MAX_LEN = 500
HOSTNAME = re.compile(r"[A-Za-z0-9.-]+|\[[0-9A-Fa-f:.]+\]")
class ValidationError(ValueError):
pass
def normalize_url(value, field: str) -> str:
"""Prüft eine http(s)-Adresse; ergänzt fehlendes "http://". Leer bleibt leer."""
v = value.strip() if isinstance(value, str) else ""
if not v:
return ""
if "://" not in v:
v = f"http://{v}" # "jellyfin.local:8096" -> "http://jellyfin.local:8096"
try:
parsed = urlparse(v)
host = parsed.netloc.rsplit("@", 1)[-1]
host = host[: host.find("]") + 1] if host.startswith("[") else host.split(":")[0]
parsed.port # wirft bei ungültigem Port
except ValueError:
host, parsed = "", None
if (
not parsed
or parsed.scheme not in ("http", "https")
or not HOSTNAME.fullmatch(host)
or re.search(r"\s", v)
or len(v) > MAX_LEN
):
raise ValidationError(f"Ungültige Adresse für {field}.")
return v
def _get(conn, user_id: int) -> dict:
rows = conn.execute("SELECT key, url FROM user_links WHERE user_id = ?", (user_id,)).fetchall()
stored = {r["key"]: r["url"] for r in rows}
return {key: stored.get(key, "") for key in KEYS}
@router.get("/api/links")
def get_links(user: dict = Depends(current_user)):
with transaction() as conn:
return _get(conn, user["id"])
@router.put("/api/links")
def update_links(data: dict = Body(...), user: dict = Depends(current_user)):
values = {key: normalize_url(data[key], key) for key in KEYS if key in data}
with transaction() as conn:
for key, url in values.items():
conn.execute("INSERT OR REPLACE INTO user_links VALUES (?, ?, ?)", (user["id"], key, url))
return _get(conn, user["id"])
backend/app/main.py
+72
View File
@@ -0,0 +1,72 @@
"""P42 Dashboard – FastAPI-Backend.
Liefert die API unter /api, die hochgeladenen Bilder unter /uploads
und (falls gebaut) das React-Frontend aus frontend/dist.
"""
import re
from fastapi import Depends, FastAPI, HTTPException, Request
from fastapi.middleware.gzip import GZipMiddleware
from fastapi.responses import FileResponse, JSONResponse
from fastapi.staticfiles import StaticFiles
from . import auth, haushalt, kanban, links, monitoring, projects, reisen, rezept
from .auth import current_user
from .config import FRONTEND_DIST
from .tracking import track
app = FastAPI(title="P42 Dashboard", docs_url="/api/docs", openapi_url="/api/openapi.json")
# Komprimiert u. a. die JS-Bundles und die Weltkarte (~740 KB → ~200 KB)
app.add_middleware(GZipMiddleware, minimum_size=1000)
# ---------- Fehler einheitlich als {"error": "..."} ----------
@app.exception_handler(HTTPException)
async def http_error(_: Request, exc: HTTPException):
return JSONResponse({"error": exc.detail}, status_code=exc.status_code)
@app.exception_handler(projects.ValidationError)
@app.exception_handler(links.ValidationError)
async def validation_error(_: Request, exc: ValueError):
return JSONResponse({"error": str(exc)}, status_code=400)
# ---------- API ----------
app.include_router(auth.router)
app.include_router(links.router)
app.include_router(projects.router)
app.include_router(monitoring.router)
app.include_router(haushalt.router)
app.include_router(rezept.router)
app.include_router(reisen.router)
app.include_router(kanban.router)
@app.get("/api/track")
async def get_tracking(number: str = "", _: dict = Depends(current_user)):
number = re.sub(r"\s+", "", number).upper()
if not re.fullmatch(r"[A-Z0-9-]{6,40}", number):
raise HTTPException(400, "Ungültige Sendungsnummer.")
return await track(number)
@app.api_route("/api/{path:path}", methods=["GET", "POST", "PUT", "PATCH", "DELETE"], include_in_schema=False)
def api_not_found(path: str):
raise HTTPException(404, "Nicht gefunden.")
# ---------- React-Frontend (nach `npm run build`) ----------
if (FRONTEND_DIST / "index.html").exists():
app.mount("/assets", StaticFiles(directory=FRONTEND_DIST / "assets"), name="assets")
@app.get("/{path:path}", include_in_schema=False)
def spa(path: str):
file = (FRONTEND_DIST / path).resolve()
if path and file.is_file() and file.is_relative_to(FRONTEND_DIST.resolve()):
return FileResponse(file)
# Alle anderen Pfade übernimmt der React-Router
return FileResponse(FRONTEND_DIST / "index.html", headers={"Cache-Control": "no-cache"})
backend/app/monitoring.py
+131
View File
@@ -0,0 +1,131 @@
"""Server-Monitoring pro Konto: eigene Serverliste + Erreichbarkeitsprüfung (HTTP oder TCP-Port)."""
import asyncio
import re
import time
import httpx
from fastapi import APIRouter, Body, Depends, HTTPException
from .auth import current_user
from .db import now_iso, transaction
from .links import ValidationError, normalize_url
router = APIRouter()
TYPES = {"http", "tcp"}
MAX_MONITORS = 50
TIMEOUT = 5
HOST_PORT = re.compile(r"(?P<host>[A-Za-z0-9.-]+|\[[0-9A-Fa-f:.]+\]):(?P<port>\d{1,5})")
def _monitor(row) -> dict:
return {"id": row["id"], "name": row["name"], "type": row["type"], "target": row["target"], "createdAt": row["created_at"]}
def _validate(data: dict) -> dict:
name = data.get("name").strip() if isinstance(data.get("name"), str) else ""
if not name or len(name) > 60:
raise ValidationError("Name fehlt oder ist zu lang (max. 60 Zeichen).")
kind = data.get("type")
if kind not in TYPES:
raise ValidationError("Ungültiger Prüf-Typ.")
target = data.get("target").strip() if isinstance(data.get("target"), str) else ""
if kind == "http":
target = normalize_url(target, "die Adresse")
if not target:
raise ValidationError("Adresse fehlt.")
else:
m = HOST_PORT.fullmatch(target)
if not m or not 0 < int(m["port"]) < 65536:
raise ValidationError("Bitte Host und Port angeben, z. B. 192.168.0.10:22")
return {"name": name, "type": kind, "target": target}
# ---------- Verwaltung ----------
@router.get("/api/monitors")
def list_monitors(user: dict = Depends(current_user)):
with transaction() as conn:
rows = conn.execute("SELECT * FROM monitors WHERE user_id = ? ORDER BY id", (user["id"],)).fetchall()
return [_monitor(r) for r in rows]
@router.post("/api/monitors", status_code=201)
def create_monitor(data: dict = Body(...), user: dict = Depends(current_user)):
values = _validate(data)
with transaction() as conn:
count = conn.execute("SELECT COUNT(*) FROM monitors WHERE user_id = ?", (user["id"],)).fetchone()[0]
if count >= MAX_MONITORS:
raise ValidationError(f"Maximal {MAX_MONITORS} Server pro Konto.")
cur = conn.execute(
"INSERT INTO monitors (user_id, name, type, target, created_at) VALUES (?, ?, ?, ?, ?)",
(user["id"], values["name"], values["type"], values["target"], now_iso()),
)
return _monitor(conn.execute("SELECT * FROM monitors WHERE id = ?", (cur.lastrowid,)).fetchone())
@router.put("/api/monitors/{monitor_id}")
def update_monitor(monitor_id: int, data: dict = Body(...), user: dict = Depends(current_user)):
values = _validate(data)
with transaction() as conn:
cur = conn.execute(
"UPDATE monitors SET name = ?, type = ?, target = ? WHERE id = ? AND user_id = ?",
(values["name"], values["type"], values["target"], monitor_id, user["id"]),
)
if not cur.rowcount:
raise HTTPException(404, "Nicht gefunden.")
return _monitor(conn.execute("SELECT * FROM monitors WHERE id = ?", (monitor_id,)).fetchone())
@router.delete("/api/monitors/{monitor_id}")
def delete_monitor(monitor_id: int, user: dict = Depends(current_user)):
with transaction() as conn:
cur = conn.execute("DELETE FROM monitors WHERE id = ? AND user_id = ?", (monitor_id, user["id"]))
if not cur.rowcount:
raise HTTPException(404, "Nicht gefunden.")
return {"ok": True}
# ---------- Prüfung ----------
async def _check_http(client: httpx.AsyncClient, url: str) -> dict:
start = time.perf_counter()
try:
res = await client.get(url)
except httpx.TimeoutException:
return {"ok": False, "detail": "Zeitüberschreitung"}
except httpx.ConnectError:
return {"ok": False, "detail": "Nicht erreichbar"}
except httpx.HTTPError as e:
return {"ok": False, "detail": type(e).__name__}
ms = round((time.perf_counter() - start) * 1000)
# 4xx heißt: Server antwortet (z. B. Anmeldung nötig) – also erreichbar
return {"ok": res.status_code < 500, "ms": ms, "detail": f"HTTP {res.status_code}"}
async def _check_tcp(target: str) -> dict:
m = HOST_PORT.fullmatch(target)
host = m["host"].strip("[]")
start = time.perf_counter()
try:
_, writer = await asyncio.wait_for(asyncio.open_connection(host, int(m["port"])), TIMEOUT)
except asyncio.TimeoutError:
return {"ok": False, "detail": "Zeitüberschreitung"}
except OSError as e:
return {"ok": False, "detail": "Verbindung abgelehnt" if isinstance(e, ConnectionRefusedError) else "Nicht erreichbar"}
ms = round((time.perf_counter() - start) * 1000)
writer.close()
return {"ok": True, "ms": ms, "detail": f"Port {m['port']} offen"}
@router.get("/api/monitors/status")
async def monitor_status(user: dict = Depends(current_user)):
with transaction() as conn:
rows = conn.execute("SELECT id, type, target FROM monitors WHERE user_id = ?", (user["id"],)).fetchall()
# Zertifikate werden nicht geprüft: im Heimnetz sind selbstsignierte Zertifikate üblich
async with httpx.AsyncClient(timeout=TIMEOUT, verify=False, follow_redirects=True) as client:
results = await asyncio.gather(*(
_check_http(client, r["target"]) if r["type"] == "http" else _check_tcp(r["target"]) for r in rows
))
checked = now_iso()
return [{"id": r["id"], "checkedAt": checked, **res} for r, res in zip(rows, results)]
backend/app/projects.py
+270
View File
@@ -0,0 +1,270 @@
"""Projekte als Artikel (SQLite) inkl. Autor/letzter Bearbeiter; Bilder als Dateien in data/uploads."""
import re
import secrets
import time
import nh3
from fastapi import APIRouter, Body, Depends, HTTPException, Request
from fastapi.responses import FileResponse
from .auth import current_user
from .config import DATA_DIR
from .db import now_iso, transaction
from .kanban import boards_for_article
router = APIRouter()
UPLOAD_DIR = DATA_DIR / "uploads"
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
STATUSES = {"planung", "aktiv", "pausiert", "abgeschlossen"}
UPLOAD_NAME = re.compile(r"[a-f0-9]{24}\.(jpg|png|gif|webp)")
CONTENT_IMAGES = re.compile(r"/uploads/([a-f0-9]{24}\.(?:jpg|png|gif|webp))")
MAX_CONTENT = 500_000
MAX_UPLOAD = 15 * 1024 * 1024
MIME = {"jpg": "image/jpeg", "png": "image/png", "gif": "image/gif", "webp": "image/webp"}
class ValidationError(ValueError):
pass
# ---------- Artikel-HTML bereinigen ----------
ALLOWED_TAGS = {"p", "br", "h2", "h3", "strong", "b", "em", "i", "u", "s", "ul", "ol", "li", "blockquote", "hr", "a", "img"}
ALLOWED_ATTRS = {"a": {"href"}, "img": {"src", "alt"}}
def _attribute_filter(tag: str, attr: str, value: str) -> str | None:
# Bilder nur aus den eigenen Uploads oder per https
if tag == "img" and attr == "src" and not (value.startswith("/uploads/") or value.startswith("https://")):
return None
return value
def sanitize(content: str) -> str:
cleaned = nh3.clean(
content,
tags=ALLOWED_TAGS,
clean_content_tags={"script", "style"},
attributes=ALLOWED_ATTRS,
attribute_filter=_attribute_filter,
url_schemes={"http", "https", "mailto"},
link_rel="noopener noreferrer",
)
# Bilder, deren Quelle entfernt wurde, ganz weglassen
return re.sub(r"<img(?![^>]*\ssrc=)[^>]*>", "", cleaned).strip()
# ---------- Validierung ----------
def _text(value, field: str, max_len: int, required: bool = False) -> str:
v = value.strip() if isinstance(value, str) else ""
if required and not v:
raise ValidationError(f"{field} fehlt.")
if len(v) > max_len:
raise ValidationError(f"{field} ist zu lang (max. {max_len} Zeichen).")
return v
def _image(value) -> str | None:
if value in (None, ""):
return None
if not isinstance(value, str) or not UPLOAD_NAME.fullmatch(value):
raise ValidationError("Ungültiges Bild.")
return value
def _status(value) -> str:
if value not in STATUSES:
raise ValidationError("Ungültiger Status.")
return value
def _content(value) -> str:
v = sanitize(_text(value, "Artikeltext", MAX_CONTENT))
return "" if v in ("<p></p>", "<p><br></p>") else v
def _content_images(content: str | None) -> list[str]:
return CONTENT_IMAGES.findall(content or "")
# ---------- Uploads ----------
def _detect_image_type(data: bytes) -> str | None:
if data[:3] == b"\xff\xd8\xff":
return "jpg"
if data[:4] == b"\x89PNG":
return "png"
if data[:3] == b"GIF":
return "gif"
if data[:4] == b"RIFF" and data[8:12] == b"WEBP":
return "webp"
return None
def _referenced_images(conn) -> set[str]:
used = set()
for row in conn.execute("SELECT cover, content FROM projects"):
if row["cover"]:
used.add(row["cover"])
used.update(_content_images(row["content"]))
return used
def _remove_unused(conn, names) -> None:
used = _referenced_images(conn)
for n in names:
if n and n not in used:
(UPLOAD_DIR / n).unlink(missing_ok=True)
def _cleanup_orphans() -> None:
"""Hochgeladene, aber nie gespeicherte Bilder (älter als 1 Tag) entfernen."""
with transaction() as conn:
used = _referenced_images(conn)
cutoff = time.time() - 24 * 60 * 60
for f in UPLOAD_DIR.iterdir():
if UPLOAD_NAME.fullmatch(f.name) and f.name not in used and f.stat().st_mtime < cutoff:
f.unlink(missing_ok=True)
_cleanup_orphans()
# ---------- Projekte ----------
_SELECT = """
SELECT p.*, cu.display_name AS created_by_name, uu.display_name AS updated_by_name
FROM projects p
LEFT JOIN users cu ON cu.id = p.created_by
LEFT JOIN users uu ON uu.id = p.updated_by
"""
def _person(user_id, name) -> dict | None:
return {"id": user_id, "name": name} if user_id is not None and name else None
def _summary(row) -> dict:
return {
"id": row["id"],
"title": row["title"],
"description": row["description"],
"status": row["status"],
"cover": row["cover"] or next(iter(_content_images(row["content"])), None),
"createdAt": row["created_at"],
"createdBy": _person(row["created_by"], row["created_by_name"]),
"updatedAt": row["updated_at"],
"updatedBy": _person(row["updated_by"], row["updated_by_name"]),
}
def _get(conn, project_id: str) -> dict | None:
row = conn.execute(f"{_SELECT} WHERE p.id = ?", (project_id,)).fetchone()
if not row:
return None
# Verknüpfte Kanban-Boards für die Seitenleiste
return {**_summary(row), "cover": row["cover"], "content": row["content"], "boards": boards_for_article(conn, project_id)}
def _not_found():
return HTTPException(404, "Nicht gefunden.")
@router.get("/api/projects")
def list_projects(_: dict = Depends(current_user)):
with transaction() as conn:
return [_summary(r) for r in conn.execute(f"{_SELECT} ORDER BY p.updated_at DESC")]
@router.post("/api/projects", status_code=201)
def create_project(data: dict = Body(...), user: dict = Depends(current_user)):
project_id = secrets.token_hex(6)
now = now_iso()
with transaction() as conn:
conn.execute(
"""INSERT INTO projects (id, title, description, status, cover, content, created_at, created_by, updated_at, updated_by)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(
project_id,
_text(data.get("title"), "Titel", 120, required=True),
_text(data.get("description"), "Kurzbeschreibung", 2000),
_status(data.get("status") or "aktiv"),
_image(data.get("cover")),
_content(data.get("content")),
now, user["id"], now, user["id"],
),
)
return _get(conn, project_id)
@router.get("/api/projects/{project_id}")
def get_project(project_id: str, _: dict = Depends(current_user)):
with transaction() as conn:
return _get(conn, project_id) or _raise(_not_found())
@router.patch("/api/projects/{project_id}")
def update_project(project_id: str, data: dict = Body(...), user: dict = Depends(current_user)):
fields = {}
if "title" in data:
fields["title"] = _text(data["title"], "Titel", 120, required=True)
if "description" in data:
fields["description"] = _text(data["description"], "Kurzbeschreibung", 2000)
if "status" in data:
fields["status"] = _status(data["status"])
if "cover" in data:
fields["cover"] = _image(data["cover"])
if "content" in data:
fields["content"] = _content(data["content"])
fields.update(updated_at=now_iso(), updated_by=user["id"])
with transaction() as conn:
old = conn.execute("SELECT cover, content FROM projects WHERE id = ?", (project_id,)).fetchone()
if not old:
raise _not_found()
assignments = ", ".join(f"{k} = ?" for k in fields)
conn.execute(f"UPDATE projects SET {assignments} WHERE id = ?", (*fields.values(), project_id))
_remove_unused(conn, [old["cover"], *_content_images(old["content"])])
return _get(conn, project_id)
@router.delete("/api/projects/{project_id}")
def delete_project(project_id: str, user: dict = Depends(current_user)):
with transaction() as conn:
row = conn.execute("SELECT cover, content, created_by FROM projects WHERE id = ?", (project_id,)).fetchone()
if not row:
raise _not_found()
if not user["isAdmin"] and row["created_by"] != user["id"]:
raise HTTPException(403, "Nur der Autor oder ein Administrator darf dieses Projekt löschen.")
conn.execute("DELETE FROM projects WHERE id = ?", (project_id,))
_remove_unused(conn, [row["cover"], *_content_images(row["content"])])
return {"ok": True}
@router.post("/api/uploads", status_code=201)
async def upload(request: Request, _: dict = Depends(current_user)):
data = bytearray()
async for chunk in request.stream():
data += chunk
if len(data) > MAX_UPLOAD:
raise HTTPException(413, "Datei zu groß (max. 15 MB).")
ext = _detect_image_type(bytes(data))
if not ext:
raise ValidationError("Bildformat nicht unterstützt (erlaubt: JPG, PNG, GIF, WebP).")
name = f"{secrets.token_hex(12)}.{ext}"
(UPLOAD_DIR / name).write_bytes(data)
return {"file": name}
@router.get("/uploads/{name}")
def get_upload(name: str, _: dict = Depends(current_user)):
path = UPLOAD_DIR / name
if not UPLOAD_NAME.fullmatch(name) or not path.exists():
raise _not_found()
return FileResponse(path, media_type=MIME[path.suffix[1:]], headers={"Cache-Control": "private, max-age=31536000, immutable"})
def _raise(exc: Exception):
raise exc
backend/app/reisen.py
+305
View File
@@ -0,0 +1,305 @@
"""Reise-Karte pro Konto: besuchte Städte und Länder sowie kurze Infos dazu.
Städte werden über die Geocoding-API von Open-Meteo gesucht (Daten: GeoNames, CC BY 4.0).
Infotexte kommen aus der deutschen Wikipedia, Einwohnerzahlen von Ländern aus Wikidata.
Alle Abrufe laufen über den Server und werden zwischengespeichert; der Browser fragt keine Fremdserver.
"""
import json
import math
import re
import time
from urllib.parse import quote
import httpx
from fastapi import APIRouter, Body, Depends, HTTPException
from .auth import current_user
from .config import WIKIPEDIA_CONTACT
from .db import now_iso, transaction
from .links import ValidationError
router = APIRouter()
HEADERS = {"User-Agent": "P42-Dashboard/1.0 (privates Heim-Dashboard)"}
# Wikimedia sperrt Clients ohne Kontaktangabe (https://w.wiki/4wJS)
WIKI_HEADERS = {"User-Agent": f"P42-Dashboard/1.0 (privates Heim-Dashboard; {WIKIPEDIA_CONTACT}) python-httpx"}
TIMEOUT = 8
MAX_CITIES = 2000
CACHE_OK = 30 * 24 * 3600 # Infos ändern sich selten
CACHE_MISS = 24 * 3600 # nichts gefunden: morgen noch mal versuchen
COUNTRY_CODE = re.compile(r"[A-Z]{2}|X-[A-Za-z. ]{1,40}") # X-…: Gebiete ohne ISO-Code auf der Karte
_search_cache: dict[str, list] = {}
def _city(row) -> dict:
return {
"id": row["id"], "geonameId": row["geoname_id"], "name": row["name"], "admin1": row["admin1"],
"country": row["country"], "lat": row["lat"], "lon": row["lon"], "population": row["population"],
"elevation": row["elevation"], "timezone": row["timezone"], "createdAt": row["created_at"],
}
def _state(conn, user_id: int) -> dict:
cities = conn.execute("SELECT * FROM travel_cities WHERE user_id = ? ORDER BY name", (user_id,)).fetchall()
countries = conn.execute("SELECT code FROM travel_countries WHERE user_id = ? ORDER BY code", (user_id,)).fetchall()
return {"cities": [_city(r) for r in cities], "countries": [r["code"] for r in countries]}
# ---------- Verwaltung ----------
@router.get("/api/reisen")
def get_travel(user: dict = Depends(current_user)):
with transaction() as conn:
return _state(conn, user["id"])
def _text(value, field: str, max_len: int, required: bool = True) -> str:
v = re.sub(r"\s+", " ", value).strip() if isinstance(value, str) else ""
if required and not v:
raise ValidationError(f"{field} fehlt.")
if len(v) > max_len:
raise ValidationError(f"{field} ist zu lang.")
return v
def _number(value, field: str, low: float, high: float) -> float:
if isinstance(value, bool) or not isinstance(value, (int, float)) or not low <= value <= high:
raise ValidationError(f"Ungültiger Wert für {field}.")
return float(value)
@router.post("/api/reisen/staedte", status_code=201)
def add_city(data: dict = Body(...), user: dict = Depends(current_user)):
geoname_id = data.get("geonameId")
if isinstance(geoname_id, bool) or not isinstance(geoname_id, int) or geoname_id <= 0:
raise ValidationError("Ungültige Stadt.")
country = data.get("country")
if not isinstance(country, str) or not re.fullmatch(r"[A-Z]{2}", country):
raise ValidationError("Ungültiges Land.")
population = data.get("population")
if population is not None and (isinstance(population, bool) or not isinstance(population, int) or population < 0):
population = None
elevation = data.get("elevation")
if elevation is not None and (isinstance(elevation, bool) or not isinstance(elevation, (int, float)) or not -500 <= elevation <= 9000):
elevation = None
timezone = data.get("timezone")
if not isinstance(timezone, str) or not re.fullmatch(r"[A-Za-z_]+(?:/[A-Za-z0-9_+-]+){0,2}", timezone):
timezone = None
values = (
user["id"], geoname_id, _text(data.get("name"), "Name", 100), _text(data.get("admin1"), "Region", 100, False),
country, _number(data.get("lat"), "Breitengrad", -90, 90), _number(data.get("lon"), "Längengrad", -180, 180),
population, elevation, timezone, now_iso(),
)
with transaction() as conn:
if conn.execute("SELECT COUNT(*) FROM travel_cities WHERE user_id = ?", (user["id"],)).fetchone()[0] >= MAX_CITIES:
raise ValidationError(f"Maximal {MAX_CITIES} Städte.")
conn.execute(
"""INSERT INTO travel_cities
(user_id, geoname_id, name, admin1, country, lat, lon, population, elevation, timezone, created_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT (user_id, geoname_id) DO NOTHING""",
values,
)
return _state(conn, user["id"])
@router.delete("/api/reisen/staedte/{city_id}")
def delete_city(city_id: int, user: dict = Depends(current_user)):
with transaction() as conn:
cur = conn.execute("DELETE FROM travel_cities WHERE id = ? AND user_id = ?", (city_id, user["id"]))
if not cur.rowcount:
raise HTTPException(404, "Stadt nicht gefunden.")
return _state(conn, user["id"])
def _country_code(code: str) -> str:
if not COUNTRY_CODE.fullmatch(code):
raise ValidationError("Ungültiges Land.")
return code
@router.put("/api/reisen/laender/{code}")
def mark_country(code: str, user: dict = Depends(current_user)):
with transaction() as conn:
conn.execute(
"INSERT OR IGNORE INTO travel_countries (user_id, code) VALUES (?, ?)", (user["id"], _country_code(code))
)
return _state(conn, user["id"])
@router.delete("/api/reisen/laender/{code}")
def unmark_country(code: str, user: dict = Depends(current_user)):
with transaction() as conn:
conn.execute("DELETE FROM travel_countries WHERE user_id = ? AND code = ?", (user["id"], _country_code(code)))
return _state(conn, user["id"])
# ---------- Städtesuche ----------
@router.get("/api/reisen/suche")
async def search_city(q: str = "", _: dict = Depends(current_user)):
q = re.sub(r"\s+", " ", q).strip()
if len(q) < 2 or len(q) > 80:
return []
key = q.casefold()
if key in _search_cache:
return _search_cache[key]
try:
async with httpx.AsyncClient(timeout=TIMEOUT, headers=HEADERS) as client:
res = await client.get(
"https://geocoding-api.open-meteo.com/v1/search",
params={"name": q, "count": 8, "language": "de", "format": "json"},
)
res.raise_for_status()
except httpx.HTTPError:
raise HTTPException(502, "Die Städtesuche ist gerade nicht erreichbar.")
results = [
{
"geonameId": r["id"], "name": r["name"], "admin1": r.get("admin1") or "", "countryName": r.get("country") or "",
"country": r["country_code"], "lat": r["latitude"], "lon": r["longitude"], "population": r.get("population"),
"elevation": r.get("elevation"), "timezone": r.get("timezone"),
}
for r in res.json().get("results", [])
if r.get("country_code") and r.get("id")
]
if len(_search_cache) > 300:
_search_cache.clear()
_search_cache[key] = results
return results
# ---------- Infos aus Wikipedia / Wikidata ----------
def _distance_km(a, b) -> float:
lat1, lon1, lat2, lon2 = map(math.radians, (a[0], a[1], b[0], b[1]))
h = math.sin((lat2 - lat1) / 2) ** 2 + math.cos(lat1) * math.cos(lat2) * math.sin((lon2 - lon1) / 2) ** 2
return 6371 * 2 * math.asin(math.sqrt(h))
async def _summary(client, title: str):
res = await client.get(f"https://de.wikipedia.org/api/rest_v1/page/summary/{quote(title.replace(' ', '_'), safe='')}")
return res.json() if res.status_code == 200 else None
async def _wikipedia(client, title: str, near=None, max_km=60):
"""Artikel zum Titel; bei Begriffsklärung oder falschem Ort über die Suche den Artikel in der Nähe finden."""
def fits(page, exact):
if not page or page.get("type") != "standard":
return False
if near is None:
return True
coords = page.get("coordinates")
if not coords:
return exact # exakter Titel ohne Koordinaten: meist trotzdem richtig
return _distance_km(near, (coords["lat"], coords["lon"])) <= max_km
page = await _summary(client, title)
if fits(page, True):
return page
res = await client.get(
"https://de.wikipedia.org/w/api.php",
params={"action": "query", "list": "search", "srsearch": title, "srlimit": 8, "format": "json"},
)
for hit in res.json().get("query", {}).get("search", []):
page = await _summary(client, hit["title"])
if fits(page, False):
return page
return None
async def _population(client, item: str):
"""Aktuelle Einwohnerzahl aus Wikidata (bevorzugter Wert, sonst der neueste)."""
res = await client.get(
"https://www.wikidata.org/w/api.php",
params={"action": "wbgetclaims", "entity": item, "property": "P1082", "format": "json"},
)
claims = res.json().get("claims", {}).get("P1082", [])
def year(c):
t = c.get("qualifiers", {}).get("P585", [{}])[0].get("datavalue", {}).get("value", {}).get("time", "")
return t[1:5] if t else ""
best = [c for c in claims if c.get("rank") == "preferred"] or claims
best = [c for c in best if c["mainsnak"].get("datavalue")]
if not best:
return None, None
claim = max(best, key=year)
return int(float(claim["mainsnak"]["datavalue"]["value"]["amount"])), year(claim) or None
async def _capital(client, item: str):
"""Aktuelle Hauptstadt(e) mit deutschem Namen aus Wikidata (ohne Enddatum)."""
res = await client.get(
"https://www.wikidata.org/w/api.php",
params={"action": "wbgetclaims", "entity": item, "property": "P36", "format": "json"},
)
claims = [
c for c in res.json().get("claims", {}).get("P36", [])
if c["mainsnak"].get("datavalue") and "P582" not in c.get("qualifiers", {}) and c.get("rank") != "deprecated"
]
ids = [c["mainsnak"]["datavalue"]["value"]["id"] for c in claims][:3]
if not ids:
return None
res = await client.get(
"https://www.wikidata.org/w/api.php",
params={"action": "wbgetentities", "ids": "|".join(ids), "props": "labels", "languages": "de|en", "format": "json"},
)
entities = res.json().get("entities", {})
names = [
(entities.get(i, {}).get("labels", {}).get("de") or entities.get(i, {}).get("labels", {}).get("en") or {}).get("value")
for i in ids
]
return ", ".join(n for n in names if n) or None
def _cache_get(conn, key):
row = conn.execute("SELECT value, fetched_at FROM info_cache WHERE key = ?", (key,)).fetchone()
if not row:
return None
value = json.loads(row["value"])
ttl = CACHE_OK if value.get("extract") else CACHE_MISS
return value if time.time() - row["fetched_at"] < ttl else None
@router.get("/api/reisen/info")
async def get_info(kind: str, name: str, code: str = "", lat: float | None = None, lon: float | None = None,
_: dict = Depends(current_user)):
"""Kurzinfo zu einem Land (kind=land, code=ISO) oder einer Stadt (kind=stadt, code=GeoNames-ID, lat/lon)."""
if kind not in ("land", "stadt") or not 1 <= len(name) <= 100 or len(code) > 45:
raise ValidationError("Ungültige Anfrage.")
if not WIKIPEDIA_CONTACT:
return {"extract": None, "url": None, "title": None, "population": None, "populationYear": None, "disabled": True}
key = f"{kind}:{code or name}"
with transaction() as conn:
cached = _cache_get(conn, key)
if cached is not None:
return cached
near = (lat, lon) if kind == "stadt" and lat is not None and lon is not None else None
try:
async with httpx.AsyncClient(timeout=TIMEOUT, headers=WIKI_HEADERS, follow_redirects=True) as client:
page = await _wikipedia(client, name, near)
info = {"extract": None, "url": None, "title": None, "population": None, "populationYear": None, "capital": None}
if page:
info.update(
extract=page.get("extract"),
url=page.get("content_urls", {}).get("desktop", {}).get("page"),
title=page.get("title"),
)
if kind == "land" and page.get("wikibase_item"):
info["population"], info["populationYear"] = await _population(client, page["wikibase_item"])
info["capital"] = await _capital(client, page["wikibase_item"])
except (httpx.HTTPError, ValueError):
# Nicht zwischenspeichern: beim nächsten Klick erneut versuchen
return {"extract": None, "url": None, "title": None, "population": None, "populationYear": None, "offline": True}
with transaction() as conn:
conn.execute(
"INSERT OR REPLACE INTO info_cache (key, value, fetched_at) VALUES (?, ?, ?)",
(key, json.dumps(info, ensure_ascii=False), time.time()),
)
return info
backend/app/reset_password.py
+30
View File
@@ -0,0 +1,30 @@
"""Notfall: Passwort eines Kontos neu setzen (z. B. wenn der einzige Admin es vergessen hat).
Aufruf im Ordner backend/: .venv/bin/python -m app.reset_password BENUTZERNAME
"""
import getpass
import sys
from .auth import MIN_PASSWORD, hash_password
from .db import transaction
def main() -> None:
if len(sys.argv) != 2:
sys.exit("Aufruf: python -m app.reset_password BENUTZERNAME")
username = sys.argv[1]
password = getpass.getpass(f"Neues Passwort für {username}: ")
if len(password) < MIN_PASSWORD:
sys.exit(f"Das Passwort muss mindestens {MIN_PASSWORD} Zeichen lang sein.")
if getpass.getpass("Wiederholen: ") != password:
sys.exit("Die Passwörter stimmen nicht überein.")
with transaction() as conn:
cur = conn.execute("UPDATE users SET password_hash = ? WHERE username = ?", (hash_password(password), username))
if not cur.rowcount:
sys.exit(f"Kein Konto mit dem Benutzernamen „{username}“.")
conn.execute("DELETE FROM sessions WHERE user_id = (SELECT id FROM users WHERE username = ?)", (username,))
print("Passwort geändert.")
if __name__ == "__main__":
main()
backend/app/rezept.py
+357
View File
@@ -0,0 +1,357 @@
"""Rezepte abgleichen: Zutaten von einer Rezeptseite (schema.org/Recipe) oder aus eingefügtem Text lesen,
den Produkten im Katalog zuordnen und Fehlendes auf die Einkaufsliste setzen.
Abgerufen wird nur die eine Seite, deren Link eingefügt wurde. Chefkoch & Co. betten die Zutaten als
JSON-LD ("recipeIngredient") ein; dafür braucht es keine API und keinen HTML-Parser.
"""
import html
import ipaddress
import json
import re
import socket
from urllib.parse import urljoin, urlsplit
import httpx
from fastapi import APIRouter, Body, Depends, HTTPException
from .auth import current_user
from .db import transaction
from .haushalt import _add, _canonical_name, _clean_name, _key, _quantity, _state, _unit
from .links import ValidationError
router = APIRouter()
MAX_PAGE = 3 * 1024 * 1024
MAX_REDIRECTS = 3
TIMEOUT = 10
MAX_INGREDIENTS = 80
MAX_TEXT = 10_000
# ---------- Seite abrufen ----------
def _check_host(url: str) -> None:
"""Nur öffentliche Adressen abrufen – sonst ließe sich über den Server das Heimnetz abfragen."""
parts = urlsplit(url)
if parts.scheme not in ("http", "https") or not parts.hostname:
raise ValidationError("Bitte einen Link mit http:// oder https:// einfügen.")
try:
infos = socket.getaddrinfo(parts.hostname, parts.port or (443 if parts.scheme == "https" else 80))
except socket.gaierror:
raise ValidationError("Die Seite wurde nicht gefunden. Stimmt der Link?")
for info in infos:
if not ipaddress.ip_address(info[4][0].split("%")[0]).is_global:
raise ValidationError("Links ins eigene Netzwerk werden nicht abgerufen.")
async def _fetch(url: str) -> str:
headers = {"User-Agent": "Mozilla/5.0 (P42-Dashboard; Rezeptabgleich)", "Accept": "text/html"}
async with httpx.AsyncClient(timeout=TIMEOUT, follow_redirects=False, headers=headers) as client:
for _ in range(MAX_REDIRECTS + 1):
_check_host(url)
try:
async with client.stream("GET", url) as res:
if res.is_redirect and res.headers.get("location"):
url = urljoin(url, res.headers["location"])
continue
if res.status_code >= 400:
raise ValidationError(f"Die Seite antwortet mit Fehler {res.status_code}.")
data = bytearray()
async for chunk in res.aiter_bytes():
data += chunk
if len(data) > MAX_PAGE:
raise ValidationError("Die Seite ist zu groß.")
return data.decode(res.encoding or "utf-8", errors="replace")
except httpx.TimeoutException:
raise ValidationError("Die Seite hat nicht rechtzeitig geantwortet.")
except httpx.HTTPError:
raise ValidationError("Die Seite konnte nicht abgerufen werden.")
raise ValidationError("Zu viele Weiterleitungen.")
def _find_recipe(page: str):
for m in re.finditer(r"<script[^>]*application/ld\+json[^>]*>(.*?)</script>", page, re.S | re.I):
try:
data = json.loads(m.group(1).strip())
except ValueError:
continue
stack = [data]
while stack:
node = stack.pop()
if isinstance(node, list):
stack.extend(node)
elif isinstance(node, dict):
kind = node.get("@type")
if kind == "Recipe" or (isinstance(kind, list) and "Recipe" in kind):
return node
stack.extend(v for k, v in node.items() if k in ("@graph", "mainEntity", "itemListElement"))
return None
def _servings(value):
"""recipeYield kommt als Zahl, Text ("4 Portionen") oder Liste davon."""
for v in value if isinstance(value, list) else [value]:
m = re.search(r"\d+", str(v or ""))
if m and 0 < int(m.group()) <= 100:
return int(m.group())
return None
# ---------- Zutaten zerlegen ----------
FRACTIONS = {"½": 0.5, "¼": 0.25, "¾": 0.75, "⅓": 1 / 3, "⅔": 2 / 3, "⅛": 0.125}
NUMBER = r"(?:\d*[½¼¾⅓⅔⅛]|\d+(?:[.,]\d+)?(?:\s+\d+/\d+)?|\d+/\d+)"
AMOUNT = re.compile(rf"^(?P<a>{NUMBER})(?:\s*[-–]\s*(?P<b>{NUMBER}))?\s*")
UNITS = {
"g", "gr", "gramm", "kg", "mg", "ml", "l", "liter", "cl", "dl", "el", "tl", "essl", "teel", "msp",
"messerspitze", "prise", "pck", "päckchen", "packung", "pkt", "dose", "becher", "bund", "zehe", "stück",
"stk", "tasse", "glas", "gläser", "scheibe", "schuss", "spritzer", "handvoll", "hand", "blatt", "blätter",
"stange", "stängel", "kopf", "würfel", "beutel", "tropfen", "zweig", "knolle", "tüte", "flasche", "rolle",
"cm", "etwas", "wenig", "einige", "tafel", "riegel", "netz", "schale", "kugel",
}
OPTIONAL = re.compile(r"^(?:evtl\.?|eventuell|nach belieben|n\.\s?b\.?|optional|ggf\.?)\s*", re.I)
SIZE_WORDS = re.compile(
r"^(?:(?:sehr\s+)?(?:kl\.|gr\.|m\.-?\s?groß(?:e|er|es|en)?|"
r"(?:groß|klein|mittelgroß|frisch|reif|weich|kalt|warm|lauwarm|gehackt|gemahlen)(?:e|er|es|en)?)\s+)+",
re.I,
)
# Eiweiß, Eigelb & Co. kauft man als ganze Eier
EGG_PARTS = {"eiweiß": "weiß", "eiweiss": "weiß", "eiklar": "weiß", "eigelb": "gelb", "eidotter": "gelb", "dotter": "gelb"}
EGG_PART = re.compile(r"^(eiweiß|eiweiss|eiklar|eigelb|eidotter|dotter)(?:e|en|s)?(?=\s|$)")
# Selbstverständliches, das meist im Haus ist – nicht vorab ankreuzen
STAPLES = {"salz", "pfeffer", "wasser", "salz und pfeffer", "eiswasser", "leitungswasser", "öl zum braten"}
def _number(text: str) -> float:
text = text.strip().replace(",", ".")
if text[-1] in FRACTIONS:
return (float(text[:-1]) if text[:-1] else 0) + FRACTIONS[text[-1]]
total = 0.0
for part in text.split():
num, _, den = part.partition("/")
total += float(num) / float(den) if den else float(num)
return total
def _recipe_unit(token: str):
t = re.sub(r"\((?:n|s|e|en|er)\)$|/(?:n|s|e|en)$", "", token.casefold().rstrip(".,;:"))
t = t.rstrip(".")
return t if t in UNITS else None
def _plural(name: str) -> str:
# "Ei(er)" → "Eier", "Zwiebel(n)" → "Zwiebeln", "Zehe/n" → "Zehen"
return re.sub(r"(\w)(?:\((\w{1,3})\)|/(\w{1,2}))(?=\W|$)", lambda m: m[1] + (m[2] or m[3]), name)
def parse_ingredient(raw: str):
text = re.sub(r"\s+", " ", html.unescape(str(raw))).strip()
if not text:
return None
rest = text
optional = bool(OPTIONAL.match(rest))
rest = OPTIONAL.sub("", rest)
amount = None
m = AMOUNT.match(rest)
if m:
# Bei "2-3" lieber die größere Menge besorgen
amount = _number(m["b"] or m["a"])
rest = rest[m.end():]
rest = SIZE_WORDS.sub("", rest) # "1 kl. Dose Mais" → Einheit Dose
unit = None
first, _, tail = rest.partition(" ")
if tail and _recipe_unit(first):
unit, rest = _recipe_unit(first), tail
if first.endswith(","): # "1 TL, gestr. Zimt" → Zusatz hinter dem Komma ist keine Zutat
rest = re.sub(r"^[^ ]*\.\s*", "", rest) if re.match(r"^\w+\.\s", rest) else rest
# Klammern: "(oder Butter)" ist eine Alternative, alles andere ein Hinweis
alternatives = [a.strip() for a in re.findall(r"\(oder ([^)]+)\)", rest, re.I)]
name = re.sub(r"\s*\([^)]*\)", "", _plural(rest))
name = name.partition(",")[0] # Hinweise hinter dem Komma ("Saft davon") gehören nicht zum Namen
# "Schmand oder Crème fraîche" → Schmand, Crème fraîche als Alternative
name, *others = re.split(r"\s+oder\s+", name.strip(), flags=re.I)
alternatives += [o.strip() for o in others if o.strip()]
name = SIZE_WORDS.sub("", name.strip()).strip(" .-")
if not name:
return None
name = name[0].upper() + name[1:]
egg = EGG_PART.match(name.casefold())
if egg or name.casefold() == "ei":
name = "Eier"
return {
"raw": text,
"amount": round(amount, 3) if amount is not None else None,
"unit": unit,
"name": name[:60],
"alternatives": [_plural(a)[:60] for a in alternatives],
"optional": optional,
"staple": name.casefold() in STAPLES,
"eggPart": EGG_PARTS[egg[1]] if egg else None,
}
def merge(ingredients: list) -> list:
"""Doppelte Zutaten (z. B. Mehl für Teig und Streusel) zusammenfassen, wenn die Einheit passt.
Eier: ganze Eier zählen voll, von Eigelb und Eiweiß nur der größere Teil –
"3 Eigelb + 3 Eiweiß" sind 3 getrennte Eier, nicht 6.
"""
merged = {}
for ing in ingredients:
key = (_key(ing["name"]), ing["unit"])
part = ing.pop("eggPart", None)
amount = ing["amount"] if ing["amount"] is not None else (1 if part else None)
if key not in merged:
merged[key] = {**ing, "amount": None, "raws": [], "whole": 0.0, "parts": {}, "counted": False}
entry = merged[key]
entry["raws"].append(ing["raw"])
entry["optional"] = entry["optional"] and ing["optional"]
if amount is None:
continue
entry["counted"] = True
if part:
entry["parts"][part] = entry["parts"].get(part, 0) + amount
else:
entry["whole"] += amount
result = []
for entry in merged.values():
if entry["counted"]:
entry["amount"] = round(entry["whole"] + max(entry["parts"].values(), default=0), 3)
entry["raw"] = " + ".join(entry["raws"])
for k in ("raws", "whole", "parts", "counted"):
del entry[k]
result.append(entry)
return result
# ---------- Zuordnung zu Produkten ----------
_UMLAUTS = str.maketrans({"ä": "a", "ö": "o", "ü": "u", "ß": "ss"})
def _stem(name: str) -> str:
"""Grobe Grundform, damit "Apfel"/"Äpfel", "Ei"/"Eier", "Tomate"/"Tomaten" zusammenfinden."""
s = name.casefold().translate(_UMLAUTS)
if s in ("ei", "eier"): # zu kurz für die Endungsregel; "Eis" soll nicht zu "Eier" werden
return "ei"
for ending in ("en", "er", "n", "e", "s"):
if s.endswith(ending) and len(s) - len(ending) >= 3:
return s[: -len(ending)]
return s
def _matcher(conn):
aliases = {r["key"]: r["product"] for r in conn.execute("SELECT key, product FROM ingredient_aliases")}
products = [r["name"] for r in conn.execute("SELECT name FROM products")]
exact = {_key(p): p for p in products}
stems = {}
for p in products:
stems.setdefault(_stem(p), p)
def lookup(name: str):
k = _key(name)
if k in aliases:
return aliases[k]
if k in exact:
return exact[k]
if _stem(name) in stems:
return stems[_stem(name)]
# "brauner Zucker" → "Zucker"
last = name.split()[-1]
if last != name:
return exact.get(_key(last)) or stems.get(_stem(last))
return None
def match(ing: dict) -> dict:
found = lookup(ing["name"])
if not found:
for alt in ing["alternatives"]:
if found := lookup(alt):
break
return {**ing, "product": found or ing["name"], "known": bool(found)}
return match
def _result(conn, title, source, servings, raw_ingredients) -> dict:
parsed = [p for p in (parse_ingredient(r) for r in raw_ingredients[:MAX_INGREDIENTS]) if p]
if not parsed:
raise ValidationError("Keine Zutaten gefunden.")
match = _matcher(conn)
return {
"title": title,
"source": source,
"servings": servings,
"ingredients": [match(i) for i in merge(parsed)],
}
# ---------- API ----------
@router.post("/api/haushalt/rezept")
async def load_recipe(data: dict = Body(...), _: dict = Depends(current_user)):
"""Zutaten aus einem Rezept-Link ({url}) oder eingefügtem Text ({text, servings}) lesen."""
url = data.get("url").strip() if isinstance(data.get("url"), str) else ""
text = data.get("text") if isinstance(data.get("text"), str) else ""
if url:
if not re.match(r"^https?://", url, re.I):
url = "https://" + url
recipe = _find_recipe(await _fetch(url))
if not recipe or not recipe.get("recipeIngredient"):
raise ValidationError(
"Auf dieser Seite wurden keine Rezeptdaten gefunden. Du kannst die Zutaten stattdessen einfügen."
)
ingredients = recipe["recipeIngredient"]
if isinstance(ingredients, str):
ingredients = ingredients.splitlines()
title = html.unescape(str(recipe.get("name") or "Rezept"))[:200]
with transaction() as conn:
return _result(conn, title, url, _servings(recipe.get("recipeYield")), ingredients)
if text.strip():
if len(text) > MAX_TEXT:
raise ValidationError("Der Text ist zu lang.")
servings = data.get("servings")
servings = servings if isinstance(servings, int) and 0 < servings <= 100 else None
lines = [line.strip(" \t-•*·") for line in text.splitlines()]
with transaction() as conn:
return _result(conn, "Eigene Zutatenliste", None, servings, [line for line in lines if line])
raise ValidationError("Bitte einen Rezept-Link oder eine Zutatenliste angeben.")
@router.post("/api/haushalt/rezept/uebernehmen")
def take_over(data: dict = Body(...), user: dict = Depends(current_user)):
"""Ausgewählte Zutaten auf die Einkaufsliste setzen und korrigierte Zuordnungen merken."""
items = data.get("items") if isinstance(data.get("items"), list) else []
aliases = data.get("aliases") if isinstance(data.get("aliases"), list) else []
if not items and not aliases:
raise ValidationError("Nichts ausgewählt.")
if len(items) > MAX_INGREDIENTS or len(aliases) > MAX_INGREDIENTS:
raise ValidationError("Zu viele Zutaten.")
# Erst alles prüfen, dann in einem Rutsch speichern
checked = []
for i in items:
if not isinstance(i, dict):
raise ValidationError("Ungültige Zutat.")
unit = _unit(i.get("unit"))
checked.append((_clean_name(i.get("name")), _quantity(i.get("quantity"), unit), unit))
learned = [(_clean_name(a.get("ingredient")), _clean_name(a.get("product"))) for a in aliases if isinstance(a, dict)]
with transaction() as conn:
for ingredient, product in learned:
product = _canonical_name(conn, product)
if _key(ingredient) == _key(product):
conn.execute("DELETE FROM ingredient_aliases WHERE key = ?", (_key(ingredient),))
else:
conn.execute(
"INSERT INTO ingredient_aliases (key, product) VALUES (?, ?) "
"ON CONFLICT (key) DO UPDATE SET product = excluded.product",
(_key(ingredient), product),
)
for name, quantity, unit in checked:
_add(conn, _canonical_name(conn, name), "einkauf", quantity, unit, user["id"])
return _state(conn)
backend/app/tracking/__init__.py
+50
View File
@@ -0,0 +1,50 @@
"""Universelle Sendungsverfolgung: erst DHL direkt, dann 17TRACK, sonst nur Erkennung + Links."""
import logging
import time
from ..config import TRACKING
from .carriers import detect_carriers, universal_link
from .dhl import track_dhl
from .seventeen import track_17
log = logging.getLogger(__name__)
CACHE_SECONDS = 120
_cache: dict[str, tuple[float, dict]] = {}
async def track(number: str) -> dict:
cached = _cache.get(number)
if cached and time.monotonic() - cached[0] < CACHE_SECONDS:
return cached[1]
candidates = detect_carriers(number)
base = {"number": number, "candidates": candidates, "universalLink": universal_link(number)}
dhl_key = TRACKING.get("dhlApiKey")
key_17 = TRACKING.get("seventeenTrackApiKey")
errors = []
if dhl_key and any(c["id"] == "dhl" for c in candidates):
try:
result = await track_dhl(number, dhl_key)
if result:
return _store(number, {**base, **result})
except Exception as e: # noqa: BLE001 – Fehler anzeigen, nächsten Anbieter probieren
log.warning("DHL-Abfrage fehlgeschlagen: %s", e)
errors.append(f"DHL: {e}")
if key_17:
try:
result = await track_17(number, key_17)
data = {**base, **result}
return _store(number, data) if result["status"] != "pending" else data
except Exception as e: # noqa: BLE001
log.warning("17TRACK-Abfrage fehlgeschlagen: %s", e)
errors.append(f"17TRACK: {e}")
return {**base, "source": None, "status": "unknown", "configured": bool(dhl_key or key_17), "errors": errors}
def _store(number: str, data: dict) -> dict:
_cache[number] = (time.monotonic(), data)
return data
backend/app/tracking/carriers.py
+37
View File
@@ -0,0 +1,37 @@
"""Erkennt anhand des Nummernformats, welche Paketdienste in Frage kommen.
Viele Formate überschneiden sich, deshalb kann es mehrere Treffer geben.
"""
import re
from urllib.parse import quote
def universal_link(number: str) -> str:
return f"https://t.17track.net/de#nums={quote(number)}"
_DHL = "https://www.dhl.de/de/privatkunden/pakete-empfangen/verfolgen.html?piececode={}"
CARRIERS = [
("dhl", "DHL", [r"\d{12}", r"\d{20}", r"(JJD|JVGL)\d{8,}", r"00340\d{15}", r"\d{10}"], _DHL),
("deutschepost", "Deutsche Post", [r"[A-Z]{2}\d{9}DE"], _DHL),
("ups", "UPS", [r"1Z[0-9A-Z]{16}"], "https://www.ups.com/track?loc=de_DE&tracknum={}"),
("hermes", "Hermes", [r"\d{14}", r"\d{16}", r"H\d{19}"],
"https://www.myhermes.de/empfangen/sendungsverfolgung/sendungsinformation#{}"),
("dpd", "DPD", [r"\d{14}", r"\d{27,28}"], "https://tracking.dpd.de/status/de_DE/parcel/{}"),
("gls", "GLS", [r"\d{11,12}", r"[A-Z0-9]{8}"], "https://gls-group.com/DE/de/paketverfolgung?match={}"),
("fedex", "FedEx", [r"\d{12}", r"\d{15}"], "https://www.fedex.com/fedextrack/?trknbr={}"),
("amazon", "Amazon Logistics", [r"TBA\d{9,}"], "https://www.amazon.de/gp/your-account/order-history"),
("upu", "Internationale Post (UPU)", [r"[A-Z]{2}\d{9}(?!DE)[A-Z]{2}"], None),
]
_COMPILED = [(cid, name, [re.compile(p) for p in patterns], link) for cid, name, patterns, link in CARRIERS]
def detect_carriers(number: str) -> list[dict]:
result = []
for cid, name, patterns, link in _COMPILED:
if any(p.fullmatch(number) for p in patterns):
url = link.format(quote(number)) if link else universal_link(number)
result.append({"id": cid, "name": name, "link": url})
return result
backend/app/tracking/dhl.py
+64
View File
@@ -0,0 +1,64 @@
"""DHL Shipment Tracking – Unified API.
Kostenloser Key: https://developer.dhl.com (API "Shipment Tracking - Unified")
"""
import httpx
API = "https://api-eu.dhl.com/track/shipments"
STATUS = {
"pre-transit": "inforeceived",
"transit": "intransit",
"delivered": "delivered",
"failure": "failed",
"unknown": "unknown",
}
def _place(loc: dict | None) -> str | None:
addr = (loc or {}).get("address")
if not addr:
return None
return ", ".join(filter(None, [addr.get("addressLocality"), addr.get("countryCode")])) or None
async def track_dhl(number: str, key: str) -> dict | None:
async with httpx.AsyncClient(timeout=15) as client:
res = await client.get(
API,
params={"trackingNumber": number, "language": "de"},
headers={"DHL-API-Key": key, "Accept": "application/json"},
)
if res.status_code == 404:
return None # keine DHL-Sendung -> nächsten Anbieter probieren
res.raise_for_status()
shipments = res.json().get("shipments") or []
if not shipments:
return None
s = shipments[0]
status = s.get("status") or {}
events = [
{"time": e.get("timestamp"), "description": e.get("description") or e.get("status"), "location": _place(e.get("location"))}
for e in s.get("events") or []
]
eta = None
if s.get("estimatedTimeOfDelivery"):
eta = {"from": s["estimatedTimeOfDelivery"], "to": None}
elif s.get("estimatedDeliveryTimeFrame"):
frame = s["estimatedDeliveryTimeFrame"]
eta = {"from": frame.get("estimatedFrom"), "to": frame.get("estimatedThrough")}
return {
"source": "DHL API",
"carrier": "DHL Express" if s.get("service") == "express" else "DHL",
"status": STATUS.get(status.get("statusCode"), "unknown"),
"statusDetail": status.get("description") or status.get("status"),
"eta": eta,
"deliveredAt": status.get("timestamp") if status.get("statusCode") == "delivered" else None,
"origin": _place(s.get("origin")),
"destination": _place(s.get("destination")),
"events": events,
}
backend/app/tracking/seventeen.py
+80
View File
@@ -0,0 +1,80 @@
"""17TRACK Tracking API v2.2 – https://api.17track.net
Nummern müssen erst registriert werden, danach lassen sich die Daten abrufen.
Bei neu registrierten Nummern dauert es oft ein paar Sekunden, bis Daten da sind.
"""
import httpx
API = "https://api.17track.net/track/v2.2"
ALREADY_REGISTERED = -18019901
STATUS = {
"NotFound": "notfound",
"InfoReceived": "inforeceived",
"InTransit": "intransit",
"Expired": "expired",
"AvailableForPickup": "pickup",
"OutForDelivery": "outfordelivery",
"DeliveryFailure": "failed",
"Delivered": "delivered",
"Exception": "exception",
}
async def _call(client: httpx.AsyncClient, endpoint: str, key: str, body: list) -> dict:
res = await client.post(f"{API}/{endpoint}", headers={"17token": key}, json=body)
res.raise_for_status()
data = res.json()
if data.get("code") != 0:
raise RuntimeError(f"API-Code {data.get('code')}")
return data["data"]
def _location(event: dict) -> str | None:
if event.get("location"):
return event["location"]
addr = event.get("address") or {}
return ", ".join(filter(None, [addr.get("city"), addr.get("country")])) or None
async def track_17(number: str, key: str) -> dict:
async with httpx.AsyncClient(timeout=15) as client:
data = await _call(client, "gettrackinfo", key, [{"number": number}])
if not data.get("accepted"):
# Noch nicht registriert -> registrieren (17TRACK erkennt den Paketdienst selbst)
reg = await _call(client, "register", key, [{"number": number}])
rejected = (reg.get("rejected") or [None])[0]
if rejected and (rejected.get("error") or {}).get("code") != ALREADY_REGISTERED:
raise RuntimeError((rejected.get("error") or {}).get("message") or "Registrierung abgelehnt")
data = await _call(client, "gettrackinfo", key, [{"number": number}])
item = (data.get("accepted") or [None])[0]
if not item or not item.get("track_info"):
return {"source": "17TRACK", "status": "pending"}
info = item["track_info"]
provider = ((info.get("tracking") or {}).get("providers") or [{}])[0]
events = [
{
"time": e.get("time_iso") or e.get("time_utc"),
"description": e.get("description"),
"location": _location(e),
}
for e in provider.get("events") or []
]
latest = info.get("latest_status") or {}
eta = (info.get("time_metrics") or {}).get("estimated_delivery_date") or {}
shipping = info.get("shipping_info") or {}
return {
"source": "17TRACK",
"carrier": (provider.get("provider") or {}).get("name"),
"status": STATUS.get(latest.get("status"), "unknown"),
"statusDetail": latest.get("sub_status"),
"eta": {"from": eta.get("from"), "to": eta.get("to")} if eta.get("from") or eta.get("to") else None,
"deliveredAt": events[0]["time"] if latest.get("status") == "Delivered" and events else None,
"origin": (shipping.get("shipper_address") or {}).get("country"),
"destination": (shipping.get("recipient_address") or {}).get("country"),
"events": events,
}
backend/requirements.txt
+4
View File
@@ -0,0 +1,4 @@
fastapi==0.143.0
uvicorn[standard]==0.54.0
httpx==0.28.1
nh3==0.3.7