guyg1474andClaude Opus 5.5
4aba48ce51
- Eine ValidationError-Klasse statt zwei (links.py, projects.py), ein Fehler-Handler in main.py - clean_text() ersetzt vier fast gleiche Text-Prüfungen (Kanban, Haushalt, Reisen, Artikel); Fehlermeldungen nennen jetzt überall die Höchstlänge - Zeitstempel (now_iso, iso_ago, iso_after) und Kontenliste (user_list) an einer Stelle; ungenutzte Importe entfernt - Tests für die Helfer und für einheitliche 400-Fehler aus allen Bereichen Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
62 lines
2.0 KiB
Python
62 lines
2.0 KiB
Python
"""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 .common import ValidationError
|
||
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:.]+\]")
|
||
|
||
|
||
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"])
|