import asyncio import json import logging import traceback import uuid import dirtyjson from fastapi import APIRouter, Depends, HTTPException from fastapi.responses import StreamingResponse from sqlalchemy.ext.asyncio import AsyncSession from constants.presentation import MAX_NUMBER_OF_SLIDES from models.presentation_outline_model import PresentationOutlineModel from models.sql.presentation import PresentationModel from models.sse_response import ( SSECompleteResponse, SSEErrorResponse, SSEResponse, SSEStatusResponse, ) from services.temp_file_service import TEMP_FILE_SERVICE from services.database import get_async_session from services.documents_loader import DocumentsLoader from services.mem0_presentation_memory_service import ( MEM0_PRESENTATION_MEMORY_SERVICE, ) from utils.llm_utils import message_content_to_text from utils.outline_utils import ( get_no_of_outlines_to_generate_for_n_slides, get_presentation_title_from_presentation_outline, ) from utils.outline_limits import normalize_outline_payload from utils.llm_calls.generate_presentation_outlines import ( OutlineGenerationStatus, generate_ppt_outline, get_messages as get_outline_messages, ) from utils.sse import safe_sse_stream from utils.web_search import get_selected_web_search_provider, get_web_search_route OUTLINES_ROUTER = APIRouter(prefix="/outlines", tags=["Outlines"]) LOGGER = logging.getLogger(__name__) @OUTLINES_ROUTER.get("/{id}", response_model=PresentationOutlineModel) async def get_outline( id: uuid.UUID, sql_session: AsyncSession = Depends(get_async_session), ): presentation = await sql_session.get(PresentationModel, id) if not presentation: raise HTTPException(status_code=404, detail="Presentation not found") if not presentation.outlines: return PresentationOutlineModel(slides=[]) return PresentationOutlineModel(**presentation.outlines) @OUTLINES_ROUTER.put("/{id}", response_model=PresentationOutlineModel) async def update_outline( id: uuid.UUID, outline: PresentationOutlineModel, sql_session: AsyncSession = Depends(get_async_session), ): presentation = await sql_session.get(PresentationModel, id) if not presentation: raise HTTPException(status_code=404, detail="Presentation not found") presentation.outlines = outline.model_dump(mode="json") presentation.n_slides = len(outline.slides) presentation.title = get_presentation_title_from_presentation_outline(outline) sql_session.add(presentation) await sql_session.commit() await MEM0_PRESENTATION_MEMORY_SERVICE.store_generated_outlines( presentation.id, presentation.outlines, ) return outline @OUTLINES_ROUTER.get("/stream/{id}") async def stream_outlines( id: uuid.UUID, sql_session: AsyncSession = Depends(get_async_session) ): presentation = await sql_session.get(PresentationModel, id) if not presentation: raise HTTPException(status_code=404, detail="Presentation not found") search_route, actual_search_provider = get_web_search_route() LOGGER.info( "Starting outline stream: presentation_id=%s web_search_enabled=%s " "selected_web_search_provider=%s web_search_route=%s actual_web_search_provider=%s", presentation.id, presentation.web_search, get_selected_web_search_provider().value, search_route, ( actual_search_provider.value if actual_search_provider else ("model-native" if search_route == "native" else "none") ), ) temp_dir = TEMP_FILE_SERVICE.create_temp_dir() async def inner(): yield SSEStatusResponse( status="Preparing your presentation outline" ).to_string() additional_context = "" if presentation.file_paths: documents_loader = DocumentsLoader( file_paths=presentation.file_paths, presentation_language=presentation.language, ) await documents_loader.load_documents(temp_dir) documents = documents_loader.documents if documents: additional_context = "\n\n".join(documents) presentation_outlines_text = "" if presentation.n_slides > 0: n_slides_to_generate = get_no_of_outlines_to_generate_for_n_slides( n_slides=presentation.n_slides, toc=presentation.include_table_of_contents, title_slide=presentation.include_title_slide, ) else: n_slides_to_generate = None outline_messages = get_outline_messages( presentation.content, n_slides_to_generate, presentation.language, additional_context, presentation.tone, presentation.verbosity, presentation.instructions, presentation.include_title_slide, presentation.include_table_of_contents, ) await MEM0_PRESENTATION_MEMORY_SERVICE.store_generation_context( presentation_id=presentation.id, system_prompt=( message_content_to_text(outline_messages[0].content) if len(outline_messages) > 0 else None ), user_prompt=( message_content_to_text(outline_messages[1].content) if len(outline_messages) > 1 else None ), extracted_document_text=additional_context, source_content=presentation.content, instructions=presentation.instructions, ) async for chunk in generate_ppt_outline( presentation.content, n_slides_to_generate, presentation.language, additional_context, presentation.tone, presentation.verbosity, presentation.instructions, presentation.include_title_slide, presentation.web_search, presentation.include_table_of_contents, emit_statuses=True, ): # Give control to the event loop await asyncio.sleep(0) if isinstance(chunk, OutlineGenerationStatus): LOGGER.info( "Outline generation status: presentation_id=%s status=%s", presentation.id, chunk.message, ) yield SSEStatusResponse(status=chunk.message).to_string() continue if isinstance(chunk, HTTPException): yield SSEErrorResponse(detail=chunk.detail).to_string() return yield SSEResponse( event="response", data=json.dumps({"type": "chunk", "chunk": chunk}), ).to_string() presentation_outlines_text += chunk try: presentation_outlines_json = dict( dirtyjson.loads(presentation_outlines_text) ) except Exception as e: traceback.print_exc() yield SSEErrorResponse( detail=f"Failed to generate presentation outlines. Please try again. {str(e)}", ).to_string() return presentation_outlines = PresentationOutlineModel( **normalize_outline_payload( presentation_outlines_json, MAX_NUMBER_OF_SLIDES, ) ) if ( n_slides_to_generate is not None and len(presentation_outlines.slides) != n_slides_to_generate ): yield SSEErrorResponse( detail=( "Failed to generate presentation outlines with requested " "number of slides. Please try again." ) ).to_string() return if n_slides_to_generate is not None: presentation_outlines.slides = presentation_outlines.slides[ :n_slides_to_generate ] if presentation.n_slides <= 0: presentation.n_slides = len(presentation_outlines.slides) presentation.outlines = presentation_outlines.model_dump() presentation.title = get_presentation_title_from_presentation_outline( presentation_outlines ) sql_session.add(presentation) await sql_session.commit() await MEM0_PRESENTATION_MEMORY_SERVICE.store_generated_outlines( presentation.id, presentation.outlines, ) yield SSECompleteResponse( key="presentation", value=presentation.model_dump(mode="json") ).to_string() async def rollback_stream_session(): await sql_session.rollback() return StreamingResponse( safe_sse_stream( inner(), logger=LOGGER, error_detail="Failed to generate presentation outlines. Please try again.", on_error=rollback_stream_session, ), media_type="text/event-stream", )