from fastapi import FastAPI, Form, HTTPException from fastapi.responses import JSONResponse, StreamingResponse import tempfile import requests from generate_storigo_content_document import generate_slide_content_document from generate_storigo_content_youtube import generate_slide_content_youtube from generate_storigo_content_html import generate_slide_content_html import json import threading import queue import time from typing import Optional, Callable import os import asyncio app = FastAPI() import logging import traceback def fetch_document_with_curl(url): response = requests.get(url, verify=False, timeout=60) if response.status_code == 200: return response.content else: raise HTTPException(status_code=404, detail=f"Document not found, status: {response.status_code}") async def process_and_generate_response(split_documents, client_id, num_slides,num_mcqs, is_image, is_question, question_position,isGPU): try: # Save embeddings create_and_save_embeddings(split_documents, client_id) merge_embeddings = merge_all_faiss(client_id) # Generate slide content slide_content = await generate_slide_content(merge_embeddings, client_id, num_slides,num_mcqs, is_image,is_question, question_position,isGPU) print(f"Slide content generated for {num_slides} slides.") # Serialize and return response response_data = slide_content.dict() return JSONResponse(content=response_data) except Exception as e: # Log the error for debugging logging.error(f"Error generating content: {str(e)}") # Return the actual error message to the frontend return JSONResponse( content={"error": f"Error: {str(e)}", "details": traceback.format_exc()}, status_code=500 ) @app.post("/generate-storigo-content") async def generate_storigo_content( client_id: str = Form(...), num_slides: int = Form(...), document_url: str = Form(...), is_image: str = Form(...), document_type:str = Form(...), is_question: str = Form(...), num_mcqs: int = Form(...), question_position : int = Form(...), isGPU: int = Form(...), ) -> JSONResponse: try: if document_type == "pdf": print(f"Fetching document from: {document_url}") document_content = fetch_document_with_curl(document_url) print(f"Document fetched, size: {len(document_content)} bytes") # Save PDF to a temporary file with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as temp_file: temp_file.write(document_content) temp_file_path = temp_file.name print(f"Temporary file saved at: {temp_file_path}") try: slide_content = await generate_slide_content_document(temp_file_path, client_id, num_slides, num_mcqs, is_image, is_question, question_position, isGPU) response_data = slide_content.dict() print(f"Generated response data with {len(response_data.get('slides', {}))} slides/items.") return JSONResponse(content=response_data) finally: os.remove(temp_file_path) # Cleanup temp file elif document_type == "youtube": print(f"Processing YouTube URL: {document_url}") slide_content = await generate_slide_content_youtube(document_url, client_id, num_slides, num_mcqs, is_image, is_question, question_position, isGPU) response_data = slide_content.dict() return JSONResponse(content=response_data) elif document_type == "html": print(f"Fetching HTML content from: {document_url}") slide_content = await generate_slide_content_html(document_url, client_id, num_slides, num_mcqs, is_image, is_question, question_position, isGPU) response_data = slide_content.dict() return JSONResponse(content=response_data) else: raise HTTPException(status_code=400, detail="Unsupported source type. Please use 'pdf', 'html', or 'youtube'.") # except Exception as e: # # Log the actual error for debugging # logging.error(f"Error generating content: {str(e)}") # # Return a generic error message to the frontend # # return JSONResponse( # # content={"error": "Something went wrong. Please try again later."}, # # status_code=500 # # ) # return JSONResponse( # status_code=500, # content={ # "error": str(e), # # only include this in dev—remove in prod! # "trace": traceback.format_exc().splitlines() # } # ) except Exception as e: # Log the error for debugging logging.error(f"Error generating content: {str(e)}") # Return the actual error message to the frontend return JSONResponse( content={"error": f"Error: {str(e)}", "details": traceback.format_exc()}, status_code=500 ) from generate_storigo_content_from_prompt import generate_slide_content_from_prompt @app.post("/generate-storigo-content-from-prompt") async def generate_storigo_content_from_prompt( client_id: int = Form(...), prompt: str = Form(...), num_slides: int = Form(...), is_image: int = Form(...), is_question: str = Form(...), num_mcqs: int = Form(...), question_position : int = Form(...), isGPU: int = Form(...), ) -> JSONResponse: try: # Generate slide content from prompt storigo_content = generate_slide_content_from_prompt(prompt, num_slides ,num_mcqs, is_image == 1, is_question, question_position , isGPU) # Convert the Pydantic model to a dictionary response_data = storigo_content.dict() return JSONResponse(content=response_data) # except Exception as e: # # Log the actual error for debugging # logging.error(f"Error generating content: {str(e)}") # # Return a generic error message to the frontend # return JSONResponse( # content={"error": "Something went wrong. Please try again later."}, # status_code=500 # ) except Exception as e: # Log the error for debugging logging.error(f"Error generating content: {str(e)}") # Return the actual error message to the frontend return JSONResponse( content={"error": f"Error: {str(e)}", "details": traceback.format_exc()}, status_code=500 ) @app.post("/generate-storigo-content-from-prompt-sse") async def generate_storigo_content_from_prompt_sse( client_id: int = Form(...), prompt: str = Form(...), num_slides: int = Form(...), is_image: int = Form(...), is_question: str = Form(...), num_mcqs: int = Form(...), question_position: int = Form(...), isGPU: int = Form(...), ): """ Stream progress events as Server-Sent Events (SSE). Each message is sent as JSON in `data:` lines. """ q: "queue.Queue[dict]" = queue.Queue() def progress_cb(evt: dict): # ensure JSON-serializable and add timestamp evt = dict(evt or {}) evt.setdefault("ts", int(time.time() * 1000)) q.put(evt) def worker(): try: progress_cb({"type": "status", "message": "Starting generation..."}) storigo_content = generate_slide_content_from_prompt( prompt=prompt, num_slides=int(num_slides), num_mcqs=int(num_mcqs), is_image=(int(is_image) == 1), is_question=is_question, question_position=question_position, GPU=isGPU, progress_cb=progress_cb, ) progress_cb({"type": "done", "result": storigo_content.dict()}) except Exception as e: progress_cb({"type": "error", "error": str(e), "details": traceback.format_exc()}) threading.Thread(target=worker, daemon=True).start() def event_stream(): while True: evt = q.get() evt_type = evt.get("type", "progress") payload = json.dumps(evt, ensure_ascii=False) yield f"event: {evt_type}\n" yield f"data: {payload}\n\n" if evt_type in ("done", "error"): break return StreamingResponse(event_stream(), media_type="text/event-stream") @app.post("/generate-storigo-content-sse") async def generate_storigo_content_sse( client_id: str = Form(...), num_slides: int = Form(...), document_url: str = Form(...), is_image: str = Form(...), document_type: str = Form(...), is_question: str = Form(...), num_mcqs: int = Form(...), question_position: int = Form(...), isGPU: int = Form(...), ): """ Stream progress events for document-based generation as Server-Sent Events (SSE). """ q: "queue.Queue[dict]" = queue.Queue() def progress_cb(evt: dict): evt = dict(evt or {}) evt.setdefault("ts", int(time.time() * 1000)) q.put(evt) def worker(): temp_file_path = None try: progress_cb({"type": "status", "message": f"Starting {document_type} generation..."}) if document_type == "pdf": progress_cb({"type": "status", "message": "Fetching PDF document..."}) document_content = fetch_document_with_curl(document_url) with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as temp_file: temp_file.write(document_content) temp_file_path = temp_file.name storigo_content = asyncio.run(generate_slide_content_document( temp_file_path, client_id, int(num_slides), int(num_mcqs), int(is_image) == 1, is_question, int(question_position), int(isGPU), progress_cb=progress_cb )) elif document_type == "youtube": storigo_content = asyncio.run(generate_slide_content_youtube( document_url, client_id, int(num_slides), int(num_mcqs), int(is_image) == 1, is_question, int(question_position), int(isGPU), progress_cb=progress_cb )) elif document_type == "html": storigo_content = asyncio.run(generate_slide_content_html( document_url, client_id, int(num_slides), int(num_mcqs), int(is_image) == 1, is_question, int(question_position), int(isGPU), progress_cb=progress_cb )) else: raise Exception(f"Unsupported document type: {document_type}") progress_cb({"type": "done", "result": storigo_content.dict()}) except Exception as e: progress_cb({"type": "error", "error": str(e), "details": traceback.format_exc()}) finally: if temp_file_path and os.path.exists(temp_file_path): os.remove(temp_file_path) threading.Thread(target=worker, daemon=True).start() def event_stream(): while True: evt = q.get() evt_type = evt.get("type", "progress") payload = json.dumps(evt, ensure_ascii=False) yield f"event: {evt_type}\n" yield f"data: {payload}\n\n" if evt_type in ("done", "error"): break return StreamingResponse(event_stream(), media_type="text/event-stream")