"""Session event handlers: participant join/leave, audio recording, transcript, cleanup.""" import asyncio import io import time import wave from datetime import datetime from loguru import logger from pipecat.frames.frames import LLMRunFrame from app.services.r2 import upload_bytes, upload_json from app.session_repository import save_session_complete from app.usage import get_usage_state, deduct_topup_credits from typing import TYPE_CHECKING if TYPE_CHECKING: from bot.core.config import BotSessionState, ChunkState, TranscriptManager def create_audio_data_handler(bot_session: "BotSessionState", chunk_state: "ChunkState"): """Handle audio buffer data — encode as WAV and upload chunk to R2.""" from bot.core.config import SAMPLE_RATE async def on_audio_data(buffer, audio, sample_rate, num_channels): if not chunk_state.session_id: chunk_state.session_id = bot_session.session_id or "unknown" session_id = chunk_state.session_id chunk_key = chunk_state.get_chunk_key(session_id, chunk_state.counter) chunk_num = chunk_state.add_chunk(chunk_key) try: wav_buffer = io.BytesIO() with wave.open(wav_buffer, "wb") as wav_file: wav_file.setnchannels(num_channels) wav_file.setsampwidth(2) wav_file.setframerate(sample_rate) wav_file.writeframes(audio) wav_buffer.seek(0) wav_data = wav_buffer.read() await asyncio.to_thread(upload_bytes, wav_data, chunk_key, "audio/wav") logger.debug(f"Chunk {chunk_num} uploaded: {chunk_key}") except Exception as e: logger.error(f"Failed to upload chunk {chunk_num}: {e}") return on_audio_data def create_transcript_handler(bot_session: "BotSessionState", transcript_manager: "TranscriptManager", task): """Capture transcript updates and send to frontend.""" async def on_transcript_update(processor, frame): try: for message in frame.messages: transcript_manager.add_message(message.role, message.content) logger.debug(f"TRANSCRIPT: session_id={bot_session.session_id} role={message.role} content={message.content[:200]}") except Exception as e: logger.error(f"Error in transcript handler: {e}", exc_info=True) return on_transcript_update def create_first_participant_joined_handler(bot_session: "BotSessionState", task, audio_buffer, chunk_state: "ChunkState"): async def on_first_participant_joined(transport, participant): join_time = time.time() participant_id = participant["id"] bot_session.current_user_id = participant_id await transport.capture_participant_transcription(participant_id) if not bot_session.session_id: bot_session.session_id = f"{datetime.now().strftime('%Y%m%d_%H%M%S')}_{participant_id}" bot_session.bot_start_time = join_time bot_session.user_joined = True bot_session.cancel_timeout_task() logger.info(f"USER_JOINED: session_id={bot_session.session_id} user_id={bot_session.user_id}") # Load usage state (one DB call) try: usage = await asyncio.to_thread(get_usage_state, bot_session.user_id) bot_session.is_premium = usage["is_premium"] bot_session.plan_images_remaining = usage["plan_images_remaining"] bot_session.plan_videos_remaining = usage["plan_videos_remaining"] bot_session.plan_ppts_remaining = usage["plan_ppts_remaining"] bot_session.topup_images_remaining = usage["topup_images_remaining"] bot_session.topup_videos_remaining = usage["topup_videos_remaining"] except Exception as e: logger.error(f"Failed to load usage state: {e}") # Start audio recording try: chunk_state.reset(bot_session.session_id) await audio_buffer.start_recording() bot_session.audio_buffer = audio_buffer except Exception as e: logger.error(f"Failed to start recording: {e}") await task.queue_frames([LLMRunFrame()]) return on_first_participant_joined def create_participant_left_handler( bot_session: "BotSessionState", transcript_manager: "TranscriptManager", chunk_state: "ChunkState", task, ): async def on_participant_left(transport, participant, reason): await _handle_participant_left( participant, reason, bot_session, transcript_manager, chunk_state, task ) return on_participant_left def create_cleanup_if_no_user(bot_session: "BotSessionState", task): from bot.core.config import cleanup_session, NO_USER_TIMEOUT async def cleanup_if_no_user(): await asyncio.sleep(NO_USER_TIMEOUT) if not bot_session.user_joined: logger.warning(f"NO_USER_TIMEOUT: {bot_session.get_session_info()} timeout={NO_USER_TIMEOUT}s") await cleanup_session(bot_session, bot_session.session_id) await task.cancel() return cleanup_if_no_user # --- Internal --- async def _handle_participant_left( participant, reason, bot_session: "BotSessionState", transcript_manager: "TranscriptManager", chunk_state: "ChunkState", task, ): from bot.core.config import cleanup_session leave_time = time.time() participant_id = participant.get("id") if isinstance(participant, dict) else str(participant) session_id = bot_session.session_id or "unknown" user_id = bot_session.user_id or "unknown" duration_seconds = int(leave_time - bot_session.bot_start_time) logger.info(f"USER_LEFT: session_id={session_id} duration={duration_seconds}s reason={reason}") bot_session.cancel_timeout_task() # 1. Stop audio recording and upload manifest try: if bot_session.audio_buffer: await bot_session.audio_buffer.stop_recording() await asyncio.sleep(1.0) if chunk_state.total_chunks > 0: manifest = { "session_id": session_id, "user_id": user_id, "total_chunks": chunk_state.total_chunks, "sample_rate": 16000, "num_channels": 1, "chunks": chunk_state.chunk_list, "created_at": datetime.now().isoformat(), } manifest_key = f"sessions/{session_id}/manifest.json" await asyncio.to_thread(upload_json, manifest, manifest_key) logger.info(f"AUDIO_SAVED: session_id={session_id} chunks={chunk_state.total_chunks}") except Exception as e: logger.error(f"Failed to save recording: {e}") # 2. Save transcript to R2 transcription_r2_key = None try: if transcript_manager.count > 0: transcript_data = { "session_id": session_id, "user_id": user_id, "participant_id": participant_id, "start_time": datetime.fromtimestamp(bot_session.bot_start_time).isoformat(), "end_time": datetime.now().isoformat(), "duration_seconds": duration_seconds, "message_count": transcript_manager.count, "messages": transcript_manager.messages, } transcription_r2_key = f"sessions/{session_id}/transcription.json" await asyncio.to_thread(upload_json, transcript_data, transcription_r2_key) logger.info(f"TRANSCRIPT_SAVED: session_id={session_id} messages={transcript_manager.count}") except Exception as e: logger.error(f"Failed to save transcript: {e}") # 3. Save session metadata to R2 try: image_gen_count = transcript_manager.get_image_generations_count() image_edit_count = transcript_manager.get_image_edits_count() metadata = { "session_id": session_id, "user_id": user_id, "character_id": bot_session.character_id, "character_name": bot_session.character_name, "started_at": datetime.fromtimestamp(bot_session.bot_start_time).isoformat(), "ended_at": datetime.now().isoformat(), "duration_seconds": duration_seconds, "statistics": { "total_messages": transcript_manager.count, "user_messages": len(transcript_manager.get_user_messages()), "assistant_messages": len(transcript_manager.get_assistant_messages()), "image_generations": image_gen_count, "image_edits": image_edit_count, "videos": len(bot_session.generated_videos), }, "transcription_r2_key": transcription_r2_key, "audio_chunks": chunk_state.total_chunks, } # Collect generated media references media_items = [] for msg in transcript_manager.get_generations(): gen_type = msg.get("generation_type") content = msg.get("content", {}) if gen_type == "image": media_items.append({ "type": "image", "r2_url": content.get("r2_url"), "prompt": content.get("description"), }) elif gen_type == "image_edit": media_items.append({ "type": "image_edit", "r2_url": content.get("edited_r2_url") or content.get("r2_url"), "original_r2_url": content.get("original_r2_url"), "prompt": content.get("edit_instruction"), }) for vid in bot_session.generated_videos: vid_meta = vid.get("metadata", {}) media_items.append({ "type": vid.get("video_type"), "r2_url": vid_meta.get("r2_url"), "prompt": vid.get("prompt"), }) metadata["media"] = media_items metadata_key = f"sessions/{session_id}/session_metadata.json" await asyncio.to_thread(upload_json, metadata, metadata_key) logger.info(f"SESSION_METADATA_SAVED: session_id={session_id} media_items={len(media_items)}") except Exception as e: logger.error(f"Failed to save session metadata: {e}") # 4. Save session to DB try: # Collect images for DB images_to_save = [] for msg in transcript_manager.get_generations(): gen_type = msg.get("generation_type") content = msg.get("content", {}) if gen_type == "image": images_to_save.append({ "image_type": "image", "metadata": {"r2_url": content.get("r2_url")}, "prompt": content.get("description"), }) elif gen_type == "image_edit": images_to_save.append({ "image_type": "image_edit", "metadata": { "r2_url": content.get("edited_r2_url") or content.get("r2_url"), "original_r2_url": content.get("original_r2_url"), }, "prompt": content.get("edit_instruction"), }) # Collect videos for DB videos_to_save = [] for vid in bot_session.generated_videos: vid_meta = vid.get("metadata", {}) videos_to_save.append({ "video_type": vid.get("video_type"), "metadata": {"r2_url": vid_meta.get("r2_url")}, "prompt": vid.get("prompt"), }) recording_r2_key = f"sessions/{session_id}/manifest.json" if chunk_state.total_chunks > 0 else None metadata_r2_key = f"sessions/{session_id}/session_metadata.json" await save_session_complete( session_id=session_id, user_id=user_id, session_data={ "character_id": bot_session.character_id, "recording_r2_key": recording_r2_key, "transcription_r2_key": transcription_r2_key, "session_metadata_r2_key": metadata_r2_key, "image_generations_count": transcript_manager.get_image_generations_count(), "image_edits_count": transcript_manager.get_image_edits_count(), "video_count": len(bot_session.generated_videos), "duration_seconds": duration_seconds, "started_at": int(bot_session.bot_start_time), "ended_at": int(leave_time), "status": "completed", }, images=images_to_save, videos=videos_to_save, ) logger.info(f"SESSION_DB_SAVED: session_id={session_id} images={len(images_to_save)} videos={len(videos_to_save)}") except Exception as e: logger.error(f"Failed to save session to DB: {e}") # 5. Deduct topup credits if any were used this session try: if bot_session._topup_images_used_this_session > 0 or bot_session._topup_videos_used_this_session > 0: await asyncio.to_thread( deduct_topup_credits, user_id, bot_session._topup_images_used_this_session, bot_session._topup_videos_used_this_session, ) except Exception as e: logger.error(f"Failed to deduct topup credits: {e}") # 6. Cleanup logger.info(f"CLEANUP_TRIGGER: session_id={session_id} reason=participant_left") await cleanup_session(bot_session, session_id, transcript_manager, chunk_state) await task.cancel()