from fastapi import APIRouter, HTTPException, Depends, Body, Query, Path from typing import List, Dict, Any, Optional from fastapi import status from datetime import datetime # Import auth module from modules.auth import get_current_active_user, get_user_context # Import interfaces from modules.lucydom_interface import get_lucydom_interface from modules.lucydom_model import Prompt # Alle Attribute des Models ermitteln (außer interne/spezielle Attribute) def get_model_attributes(model_class): return [attr for attr in dir(model_class) if not callable(getattr(model_class, attr)) and not attr.startswith('_') and attr != 'metadata' and attr != 'query' and attr != 'query_class' and attr != 'label' and attr != 'field_labels'] # Modell-Attribute für Prompt prompt_attributes = get_model_attributes(Prompt) # Router für Prompt-Endpunkte erstellen router = APIRouter( prefix="/api/prompts", tags=["Prompts"], responses={404: {"description": "Not found"}} ) @router.get("", response_model=List[Dict[str, Any]]) async def get_prompts( current_user: Dict[str, Any] = Depends(get_current_active_user) ): """Alle Prompts abrufen""" mandate_id, user_id = await get_user_context(current_user) # LucyDOM-Interface mit Benutzerkontext initialisieren lucy_interface = get_lucydom_interface(mandate_id, user_id) # Prompts generisch abrufen return lucy_interface.get_all_prompts() @router.post("", response_model=Dict[str, Any]) async def create_prompt( prompt: Dict[str, Any] = Body(...), current_user: Dict[str, Any] = Depends(get_current_active_user) ): """Einen neuen Prompt erstellen""" mandate_id, user_id = await get_user_context(current_user) # LucyDOM-Interface mit Benutzerkontext initialisieren lucy_interface = get_lucydom_interface(mandate_id, user_id) # Attribute aus dem Request dynamisch setzen prompt_data = {} for attr in prompt_attributes: if attr in prompt: prompt_data[attr] = prompt[attr] # Pflichtfelder mit Standardwerten content = prompt.get("content", "") name = prompt.get("name", "Neuer Prompt") # Prompt erstellen new_prompt = lucy_interface.create_prompt( content=content, name=name ) # Aktuelle Zeit für created_at setzen, wenn es im Modell existiert if "created_at" in prompt_attributes and hasattr(new_prompt, "created_at"): new_prompt["created_at"] = datetime.now().isoformat() return new_prompt @router.get("/{prompt_id}", response_model=Dict[str, Any]) async def get_prompt( prompt_id: int, current_user: Dict[str, Any] = Depends(get_current_active_user) ): """Einen bestimmten Prompt abrufen""" mandate_id, user_id = await get_user_context(current_user) # LucyDOM-Interface mit Benutzerkontext initialisieren lucy_interface = get_lucydom_interface(mandate_id, user_id) # Prompt generisch abrufen prompt = lucy_interface.get_prompt(prompt_id) if not prompt: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Prompt mit ID {prompt_id} nicht gefunden" ) return prompt @router.put("/{prompt_id}", response_model=Dict[str, Any]) async def update_prompt( prompt_id: int, prompt_data: Dict[str, Any] = Body(...), current_user: Dict[str, Any] = Depends(get_current_active_user) ): """Einen bestehenden Prompt aktualisieren""" mandate_id, user_id = await get_user_context(current_user) # LucyDOM-Interface mit Benutzerkontext initialisieren lucy_interface = get_lucydom_interface(mandate_id, user_id) # Prüfe, ob der Prompt existiert existing_prompt = lucy_interface.get_prompt(prompt_id) if not existing_prompt: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Prompt mit ID {prompt_id} nicht gefunden" ) # Attribute aus dem Request dynamisch filtern update_data = {} for attr in prompt_attributes: if attr in prompt_data: update_data[attr] = prompt_data[attr] # Standardfelder für Update content = prompt_data.get("content") name = prompt_data.get("name") # Prompt aktualisieren updated_prompt = lucy_interface.update_prompt( prompt_id=prompt_id, content=content, name=name ) if not updated_prompt: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Fehler beim Aktualisieren des Prompts" ) return updated_prompt @router.delete("/{prompt_id}", response_model=Dict[str, Any]) async def delete_prompt( prompt_id: int, current_user: Dict[str, Any] = Depends(get_current_active_user) ): """Einen Prompt löschen""" mandate_id, user_id = await get_user_context(current_user) # LucyDOM-Interface mit Benutzerkontext initialisieren lucy_interface = get_lucydom_interface(mandate_id, user_id) # Prüfe, ob der Prompt existiert existing_prompt = lucy_interface.get_prompt(prompt_id) if not existing_prompt: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Prompt mit ID {prompt_id} nicht gefunden" ) success = lucy_interface.delete_prompt(prompt_id) if not success: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Fehler beim Löschen des Prompts" ) return {"message": f"Prompt mit ID {prompt_id} wurde erfolgreich gelöscht"}