"""Presentation generation service using Gamma API with R2 storage.""" import asyncio import io import json import os import time import uuid import httpx import fitz # PyMuPDF from loguru import logger from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.frames.frames import OutputTransportMessageUrgentFrame from pipecat.services.llm_service import FunctionCallParams from app.services.r2 import upload_bytes, generate_key from typing import TYPE_CHECKING if TYPE_CHECKING: from bot.core.config import BotSessionState, TranscriptManager GAMMA_API_URL = "https://public-api.gamma.app/v1.0/generations" GAMMA_API_KEY = os.getenv("GAMMA_API_KEY", "") POLL_INTERVAL = 5 # seconds between status checks (per Gamma docs) POLL_TIMEOUT = 120 # max seconds to wait # --- Function Schema --- generate_ppt_function = FunctionSchema( name="generate_ppt", description=( "Generate a presentation (PPT/PDF) from a text description using AI. " "The user describes the topic and you create a professional presentation. " "CRITICAL: Do NOT call this function again if generation is already in progress." ), properties={ "topic": { "type": "string", "description": "The topic or detailed description for the presentation", }, "num_slides": { "type": "integer", "description": "Number of slides to generate (default 8, max 15)", }, }, required=["topic"], ) # --- Helpers --- async def _send_rtvi(task, data: dict): try: frame = OutputTransportMessageUrgentFrame(message={ "label": "rtvi-ai", "type": "server-message", "data": data, }) await task.queue_frame(frame) except Exception as e: logger.warning(f"Failed to send RTVI message: {e}") async def _create_generation(topic: str, num_slides: int) -> str: """Start a Gamma generation and return the generationId.""" api_key = GAMMA_API_KEY or os.getenv("GAMMA_API_KEY", "") if not api_key: raise Exception("GAMMA_API_KEY not set") async with httpx.AsyncClient(timeout=30.0) as client: response = await client.post( GAMMA_API_URL, headers={ "Content-Type": "application/json", "X-API-KEY": api_key, }, json={ "inputText": topic, "textMode": "generate", "format": "presentation", "numCards": min(num_slides, 15), "exportAs": "pdf", }, ) response.raise_for_status() data = response.json() generation_id = data.get("generationId") if not generation_id: raise Exception(f"No generationId in response: {data}") return generation_id async def _poll_generation(generation_id: str) -> dict: """Poll Gamma until generation completes. Returns the final response dict.""" api_key = GAMMA_API_KEY or os.getenv("GAMMA_API_KEY", "") url = f"{GAMMA_API_URL}/{generation_id}" headers = { "Content-Type": "application/json", "X-API-KEY": api_key, } start = time.time() async with httpx.AsyncClient(timeout=30.0) as client: while time.time() - start < POLL_TIMEOUT: response = await client.get(url, headers=headers) response.raise_for_status() data = response.json() status = data.get("status", "") if status == "completed": return data elif status in ("failed", "error"): raise Exception(f"Gamma generation failed: {data}") await asyncio.sleep(POLL_INTERVAL) raise Exception(f"Gamma generation timed out after {POLL_TIMEOUT}s") async def _download_and_store_pdf(export_url: str, session_id: str) -> tuple[str, str]: """Download PDF from Gamma, extract first page as thumbnail, upload both to R2. Returns (pdf_url, thumbnail_url).""" async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as client: response = await client.get(export_url) response.raise_for_status() pdf_bytes = response.content logger.info(f"PPT_PDF_DOWNLOADED: session={session_id} size={len(pdf_bytes)}") # Upload PDF to R2 pdf_key = generate_key(session_id, "ppt", ".pdf") pdf_url = await asyncio.get_event_loop().run_in_executor( None, upload_bytes, pdf_bytes, pdf_key, "application/pdf" ) # Extract first page as thumbnail using PyMuPDF thumbnail_url = "" try: doc = fitz.open(stream=pdf_bytes, filetype="pdf") if doc.page_count > 0: page = doc[0] # Render at 2x for good quality thumbnail mat = fitz.Matrix(2, 2) pix = page.get_pixmap(matrix=mat) thumb_bytes = pix.tobytes("png") doc.close() thumb_key = generate_key(session_id, "ppt_thumb", ".png") thumbnail_url = await asyncio.get_event_loop().run_in_executor( None, upload_bytes, thumb_bytes, thumb_key, "image/png" ) logger.info(f"PPT_THUMBNAIL_CREATED: session={session_id} size={len(thumb_bytes)}") else: doc.close() except Exception as e: logger.warning(f"PPT_THUMBNAIL_FAILED: session={session_id} error={e}") return pdf_url, thumbnail_url # --- Handler factory --- def create_ppt_handler( bot_session: "BotSessionState", transcript_manager: "TranscriptManager", task, ): """Create presentation generation handler.""" async def handle_generate_ppt(params: FunctionCallParams): args = params.arguments topic = args.get("topic", "") num_slides = args.get("num_slides", 8) session_id = bot_session.session_id or "unknown" if not topic: await params.result_callback(json.dumps({ "status": "error", "assistant_instruction": "No topic provided. Ask the user what presentation they want.", })) return if bot_session.is_processing: await params.result_callback(json.dumps({ "status": "busy", "assistant_instruction": "A generation is already in progress. Wait silently.", })) return # Usage limit check limit_reason = bot_session.can_generate_ppt() if limit_reason == "subscribe": await _send_rtvi(task, { "type": "limit_reached", "data": {"media_type": "ppt", "reason": "subscribe"}, }) await params.result_callback(json.dumps({ "status": "limit_reached", "assistant_instruction": "The user has used all their free presentation credits. Tell them to subscribe to Maya Pro for more.", })) return elif limit_reason == "topup": await _send_rtvi(task, { "type": "limit_reached", "data": {"media_type": "ppt", "reason": "topup"}, }) await params.result_callback(json.dumps({ "status": "limit_reached", "assistant_instruction": "The user has used all their presentation credits for this subscription period. Tell them they can buy a top-up pack.", })) return bot_session.is_processing = True ppt_id = f"ppt_{int(time.time())}_{uuid.uuid4().hex[:4]}" await _send_rtvi(task, {"type": "generation_started", "data": {"type": "ppt", "status": "in_progress"}}) try: logger.info(f"PPT_GENERATE_START: session={session_id} topic={topic[:60]} slides={num_slides}") start_time = time.time() # Start generation generation_id = await _create_generation(topic, num_slides) logger.info(f"PPT_GENERATION_CREATED: session={session_id} generation_id={generation_id}") # Poll until complete result = await _poll_generation(generation_id) export_url = result.get("exportUrl", "") gamma_url = result.get("gammaUrl", "") gamma_id = result.get("gammaId", "") # Download PDF and create thumbnail, upload to R2 pdf_url, thumbnail_url = await _download_and_store_pdf(export_url, session_id) total_time = time.time() - start_time logger.info( f"PPT_GENERATED: session={session_id} ppt_id={ppt_id} " f"gamma_id={gamma_id} time={total_time:.1f}s pdf={pdf_url}" ) transcript_manager.add_generation("ppt", { "ppt_id": ppt_id, "topic": topic, "gamma_id": gamma_id, "gamma_url": gamma_url, "pdf_url": pdf_url, "thumbnail_url": thumbnail_url, "export_url": export_url, "total_time": round(total_time, 1), }) bot_session.use_ppt() await _send_rtvi(task, { "type": "ppt_generation_complete", "data": { "ppt_id": ppt_id, "pdf_url": pdf_url, "thumbnail_url": thumbnail_url, "topic": topic, }, }) await params.result_callback(json.dumps({ "status": "completed", "ppt_id": ppt_id, "pdf_url": pdf_url, "thumbnail_url": thumbnail_url, "assistant_instruction": "Presentation is ready! Tell the user their presentation has been generated and they can download the PDF.", })) except Exception as e: logger.error(f"PPT_ERROR: session={session_id} error={e}", exc_info=True) await _send_rtvi(task, { "type": "generation_failed", "data": {"error": str(e)[:100], "media_type": "ppt"}, }) await params.result_callback(json.dumps({ "status": "error", "assistant_instruction": f"Presentation generation failed: {str(e)[:100]}. Apologize and suggest trying again.", })) finally: bot_session.is_processing = False return handle_generate_ppt