# Copyright (c) 2025 Patrick Motsch # All rights reserved. """ UAM models: User, Mandate, UserConnection. Multi-Tenant Design: - User gehört NICHT direkt zu einem Mandanten - Zugehörigkeit wird über UserMandate gesteuert (siehe datamodelMembership.py) - Zwei orthogonale Plattform-Autoritäts-Flags: * isSysAdmin → Infrastruktur-Operator (Logs, Tokens, DB-Health, i18n-Master, Registry). RBAC-Engine-Bypass. KEIN Cross-Mandate-Governance. * isPlatformAdmin → Cross-Mandate-Governance (User-/Mandate-/RBAC-/ Feature-Verwaltung über alle Mandanten). KEIN RBAC-Bypass. Beide einzeln vergebbar, einzeln auditierbar. Siehe wiki/c-work/4-done/2026-04-sysadmin-authority-split.md """ import uuid from typing import Optional, List, Dict, Any from enum import Enum from pydantic import BaseModel, Field, EmailStr, field_validator, computed_field from modules.datamodels.datamodelBase import PowerOnModel from modules.shared.i18nRegistry import i18nModel, normalizePrimaryLanguageTag from modules.shared.mandateNameUtils import MANDATE_NAME_MAX_LEN, MANDATE_NAME_MIN_LEN from modules.shared.timeUtils import getUtcTimestamp class AuthAuthority(str, Enum): LOCAL = "local" GOOGLE = "google" MSFT = "msft" CLICKUP = "clickup" INFOMANIAK = "infomaniak" class ConnectionStatus(str, Enum): ACTIVE = "active" EXPIRED = "expired" REVOKED = "revoked" PENDING = "pending" class AccessLevel(str, Enum): """Access level enumeration for RBAC""" ALL = "a" # All records MY = "m" # My records (created by me) GROUP = "g" # Group records (group context is the mandate) NONE = "n" # No access class UserPermissions(BaseModel): """User permissions model for RBAC""" view: bool = Field( default=False, description="View permission: if true, item is visible/enabled" ) read: AccessLevel = Field( default=AccessLevel.NONE, description="Read permission level" ) create: AccessLevel = Field( default=AccessLevel.NONE, description="Create permission level" ) update: AccessLevel = Field( default=AccessLevel.NONE, description="Update permission level" ) delete: AccessLevel = Field( default=AccessLevel.NONE, description="Delete permission level" ) class InvoiceAddress(BaseModel): """ Historische strukturierte Rechnungsadresse. NICHT MEHR aktiv verwendet -- die Felder sind seit 2026-04-20 als ``invoiceCompanyName`` / ``invoiceLine1`` / ``invoicePostalCode`` / ... direkt auf ``Mandate`` deklariert (siehe dort). Diese Klasse bleibt nur noch erhalten, falls Bestandscode irgendwo das Schema dokumentiert oder alte JSONB-Dicts serialisiert; sie wird vom Mandate-Modell nicht mehr referenziert. """ companyName: Optional[str] = Field( default=None, description="Firmenname / Empfaenger der Rechnung (falls abweichend vom Mandate.label)", ) contactName: Optional[str] = Field( default=None, description="Ansprechperson (z. B. Buchhaltung)", ) email: Optional[EmailStr] = Field( default=None, description="E-Mail-Adresse fuer den Versand der Stripe-Rechnung", ) line1: Optional[str] = Field( default=None, description="Strasse + Nr. (Adresszeile 1)", ) line2: Optional[str] = Field( default=None, description="Adresszeile 2 (z. B. c/o, Postfach)", ) postalCode: Optional[str] = Field( default=None, description="PLZ", ) city: Optional[str] = Field( default=None, description="Ort", ) state: Optional[str] = Field( default=None, description="Kanton / Bundesland", ) country: Optional[str] = Field( default="CH", description="ISO-3166 Alpha-2 Laendercode (Default: CH)", ) vatNumber: Optional[str] = Field( default=None, description="UID / MWST-Nummer des Empfaengers (z. B. CHE-123.456.789 MWST)", ) @i18nModel("Mandant") class Mandate(PowerOnModel): """ Mandate (Mandant/Tenant) model. Ein Mandant ist ein isolierter Bereich für Daten und Berechtigungen. Semantik: - ``name`` (Kurzzeichen): plattformweit eindeutiger, stabiler technischer Code (Slug), Audit-/Referenz-Identifier. Nur Kleinbuchstaben, Ziffern und ``-`` (Länge 2–32). - ``label`` (Voller Name): Anzeigename im UI, frei änderbar unabhängig vom Slug. """ id: str = Field( default_factory=lambda: str(uuid.uuid4()), description="Unique ID of the mandate", json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_visible": False, "frontend_required": False, "label": "ID"}, ) name: str = Field( description="Unique stable mandate code (slug); lowercase, digits, hyphen segments only.", min_length=MANDATE_NAME_MIN_LEN, max_length=MANDATE_NAME_MAX_LEN, pattern=r"^[a-z0-9]+(-[a-z0-9]+)*$", json_schema_extra={ "frontend_type": "slug", "frontend_readonly": False, "frontend_required": True, "label": "Kurzzeichen", }, ) label: str = Field( description="Human-readable mandate name shown in the UI (Voller Name).", min_length=1, json_schema_extra={ "frontend_type": "text", "frontend_readonly": False, "frontend_required": True, "label": "Voller Name", }, ) enabled: bool = Field( default=True, description="Indicates whether the mandate is enabled", json_schema_extra={ "frontend_type": "checkbox", "frontend_readonly": False, "frontend_required": False, "label": "Aktiviert", # Render boolean as i18n-translatable label tuple [true, neutral, false]. "frontend_format_labels": ["Ja", "-", "Nein"], }, ) isSystem: bool = Field( default=False, description="Whether this is a system mandate (e.g. root mandate). Cannot be deleted.", json_schema_extra={ "frontend_type": "checkbox", "frontend_readonly": True, "frontend_required": False, "label": "System-Mandant", "frontend_format_labels": ["Ja", "-", "Nein"], }, ) deletedAt: Optional[float] = Field( default=None, description="Timestamp when the mandate was soft-deleted. After 30 days, hard-delete is triggered.", json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False, "label": "Gelöscht am"}, ) # ------------------------------------------------------------------ # Rechnungsadresse (CH-Treuhand-konform, strukturiert) # ------------------------------------------------------------------ # Einzelne Felder statt eines nested Objekts/Freitexts, damit # (a) der FormGenerator sie automatisch als Eingabezeilen rendert, # (b) der Stripe-Checkout sie 1:1 in `customer.address`, # `customer.email`, `customer.tax_id_data` mappen kann # (Stripe verlangt die Adresse strukturiert, nicht als Freitext). # ``order`` 200-209 gruppiert die Felder visuell am Ende des Formulars. invoiceCompanyName: Optional[str] = Field( default=None, description="Firmenname / Empfaenger der Rechnung (falls abweichend vom Voller Name).", max_length=200, json_schema_extra={ "frontend_type": "text", "frontend_required": False, "label": "Rechnungsadresse - Firma", "order": 200, "placeholder": "Muster Treuhand AG", }, ) invoiceContactName: Optional[str] = Field( default=None, description="Ansprechperson z. H. (z. B. Buchhaltung).", max_length=200, json_schema_extra={ "frontend_type": "text", "frontend_required": False, "label": "Rechnungsadresse - z. H.", "order": 201, "placeholder": "Buchhaltung", }, ) invoiceEmail: Optional[str] = Field( default=None, description="E-Mail-Adresse fuer den Versand der Stripe-Rechnung.", max_length=254, json_schema_extra={ "frontend_type": "email", "frontend_required": False, "label": "Rechnungsadresse - E-Mail", "order": 202, "placeholder": "rechnungen@firma.ch", }, ) invoiceLine1: Optional[str] = Field( default=None, description="Adresszeile 1 (Strasse + Nr.). Pflichtfeld fuer Stripe-Customer-Adresse.", max_length=200, json_schema_extra={ "frontend_type": "text", "frontend_required": False, "label": "Rechnungsadresse - Strasse + Nr.", "order": 203, "placeholder": "Bahnhofstrasse 1", }, ) invoiceLine2: Optional[str] = Field( default=None, description="Adresszeile 2 (z. B. c/o, Postfach).", max_length=200, json_schema_extra={ "frontend_type": "text", "frontend_required": False, "label": "Rechnungsadresse - Adresszusatz", "order": 204, "placeholder": "c/o Buchhaltung", }, ) invoicePostalCode: Optional[str] = Field( default=None, description="PLZ.", max_length=20, json_schema_extra={ "frontend_type": "text", "frontend_required": False, "label": "Rechnungsadresse - PLZ", "order": 205, "placeholder": "8000", }, ) invoiceCity: Optional[str] = Field( default=None, description="Ort.", max_length=100, json_schema_extra={ "frontend_type": "text", "frontend_required": False, "label": "Rechnungsadresse - Ort", "order": 206, "placeholder": "Zuerich", }, ) invoiceState: Optional[str] = Field( default=None, description="Kanton / Bundesland (optional).", max_length=100, json_schema_extra={ "frontend_type": "text", "frontend_required": False, "label": "Rechnungsadresse - Kanton", "order": 207, "placeholder": "ZH", }, ) invoiceCountry: Optional[str] = Field( default="CH", description="ISO-3166 Alpha-2 Laendercode (Default: CH).", max_length=2, pattern=r"^[A-Z]{2}$", json_schema_extra={ "frontend_type": "text", "frontend_required": False, "label": "Rechnungsadresse - Land (ISO)", "order": 208, "placeholder": "CH", }, ) invoiceVatNumber: Optional[str] = Field( default=None, description="UID / MWST-Nummer des Empfaengers (z. B. CHE-123.456.789 MWST). Wird Stripe als `tax_id_data` mitgegeben.", max_length=50, json_schema_extra={ "frontend_type": "text", "frontend_required": False, "label": "Rechnungsadresse - UID-Nr.", "order": 209, "placeholder": "CHE-123.456.789 MWST", }, ) @field_validator( "invoiceCompanyName", "invoiceContactName", "invoiceEmail", "invoiceLine1", "invoiceLine2", "invoicePostalCode", "invoiceCity", "invoiceState", "invoiceVatNumber", mode="before", ) @classmethod def _coerceInvoiceTextField(cls, v): """Trim incoming address strings; treat empty as ``None``.""" if v is None: return None if isinstance(v, str): trimmed = v.strip() return trimmed or None return v @field_validator("invoiceCountry", mode="before") @classmethod def _coerceInvoiceCountry(cls, v): """Normalize country code: trim, upper-case, empty -> default ``CH``.""" if v is None: return "CH" if isinstance(v, str): trimmed = v.strip().upper() return trimmed or "CH" return v @field_validator('isSystem', mode='before') @classmethod def _coerceIsSystem(cls, v): """Coerce None to False (for existing DB records without isSystem field).""" if v is None: return False return v @field_validator("name", mode="before") @classmethod def _stripName(cls, v): if v is None: return "" if isinstance(v, str): return v.strip() return v @field_validator("label", mode="before") @classmethod def _coerceLabel(cls, v): if v is None: return "" return v @field_validator("label") @classmethod def _validateMandateLabel(cls, v: str) -> str: s = v.strip() if len(s) < 1: raise ValueError("Mandate Voller Name (label) must not be empty.") return s @i18nModel("Benutzerverbindung") class UserConnection(PowerOnModel): id: str = Field( default_factory=lambda: str(uuid.uuid4()), description="Unique ID of the connection", json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False, "label": "ID"}, ) userId: str = Field( description="ID of the user this connection belongs to", json_schema_extra={ "frontend_type": "text", "frontend_readonly": True, "frontend_required": False, "label": "Benutzer-ID", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}, }, ) authority: AuthAuthority = Field( description="Authentication authority", json_schema_extra={ "frontend_type": "select", "frontend_readonly": True, "frontend_required": False, "frontend_options": "/api/connections/authorities/options", "label": "Autorität", }, ) externalId: str = Field( description="User ID in the external system", json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": False, "label": "Externe ID"}, ) externalUsername: str = Field( description="Username in the external system", json_schema_extra={"frontend_type": "text", "frontend_readonly": False, "frontend_required": False, "label": "Externer Benutzername"}, ) externalEmail: Optional[EmailStr] = Field( None, description="Email in the external system", json_schema_extra={"frontend_type": "email", "frontend_readonly": False, "frontend_required": False, "label": "Externe E-Mail"}, ) status: ConnectionStatus = Field( default=ConnectionStatus.ACTIVE, description="Connection status", json_schema_extra={ "frontend_type": "select", "frontend_readonly": False, "frontend_required": False, "frontend_options": "/api/connections/statuses/options", "label": "Status", }, ) connectedAt: float = Field( default_factory=getUtcTimestamp, description="When the connection was established (UTC timestamp in seconds)", json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False, "label": "Verbunden am"}, ) lastChecked: float = Field( default_factory=getUtcTimestamp, description="When the connection was last verified (UTC timestamp in seconds)", json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False, "label": "Zuletzt geprüft"}, ) expiresAt: Optional[float] = Field( None, description="When the connection expires (UTC timestamp in seconds)", json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False, "label": "Läuft ab am"}, ) tokenStatus: Optional[str] = Field( None, description="Current token status: active, expired, none", json_schema_extra={ "frontend_type": "select", "frontend_readonly": True, "frontend_required": False, "frontend_options": [ {"value": "active", "label": "Active"}, {"value": "expired", "label": "Expired"}, {"value": "none", "label": "None"}, ], "label": "Verbindungsstatus", }, ) tokenExpiresAt: Optional[float] = Field( None, description="When the current token expires (UTC timestamp in seconds)", json_schema_extra={"frontend_type": "timestamp", "frontend_readonly": True, "frontend_required": False, "label": "Token läuft ab am"}, ) grantedScopes: Optional[List[str]] = Field( None, description="OAuth scopes granted for this connection", json_schema_extra={"frontend_type": "list", "frontend_readonly": True, "frontend_required": False, "label": "Gewährte Berechtigungen"}, ) @computed_field @property def connectionReference(self) -> str: """Generate connection reference string in format: connection:{authority}:{username}""" return f"connection:{self.authority.value}:{self.externalUsername}" @computed_field @property def displayLabel(self) -> str: """Human-readable label for display in dropdowns""" authorityLabels = { "msft": "Microsoft", "google": "Google", "local": "Local", "clickup": "ClickUp", } return f"{authorityLabels.get(self.authority.value, self.authority.value)}: {self.externalUsername}" @i18nModel("Benutzer") class User(PowerOnModel): """ User model. Multi-Tenant Design: - User gehört NICHT direkt zu einem Mandanten - Zugehörigkeit wird über UserMandate gesteuert (siehe datamodelMembership.py) - Rollen werden über UserMandateRole gesteuert (mandanten-scoped) - Plattform-Autorität via zwei orthogonalen Flags: * isSysAdmin → Infrastruktur (Bypass der RBAC-Engine, KEIN Cross-Mandate-Governance) * isPlatformAdmin → Cross-Mandate-Governance (KEIN RBAC-Bypass) """ id: str = Field( default_factory=lambda: str(uuid.uuid4()), description="Unique ID of the user", json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_visible": False, "frontend_required": False, "label": "ID"}, ) username: str = Field( description="Username for login (immutable after creation)", json_schema_extra={"frontend_type": "text", "frontend_readonly": True, "frontend_required": True, "label": "Benutzername"}, ) email: Optional[EmailStr] = Field( default=None, description="Email address of the user", json_schema_extra={"frontend_type": "email", "frontend_readonly": False, "frontend_required": True, "label": "E-Mail"}, ) fullName: Optional[str] = Field( default=None, description="Full name of the user", json_schema_extra={"frontend_type": "text", "frontend_readonly": False, "frontend_required": False, "label": "Vollständiger Name"}, ) language: str = Field( default="de", description="Preferred UI language code (must exist as UiLanguageSet).", json_schema_extra={ "frontend_type": "select", "frontend_readonly": False, "frontend_required": True, "frontend_options": "/api/i18n/codes", "label": "Sprache", }, ) @field_validator('language', mode='before') @classmethod def _normalizeLanguage(cls, v): """Normalize to primary language subtag (2–8 letters); default remains ``de``.""" if v is None: return "de" langMap = { 'english': 'en', 'englisch': 'en', 'german': 'de', 'deutsch': 'de', 'french': 'fr', 'französisch': 'fr', 'francais': 'fr', 'italian': 'it', 'italienisch': 'it', 'italiano': 'it', } normalized = str(v).lower().strip() if normalized in langMap: return langMap[normalized] return normalizePrimaryLanguageTag(normalized, "de") enabled: bool = Field( default=True, description="Indicates whether the user is enabled", json_schema_extra={ "frontend_type": "checkbox", "frontend_readonly": False, "frontend_required": False, "label": "Aktiviert", "frontend_format_labels": ["Ja", "-", "Nein"], }, ) isSysAdmin: bool = Field( default=False, description=( "Infrastructure/System Operator flag. Erlaubt RBAC-Engine-Bypass " "und Zugriff auf Infrastruktur-Operationen (Logs, Tokens, DB-Health, " "i18n-Master, Registry). Gibt KEIN Cross-Mandate-Governance-Recht " "(dafür ist isPlatformAdmin zuständig)." ), json_schema_extra={"frontend_type": "checkbox", "frontend_readonly": False, "frontend_required": False, "label": "System-Admin"}, ) @field_validator('isSysAdmin', mode='before') @classmethod def _coerceIsSysAdmin(cls, v): """Konvertiert None zu False (für bestehende DB-Einträge ohne isSysAdmin Feld).""" if v is None: return False return v isPlatformAdmin: bool = Field( default=False, description=( "Platform/Cross-Mandate Governance flag. Erlaubt mandanten-übergreifende " "Verwaltungsoperationen (User-/Mandate-/RBAC-/Feature-Registry). " "KEIN RBAC-Engine-Bypass und KEIN impliziter Zugriff auf Mandanten-Daten." ), json_schema_extra={"frontend_type": "checkbox", "frontend_readonly": False, "frontend_required": False, "label": "Plattform-Admin"}, ) @field_validator('isPlatformAdmin', mode='before') @classmethod def _coerceIsPlatformAdmin(cls, v): """Konvertiert None zu False (für bestehende DB-Einträge ohne isPlatformAdmin Feld).""" if v is None: return False return v authenticationAuthority: AuthAuthority = Field( default=AuthAuthority.LOCAL, description="Primary authentication authority", json_schema_extra={ "frontend_type": "select", "frontend_readonly": True, "frontend_required": False, "frontend_options": "/api/connections/authorities/options", "label": "Authentifizierung", }, ) roleLabels: List[str] = Field( default_factory=list, description="Role labels (from DB or enriched when loading users)", json_schema_extra={ "frontend_type": "multiselect", "frontend_readonly": True, "frontend_visible": False, "frontend_required": False, "label": "Rollen-Labels", }, ) @i18nModel("Benutzerzugang") class UserInDB(User): """User model with password hash for database storage.""" hashedPassword: Optional[str] = Field( None, description="Hash of the user password", json_schema_extra={"label": "Passwort-Hash"}, ) resetToken: Optional[str] = Field( None, description="Password reset token (UUID)", json_schema_extra={"label": "Reset-Token"}, ) resetTokenExpires: Optional[float] = Field( None, description="Reset token expiration (UTC timestamp in seconds)", json_schema_extra={"label": "Token läuft ab", "frontend_type": "timestamp"}, ) def normalizeTtsVoiceMap(value: Any) -> Optional[Dict[str, str]]: """ Coerce ttsVoiceMap payloads to Dict[str, str]. UI/clients may send per-locale objects like {"voiceName": "de-DE-Chirp3-HD-Achird"}; storage and model field type are locale -> voice id string. """ if value is None: return None if not isinstance(value, dict): return None out: Dict[str, str] = {} for rawKey, rawVal in value.items(): key = str(rawKey) if rawVal is None: continue if isinstance(rawVal, str): out[key] = rawVal elif isinstance(rawVal, dict): vn = rawVal.get("voiceName") if vn is not None and str(vn).strip() != "": out[key] = str(vn).strip() else: out[key] = str(rawVal) return out if out else None @i18nModel("Spracheinstellungen") class UserVoicePreferences(PowerOnModel): """User-level voice/language preferences, shared across all features.""" id: str = Field( default_factory=lambda: str(uuid.uuid4()), description="Primary key", json_schema_extra={"label": "ID"}, ) userId: str = Field( description="User ID", json_schema_extra={"label": "Benutzer-ID", "fk_target": {"db": "poweron_app", "table": "UserInDB", "labelField": "username"}}, ) mandateId: Optional[str] = Field( default=None, description="Mandate scope (None = global for user)", json_schema_extra={"label": "Mandanten-ID", "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"}}, ) sttLanguage: str = Field( default="de-DE", description="Speech-to-text language code", json_schema_extra={"label": "STT-Sprache"}, ) ttsLanguage: str = Field( default="de-DE", description="Text-to-speech language code", json_schema_extra={"label": "TTS-Sprache"}, ) ttsVoice: Optional[str] = Field( default=None, description="Preferred TTS voice identifier", json_schema_extra={"label": "TTS-Stimme"}, ) ttsVoiceMap: Optional[Dict[str, str]] = Field( default=None, description="Language-to-voice mapping", json_schema_extra={"label": "Stimmen-Zuordnung"}, ) translationSourceLanguage: Optional[str] = Field( default=None, description="Source language for translations", json_schema_extra={"label": "Übersetzung Quelle"}, ) translationTargetLanguage: Optional[str] = Field( default=None, description="Target language for translations", json_schema_extra={"label": "Übersetzung Ziel"}, ) @field_validator("ttsVoiceMap", mode="before") @classmethod def _validateTtsVoiceMap(cls, value: Any) -> Optional[Dict[str, str]]: return normalizeTtsVoiceMap(value)