# Copyright (c) 2025 Patrick Motsch # All rights reserved. """ Chatbot routes for the backend API. Implements simple chatbot endpoints using direct AI center calls via chatbot feature. """ import logging import json import asyncio import math from typing import Optional, Any, Dict, Union from fastapi import APIRouter, HTTPException, Depends, Body, Path, Query, Request, status from fastapi.responses import StreamingResponse from modules.shared.timeUtils import parseTimestamp # Import auth modules from modules.auth import limiter, getCurrentUser # Import interfaces import modules.interfaces.interfaceDbChatObjects as interfaceDbChatObjects from modules.interfaces.interfaceRbac import getRecordsetWithRBAC # Import models from modules.datamodels.datamodelChat import ChatWorkflow, UserInputRequest, WorkflowModeEnum from modules.datamodels.datamodelUam import User from modules.datamodels.datamodelPagination import PaginationParams, PaginatedResponse # Import chatbot feature from modules.features.chatbot import chatProcess from modules.features.chatbot.eventManager import get_event_manager # Import workflow control functions from modules.features.workflow import chatStop # Configure logger logger = logging.getLogger(__name__) # Create router for chatbot endpoints router = APIRouter( prefix="/api/chatbot", tags=["Chatbot"], responses={404: {"description": "Not found"}} ) def getServiceChat(currentUser: User): return interfaceDbChatObjects.getInterface(currentUser) # Chatbot streaming endpoint (SSE) @router.post("/start/stream") @limiter.limit("120/minute") async def stream_chatbot_start( request: Request, workflowId: Optional[str] = Query(None, description="Optional ID of the workflow to continue (can also be in request body)"), userInput: UserInputRequest = Body(...), currentUser: User = Depends(getCurrentUser) ) -> StreamingResponse: """ Starts a new chatbot workflow or continues an existing one with SSE streaming. Streams progress updates in real-time via Server-Sent Events. workflowId can be provided either: - As a query parameter: /api/chatbot/start/stream?workflowId=xxx - In the request body as part of UserInputRequest - Query parameter takes precedence if both are provided """ event_manager = get_event_manager() try: # Use workflowId from query parameter if provided, otherwise from request body final_workflow_id = workflowId or userInput.workflowId # Start background processing (this will create the workflow and event queue) workflow = await chatProcess(currentUser, userInput, final_workflow_id) # Get event queue for the workflow queue = event_manager.get_queue(workflow.id) if not queue: # Create queue if it doesn't exist queue = event_manager.create_queue(workflow.id) async def event_stream(): """Async generator for SSE events.""" try: # Get interface for status checks and chat data interfaceDbChat = getServiceChat(currentUser) # Send initial chat data (exact format as chatData endpoint) try: chatData = interfaceDbChat.getUnifiedChatData(workflow.id, None) if chatData.get("items"): for item in chatData["items"]: # Emit item directly in exact chatData format: {type, createdAt, item} yield f"data: {json.dumps(item)}\n\n" # Set initial timestamp for incremental fetching if chatData["items"]: timestamps = [parseTimestamp(item.get("createdAt"), default=0) for item in chatData["items"]] last_chatdata_timestamp = max(timestamps) if timestamps else None else: last_chatdata_timestamp = None else: last_chatdata_timestamp = None except Exception as e: logger.warning(f"Error fetching initial chat data: {e}") last_chatdata_timestamp = None # Keepalive interval (30 seconds) keepalive_interval = 30.0 last_keepalive = asyncio.get_event_loop().time() # Status check interval (check workflow status every 3 seconds) status_check_interval = 3.0 last_status_check = asyncio.get_event_loop().time() # Chat data fetch interval (fetch chat data every 0.5 seconds for real-time updates) chatdata_fetch_interval = 0.5 last_chatdata_fetch = asyncio.get_event_loop().time() # Stream events until completion or timeout timeout = 300.0 # 5 minutes max start_time = asyncio.get_event_loop().time() while True: # Check timeout elapsed = asyncio.get_event_loop().time() - start_time if elapsed > timeout: # Timeout - just close stream, don't emit non-chatData format events logger.info(f"Stream timeout for workflow {workflow.id}") break # Check for client disconnection if await request.is_disconnected(): logger.info(f"Client disconnected for workflow {workflow.id}") break current_time = asyncio.get_event_loop().time() # Periodically check workflow status and fetch chat data if current_time - last_status_check >= status_check_interval: try: current_workflow = interfaceDbChat.getWorkflow(workflow.id) if current_workflow and current_workflow.status == "stopped": logger.info(f"Workflow {workflow.id} was stopped, closing stream") # Don't emit stopped event - just close stream break except Exception as e: logger.warning(f"Error checking workflow status: {e}") last_status_check = current_time # Periodically fetch and emit chat data if current_time - last_chatdata_fetch >= chatdata_fetch_interval: try: chatData = interfaceDbChat.getUnifiedChatData(workflow.id, last_chatdata_timestamp) if chatData.get("items"): # Emit items directly in exact chatData format: {type, createdAt, item} for item in chatData["items"]: yield f"data: {json.dumps(item)}\n\n" # Update timestamp to only get new items next time if chatData["items"]: # Parse timestamps and get the maximum timestamps = [] for item in chatData["items"]: ts = parseTimestamp(item.get("createdAt"), default=0) timestamps.append(ts) if timestamps: last_chatdata_timestamp = max(timestamps) except Exception as e: logger.warning(f"Error fetching chat data: {e}") last_chatdata_fetch = current_time # Try to get event with timeout try: event = await asyncio.wait_for(queue.get(), timeout=1.0) # Only emit chatdata events (messages, logs, stats) in exact chatData format # Ignore status/progress/complete/stopped/error events that don't match the format if event.get("type") == "chatdata" and event.get("data"): # Emit item directly in exact chatData format: {type, createdAt, item} chatdata_item = event.get("data") yield f"data: {json.dumps(chatdata_item)}\n\n" # Update timestamp for incremental fetching if chatdata_item.get("createdAt"): last_chatdata_timestamp = parseTimestamp(chatdata_item["createdAt"], default=None) # Check if this is a completion/stopped event to close stream if event.get("type") == "complete": logger.info(f"Workflow {workflow.id} completed, closing stream") break elif event.get("type") == "stopped": # Workflow was stopped, close stream logger.info(f"Workflow {workflow.id} stopped, closing stream") break elif event.get("type") == "error" and event.get("step") == "error": # Final error, close stream logger.warning(f"Workflow {workflow.id} error, closing stream") break last_keepalive = asyncio.get_event_loop().time() except asyncio.TimeoutError: # Send keepalive if needed current_time = asyncio.get_event_loop().time() if current_time - last_keepalive >= keepalive_interval: yield f": keepalive\n\n" last_keepalive = current_time continue except Exception as e: logger.error(f"Error in event stream: {e}") yield f"data: {json.dumps({'type': 'error', 'message': f'Stream error: {str(e)}'})}\n\n" break except Exception as e: logger.error(f"Error in event stream generator: {e}", exc_info=True) # Don't emit error events that don't match chatData format finally: # Stream ends - no final event needed as it doesn't match chatData format pass return StreamingResponse( event_stream(), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", "Connection": "keep-alive", "X-Accel-Buffering": "no" # Disable buffering for nginx } ) except Exception as e: logger.error(f"Error in stream_chatbot_start: {str(e)}", exc_info=True) raise HTTPException( status_code=500, detail=str(e) ) # Workflow stop endpoint @router.post("/{workflowId}/stop", response_model=ChatWorkflow) @limiter.limit("120/minute") async def stop_chatbot( request: Request, workflowId: str = Path(..., description="ID of the workflow to stop"), currentUser: User = Depends(getCurrentUser) ) -> ChatWorkflow: """Stops a running chatbot workflow.""" try: workflow = await chatStop(currentUser, workflowId) # Emit stopped event to active streams event_manager = get_event_manager() await event_manager.emit_event( workflowId, "stopped", "Workflow stopped by user", "stopped" ) logger.info(f"Emitted stopped event for workflow {workflowId}") return workflow except Exception as e: logger.error(f"Error in stop_chatbot: {str(e)}") raise HTTPException( status_code=500, detail=str(e) ) # Delete chatbot workflow endpoint @router.delete("/{workflowId}", response_model=Dict[str, Any]) @limiter.limit("120/minute") async def delete_chatbot( request: Request, workflowId: str = Path(..., description="ID of the workflow to delete"), currentUser: User = Depends(getCurrentUser) ) -> Dict[str, Any]: """Deletes a chatbot workflow and its associated data.""" try: # Get service center interfaceDbChat = getServiceChat(currentUser) # Check workflow access and permission using RBAC workflows = getRecordsetWithRBAC( interfaceDbChat.db, ChatWorkflow, currentUser, recordFilter={"id": workflowId} ) if not workflows: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"Workflow with ID {workflowId} not found" ) workflow_data = workflows[0] # Check if workflow is a chatbot workflow if workflow_data.get("workflowMode") != WorkflowModeEnum.WORKFLOW_CHATBOT.value: raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=f"Workflow {workflowId} is not a chatbot workflow" ) # Check if user has permission to delete using RBAC if not interfaceDbChat.checkRbacPermission(ChatWorkflow, "delete", workflowId): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="You don't have permission to delete this workflow" ) # Delete workflow success = interfaceDbChat.deleteWorkflow(workflowId) if not success: raise HTTPException( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to delete workflow" ) return { "id": workflowId, "message": "Chatbot workflow and associated data deleted successfully" } except HTTPException: raise except Exception as e: logger.error(f"Error in delete_chatbot: {str(e)}", exc_info=True) raise HTTPException( status_code=500, detail=f"Error deleting chatbot workflow: {str(e)}" ) # List chatbot threads/workflows or get specific thread details @router.get("/threads") @limiter.limit("120/minute") async def get_chatbot_threads( request: Request, workflowId: Optional[str] = Query(None, description="Optional workflow ID to get details and chat data for a specific thread"), pagination: Optional[str] = Query(None, description="JSON-encoded PaginationParams object (only used when workflowId is not provided)"), currentUser: User = Depends(getCurrentUser) ) -> Union[PaginatedResponse[ChatWorkflow], Dict[str, Any]]: """ List all chatbot workflows (threads) for the current user, or get details and chat data for a specific thread. - If workflowId is provided: Returns the workflow details and all chat data (messages, logs, stats) - If workflowId is not provided: Returns a paginated list of all workflows """ try: interfaceDbChat = getServiceChat(currentUser) # If workflowId is provided, return single workflow with chat data if workflowId: workflow = interfaceDbChat.getWorkflow(workflowId) if not workflow: raise HTTPException( status_code=404, detail=f"Workflow with ID {workflowId} not found" ) # Get unified chat data for this workflow chatData = interfaceDbChat.getUnifiedChatData(workflowId, None) return { "workflow": workflow, "chatData": chatData } # Otherwise, return paginated list of workflows # Parse pagination parameter paginationParams = None if pagination: try: paginationDict = json.loads(pagination) paginationParams = PaginationParams(**paginationDict) if paginationDict else None except (json.JSONDecodeError, ValueError) as e: raise HTTPException( status_code=400, detail=f"Invalid pagination parameter: {str(e)}" ) # Get all workflows filtered by mandateId (RBAC handles this automatically) # We get all workflows first to filter by workflowMode before pagination all_workflows = interfaceDbChat.getWorkflows(pagination=None) # Filter to only include chatbot workflows chatbot_workflows_data = [ wf for wf in all_workflows if wf.get("workflowMode") == WorkflowModeEnum.WORKFLOW_CHATBOT.value ] # Apply pagination if requested if paginationParams: # Apply sorting if provided if paginationParams.sort: chatbot_workflows_data = interfaceDbChat._applySorting(chatbot_workflows_data, paginationParams.sort) # Count total items after filtering totalItems = len(chatbot_workflows_data) totalPages = math.ceil(totalItems / paginationParams.pageSize) if totalItems > 0 else 0 # Apply pagination (skip/limit) startIdx = (paginationParams.page - 1) * paginationParams.pageSize endIdx = startIdx + paginationParams.pageSize workflows_data = chatbot_workflows_data[startIdx:endIdx] else: workflows_data = chatbot_workflows_data totalItems = len(chatbot_workflows_data) totalPages = 1 # Convert raw dictionaries to ChatWorkflow objects workflows = [] for workflow_data in workflows_data: try: # Load the workflow properly workflow = interfaceDbChat.getWorkflow(workflow_data["id"]) if workflow: workflows.append(workflow) except Exception as e: logger.warning(f"Error loading workflow {workflow_data.get('id')}: {e}") continue # Create paginated response from modules.datamodels.datamodelPagination import PaginationMetadata metadata = PaginationMetadata( currentPage=paginationParams.page if paginationParams else 1, pageSize=paginationParams.pageSize if paginationParams else len(workflows), totalItems=totalItems, totalPages=totalPages, sort=paginationParams.sort if paginationParams else [], filters=paginationParams.filters if paginationParams else None ) return PaginatedResponse( items=workflows, pagination=metadata ) except HTTPException: raise except Exception as e: logger.error(f"Error getting chatbot threads: {str(e)}", exc_info=True) raise HTTPException( status_code=500, detail=f"Error getting chatbot threads: {str(e)}" )