gateway/modules/datamodels/datamodelUtils.py
2026-04-12 00:29:00 +02:00

140 lines
5.4 KiB
Python

# Copyright (c) 2025 Patrick Motsch
# All rights reserved.
"""Utility datamodels: Prompt, TextMultilingual."""
import json
from typing import Any, Dict
from pydantic import BaseModel, Field, field_validator, model_validator
from modules.datamodels.datamodelBase import PowerOnModel
from modules.shared.i18nRegistry import i18nModel
import uuid
@i18nModel("Prompt")
class Prompt(PowerOnModel):
"""Benutzer- oder System-Prompt fuer die KI."""
id: str = Field(
default_factory=lambda: str(uuid.uuid4()),
description="Primary key",
json_schema_extra={"label": "ID", "frontend_type": "text", "frontend_readonly": True, "frontend_required": False},
)
mandateId: str = Field(
default="",
description="ID of the mandate this prompt belongs to",
json_schema_extra={"label": "Mandanten-ID", "frontend_type": "text", "frontend_readonly": True, "frontend_required": False},
)
isSystem: bool = Field(
default=False,
description="System prompt visible to all users (read-only for non-SysAdmin)",
json_schema_extra={"label": "System", "frontend_type": "boolean", "frontend_readonly": True, "frontend_required": False},
)
content: str = Field(
description="Content of the prompt",
json_schema_extra={"label": "Inhalt", "frontend_type": "textarea", "frontend_readonly": False, "frontend_required": True},
)
name: str = Field(
description="Name of the prompt",
json_schema_extra={"label": "Name", "frontend_type": "text", "frontend_readonly": False, "frontend_required": True},
)
@field_validator('isSystem', mode='before')
@classmethod
def _coerceIsSystem(cls, v):
if v is None:
return False
return v
class TextMultilingual(BaseModel):
"""Multilingual text field stored as JSONB: {"xx": "source text", "de": "...", "en": "...", ...}.
- xx = source/default text (required). Same role as xx in the UI i18n system.
- All language codes (de, en, fr, ...) are dynamic, populated via batch translation.
- No hardcoded language fields. The DB column is JSONB with arbitrary keys.
"""
model_config = {"extra": "allow"}
xx: str = Field(description="Source/default text (required)")
@model_validator(mode='before')
@classmethod
def _ensureXx(cls, data: Any) -> Any:
"""Derive xx from existing language keys when missing (legacy DB rows)."""
if not isinstance(data, dict):
return data
if data.get('xx') and isinstance(data['xx'], str) and data['xx'].strip():
return data
fallback = data.get('de') or data.get('en')
if not fallback or not isinstance(fallback, str) or not fallback.strip():
for v in data.values():
if v and isinstance(v, str) and v.strip():
fallback = v
break
data['xx'] = fallback.strip() if fallback and isinstance(fallback, str) else ''
return data
@field_validator('xx')
@classmethod
def _validateXxRequired(cls, v):
if not v or not v.strip():
raise ValueError("Source text (xx) is required and cannot be empty")
return v
def model_dump(self, **kwargs) -> Dict[str, str]:
result = {"xx": self.xx}
if self.__pydantic_extra__:
for k, v in self.__pydantic_extra__.items():
if v is not None and isinstance(v, str):
result[k] = v
return result
@classmethod
def from_dict(cls, data: Dict[str, str]) -> 'TextMultilingual':
cleaned = {k: v for k, v in data.items() if v is not None and isinstance(v, str)}
if not cleaned.get('xx'):
cleaned['xx'] = cleaned.get('de') or next((v for v in cleaned.values() if v), '')
return cls(**cleaned)
def get_text(self, lang: str = 'de') -> str:
"""Get text for a language. Falls back to xx (source text)."""
if lang == 'xx':
return self.xx
extra = self.__pydantic_extra__ or {}
value = extra.get(lang)
if value and isinstance(value, str):
return value
return self.xx
@classmethod
def fromUniform(cls, text: str) -> "TextMultilingual":
"""Create with source text only. Languages are populated by batch translation."""
t = text.strip()
if not t:
raise ValueError("Text must be non-empty")
return cls(xx=t)
def coerce_text_multilingual(val: Any) -> TextMultilingual:
"""Normalize str, dict, or TextMultilingual into a valid TextMultilingual instance."""
if isinstance(val, TextMultilingual):
return val
if isinstance(val, dict):
if not val:
return TextMultilingual.fromUniform("")
cleaned = {k: v for k, v in val.items() if v is not None and isinstance(v, str)}
if not cleaned.get("xx"):
cleaned["xx"] = cleaned.get("de") or next((v for v in cleaned.values() if v), "")
return TextMultilingual(**cleaned)
if isinstance(val, str) and val.strip():
s = val.strip()
if s.startswith("{") and s.endswith("}"):
try:
parsed = json.loads(s)
if isinstance(parsed, dict):
return coerce_text_multilingual(parsed)
except json.JSONDecodeError:
pass
return TextMultilingual.fromUniform(s)
return TextMultilingual.fromUniform("")