gateway/modules/datamodels/datamodelUtils.py

# 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,
            "fk_target": {"db": "poweron_app", "table": "Mandate", "labelField": "label"},
        },
    )
    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("—")