my_voice_assistant_v3_jamulix/app/schemas.py

import re
from datetime import date
from typing import Literal

from pydantic import BaseModel, Field, field_validator


class EndpointCapabilities(BaseModel):
    id: str
    kind: str
    direction: Literal["input", "output"]
    sample_rate: int = 16000
    channels: int = 1
    latency_class: Literal["low", "medium", "high"] = "medium"
    supports_aec: bool = False
    supports_barge_in: bool = False
    networked: bool = False
    bluetooth: bool = False
    mobile: bool = False
    default: bool = False


class AudioChunk(BaseModel):
    data: bytes
    sample_rate: int = 16000
    channels: int = 1
    format: str = "wav"
    timestamp_ms: int = 0


class PipelineTrace(BaseModel):
    raw_transcript: str | None = None
    cleaned_transcript: str | None = None
    semantic_response: str | None = None
    spoken_response: str | None = None
    tts_ready_text: str | None = None


class SpeakRequest(BaseModel):
    text: str = Field(min_length=1)
    voice: str | None = None
    language: str | None = None
    output_endpoint: str | None = None
    tts_provider: str | None = None


class ChatRequest(BaseModel):
    text: str = Field(min_length=1)
    input_endpoint: str | None = None
    output_endpoint: str | None = None
    language: str | None = None
    voice: str | None = None
    stt_provider: str | None = None
    llm_provider: str | None = None
    tts_provider: str | None = None
    text_only: bool | None = None  # True -> kein Server-Audio (Geräte-TTS spricht selbst)


class SessionRouteRequest(BaseModel):
    input_endpoint: str | None = None
    output_endpoint: str | None = None
    stt_provider: str | None = None
    llm_provider: str | None = None
    tts_provider: str | None = None
    language: str | None = None


class RouteInfo(BaseModel):
    input_endpoint: str
    output_endpoint: str
    stt_provider: str
    llm_provider: str
    tts_provider: str
    language: str


class UserCreate(BaseModel):
    display_name: str = Field(min_length=1)


class UserCreated(BaseModel):
    user_id: str
    display_name: str
    token: str  # nur bei Erstellung sichtbar


class UserUpdate(BaseModel):
    display_name: str = Field(min_length=1)


class UserPrefs(BaseModel):
    input_endpoint: str | None = None
    output_endpoint: str | None = None
    stt_provider: str | None = None
    llm_provider: str | None = None
    tts_provider: str | None = None
    language: str | None = None


class MemoryCreate(BaseModel):
    content: str = Field(min_length=1)


class MemoryOut(BaseModel):
    id: int
    content: str
    created_at: str



class GeoFix(BaseModel):
    """Geoposition des auslösenden Geräts (One-Shot, vom Browser erfasst)."""
    lat: float
    lon: float
    accuracy: float | None = None   # geschätzte Genauigkeit in Metern
    ts: str | None = None           # ISO-Zeitstempel des Fixes


class DeviceInfo(BaseModel):
    """Geräteart/Plattform des auslösenden Geräts (aus Browser-Daten)."""
    mobile: bool | None = None
    platform: str | None = None


class EmergencyRequest(BaseModel):
    """Vom Nutzer ausgelöster Notruf (Knopf). language für den Hinweistext."""
    language: str | None = None
    geo: GeoFix | None = None
    geo_error: str | None = None      # Grund, falls die Geo-Erfassung scheiterte
    device: DeviceInfo | None = None


# Format-Validierung der Notfall-Profilfelder (spiegelt die Frontend-Prüfung).
_PLZ_RE = re.compile(r"^\d{5}$")
_PHONE_RE = re.compile(r"^\+?[\d\s/().\-]{6,20}$")
_EMAIL_RE = re.compile(r"^[^\s@]+@[^\s@]+\.[^\s@]+$")
_DATE_RE = re.compile(r"^(\d{2})\.(\d{2})\.(\d{4})$")


def _valid_german_date(v: str) -> bool:
    m = _DATE_RE.match(v)
    if not m:
        return False
    d, mo, y = int(m.group(1)), int(m.group(2)), int(m.group(3))
    if not (1900 <= y <= 2100):
        return False
    try:
        date(y, mo, d)
    except ValueError:
        return False
    return True


def _csv_all(v: str, rx: "re.Pattern") -> bool:
    return all(rx.match(p.strip()) for p in v.split(",") if p.strip())


class AdminUserPrefsUpdate(BaseModel):
    """Admin setzt Nutzer-Einstellungen: erlaubte Sprachen + Notfall-Profil.

    Optionale Felder dürfen leer/None sein; nicht-leere Werte werden auf ihr
    Format geprüft (serverseitige Härtung zusätzlich zur Frontend-Validierung).
    """
    allowed_languages: str | None = None
    # Notfall-Profil (in user.prefs gespeichert, fließt in die Notruf-Meldungen ein)
    full_name: str | None = None
    street: str | None = None
    postal_code: str | None = None
    city: str | None = None
    phone_landline: str | None = None
    phone_mobile: str | None = None
    birth_date: str | None = None
    medical_notes: str | None = None
    emergency_contacts: str | None = None   # Kontakt-E-Mail(s), CSV
    emergency_phones: str | None = None     # Kontakt-Telefon(e), CSV
    location_consent: bool | None = None    # Standort im Notfall mitsenden?
    # Alarmpause pro Nutzer (ganze Minuten, als String). Leer = globaler Standard,
    # "0" = aus. Präzedenz: dieser Wert > globale Einstellung > 5.
    emergency_cooldown_minutes: str | None = None

    @field_validator("postal_code")
    @classmethod
    def _check_plz(cls, v):
        if v and not _PLZ_RE.match(v.strip()):
            raise ValueError("PLZ muss genau 5 Ziffern sein")
        return v

    @field_validator("phone_mobile", "phone_landline")
    @classmethod
    def _check_phone(cls, v):
        if v and not _PHONE_RE.match(v.strip()):
            raise ValueError("ungültige Telefonnummer")
        return v

    @field_validator("birth_date")
    @classmethod
    def _check_date(cls, v):
        if v and not _valid_german_date(v.strip()):
            raise ValueError("Geburtsdatum muss TT.MM.JJJJ (gültiges Datum) sein")
        return v

    @field_validator("emergency_contacts")
    @classmethod
    def _check_emails(cls, v):
        if v and not _csv_all(v, _EMAIL_RE):
            raise ValueError("Kontakt-E-Mail(s): ungültige Adresse")
        return v

    @field_validator("emergency_phones")
    @classmethod
    def _check_phones_csv(cls, v):
        if v and not _csv_all(v, _PHONE_RE):
            raise ValueError("Kontakt-Telefon(e): ungültige Nummer")
        return v

    @field_validator("emergency_cooldown_minutes")
    @classmethod
    def _check_cooldown(cls, v):
        if v is None or str(v).strip() == "":
            return v   # leer = globaler Standard (erben)
        s = str(v).strip()
        if not s.isdigit():
            raise ValueError("Pause: nur ganze Minuten (0–60)")
        if int(s) > 60:
            raise ValueError("Pause: höchstens 60 Minuten")
        return s


class UserAdminUpdate(BaseModel):
    """Admin schaltet Admin-Rechte für einen Nutzer ein/aus."""
    is_admin: bool