"""Video ads generation service using OpenAI Sora API with R2 storage.""" import asyncio import json import os import time import uuid 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 from typing import TYPE_CHECKING if TYPE_CHECKING: from bot.core.config import BotSessionState, TranscriptManager # --- Function Schema --- generate_video_ad_function = FunctionSchema( name="generate_video_ad", description=( "Generate a video advertisement using AI. Use when the user describes a video ad they want to create. " "Craft a detailed, cinematic prompt based on the user's product/brand, target audience, and style preferences. " "Do NOT call if video generation is already in progress." ), properties={ "prompt": { "type": "string", "description": ( "Detailed cinematic description of the video ad to generate. Include: " "product/brand visuals, scene setting, camera movement, lighting, mood, " "and any text overlays or call-to-action moments." ), }, }, required=["prompt"], ) # --- 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}") def _get_openai_client(): """Get OpenAI client with API key.""" import openai api_key = os.getenv("OPENAI_API_KEY", "") if not api_key: raise ValueError("OPENAI_API_KEY not set") return openai.OpenAI(api_key=api_key) async def _generate_sora_video(prompt: str) -> bytes: """Generate a video using OpenAI Sora API. Returns video bytes. Sora is async: create job -> poll until completed -> download content. Uses the SDK's create_and_poll convenience method. """ def _run(): client = _get_openai_client() # Create video and poll until completed video = client.videos.create_and_poll( model="sora-2", prompt=prompt, size="1280x720", seconds="4", ) if video.status != "completed": error_msg = video.error.message if video.error else "Unknown error" raise Exception(f"Sora generation failed: {error_msg}") # Download the video content response = client.videos.download_content(video.id, variant="video") return response.read() return await asyncio.to_thread(_run) # --- Handler --- def create_video_ad_handler( bot_session: "BotSessionState", transcript_manager: "TranscriptManager", task, ): """Create video ad generation handler using Sora.""" async def handle_generate_video_ad(params: FunctionCallParams): prompt = params.arguments.get("prompt", "") session_id = bot_session.session_id or "unknown" if not prompt: await params.result_callback(json.dumps({ "status": "error", "assistant_instruction": "No prompt provided. Ask the user what kind of video ad they want to create.", })) 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_video() if limit_reason == "subscribe": await _send_rtvi(task, { "type": "limit_reached", "data": {"media_type": "video", "reason": "subscribe"}, }) await params.result_callback(json.dumps({ "status": "limit_reached", "assistant_instruction": "The user has used all their free video credits. Tell them to subscribe to Maya Pro for more videos.", })) return elif limit_reason == "topup": await _send_rtvi(task, { "type": "limit_reached", "data": {"media_type": "video", "reason": "topup"}, }) await params.result_callback(json.dumps({ "status": "limit_reached", "assistant_instruction": "The user has used all their video credits. Tell them to buy a top-up pack for more credits.", })) return bot_session.is_processing = True video_id = f"ad_{int(time.time())}_{uuid.uuid4().hex[:4]}" await _send_rtvi(task, {"type": "generation_started", "data": {"type": "video", "status": "in_progress"}}) try: logger.info(f"VIDEO_AD_START: session={session_id} prompt={prompt[:80]}") start_time = time.time() # Generate video via Sora (create + poll + download) video_bytes = await _generate_sora_video(prompt) # Upload to R2 r2_key = f"sessions/{session_id}/videos/{video_id}.mp4" r2_url = await asyncio.to_thread(upload_bytes, video_bytes, r2_key, "video/mp4") total_time = time.time() - start_time logger.info(f"VIDEO_AD_GENERATED: session={session_id} video_id={video_id} url={r2_url} time={total_time:.1f}s") bot_session.add_generated_video( video_type="video_ad", metadata={"video_id": video_id, "r2_url": r2_url, "model": "sora-2", "total_time": total_time}, prompt=prompt, ) bot_session.use_video() await _send_rtvi(task, { "type": "video_generation_complete", "data": {"video_url": r2_url, "video_id": video_id, "description": prompt}, }) await params.result_callback(json.dumps({ "status": "completed", "video_url": r2_url, "video_id": video_id, "assistant_instruction": "Video ad is ready! Tell the user their video advertisement has been generated and they can see it on screen.", })) except Exception as e: logger.error(f"VIDEO_AD_ERROR: session={session_id} error={e}", exc_info=True) await _send_rtvi(task, { "type": "generation_failed", "data": {"error": str(e)[:100], "media_type": "video"}, }) await params.result_callback(json.dumps({ "status": "error", "assistant_instruction": f"Video ad generation failed: {str(e)[:100]}. Apologize and suggest trying again.", })) finally: bot_session.is_processing = False return handle_generate_video_ad