Blame for local/server.py

"""
Local FastAPI server for Pipecat development.
"""

import asyncio
import json
import os
import subprocess
import sys
import uuid
import time
from typing import Dict, Any, Optional

import aiohttp
from fastapi import FastAPI, HTTPException, WebSocket
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel
import uvicorn
from loguru import logger
from dotenv import load_dotenv

# Add parent directory for imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from utils.session import get_max_session_duration

from pipecat.transports.daily.utils import (
    DailyRESTHelper,
    DailyRoomParams,
    DailyRoomProperties,
)

# Load environment variables
load_dotenv()

app = FastAPI(title="Local Pipecat Server", version="1.0.0")

# Add CORS middleware
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)


class BotStartRequest(BaseModel):
    createDailyRoom: bool = True
    dailyRoomProperties: Optional[Dict[str, Any]] = None
    body: Dict[str, Any]


class BotStartResponse(BaseModel):
    dailyRoom: str
    dailyToken: str
    session_id: Optional[str] = None


@app.post("/", response_model=BotStartResponse)
async def start_bot(request: BotStartRequest):
    """Start a bot process locally."""
    try:
        session_id = str(uuid.uuid4())

        logger.info(f"🚀 Starting bot session: {session_id}")

        # Get max session duration with priority: API > ENV > Default
        max_session_duration = get_max_session_duration(request.body)

        # Add max_session_duration to bot_data if not already present
        bot_data = request.body.copy()
        if "max_session_duration" not in bot_data:
            bot_data["max_session_duration"] = max_session_duration

        # Create Daily room and token
        room_url, token = await create_daily_room_and_token(session_id, request)

        # Start bot process with piped logs
        await start_bot_process(session_id, room_url, token, bot_data)

        logger.info(f"✅ Bot started successfully for session: {session_id}")

        return BotStartResponse(
            dailyRoom=room_url, dailyToken=token, session_id=session_id
        )

    except Exception as e:
        logger.error(f"❌ Error starting bot: {e}")
        raise HTTPException(status_code=500, detail=f"Failed to start bot: {str(e)}")


async def create_daily_room_and_token(
    session_id: str, request: BotStartRequest
) -> tuple[str, str]:
    """Create a Daily room and meeting token."""
    daily_api_key = os.getenv("DAILY_API_KEY")
    if not daily_api_key:
        raise HTTPException(
            status_code=500, detail="DAILY_API_KEY environment variable is required"
        )

    # Get max session duration with priority: API > ENV > Default
    max_session_duration = get_max_session_duration(request.body)

    async with aiohttp.ClientSession() as http_session:
        daily_helper = DailyRESTHelper(
            daily_api_key=daily_api_key, aiohttp_session=http_session
        )

        enable_recording = None
        recordings_bucket = None
        recordings_template = None
        if request.dailyRoomProperties:
            enable_recording = request.dailyRoomProperties.get(
                "enable_recording", False
            )
            recordings_bucket = request.dailyRoomProperties.get(
                "recordings_bucket", None
            )
            recordings_template = request.dailyRoomProperties.get(
                "recordings_template", None
            ).replace(" ", "_")

        room_properties = DailyRoomProperties(
            exp=int(time.time() + max_session_duration),
            enable_chat=False,
            enable_prejoin_ui=False,
            start_video_off=True,
            eject_at_room_exp=True,
            enable_recording=enable_recording,
            recordings_bucket=recordings_bucket,
            recordings_template=recordings_template,  # type: ignore[attr-defined]
            dialout_config={"max_idle_timeout_sec": 0},
        )

        room_name = f"pipecat-local-{session_id[:8]}"
        room_params = DailyRoomParams(
            name=room_name, privacy="public", properties=room_properties
        )

        logger.info(f"🏠 Creating Daily room: {room_name}")
        room = await daily_helper.create_room(room_params)

        logger.info(f"🎫 Creating meeting token for room: {room.name}")
        token = await daily_helper.get_token(
            room_url=room.url, expiry_time=3600, eject_at_token_exp=True, owner=True
        )

        logger.info(f"✅ Daily room created: {room.url}")
        return room.url, token


async def start_bot_process(
    session_id: str, room_url: str, token: str, bot_data: Dict[str, Any]
):
    """Start the bot.py process with piped stdout/stderr."""
    try:
        # Get project root (parent of local/)
        project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))

        cmd = [
            "uv",
            "run",
            "local/runner.py",
            "--session-id",
            session_id,
            "--room-url",
            room_url,
            "--token",
            token,
            "--bot-data",
            json.dumps(bot_data),
        ]

        logger.info(f"🔧 Starting bot process in {project_root}")
        logger.info(f"🔧 Command: {' '.join(cmd)}")

        # Start process with logs piped to this process
        process = subprocess.Popen(
            cmd,
            cwd=project_root,
            env=os.environ.copy(),
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,  # Merge stderr into stdout
            text=True,
            bufsize=1,
            universal_newlines=True,
        )

        logger.info(f"✅ Bot process started with PID: {process.pid}")

        # Start task to pipe logs
        asyncio.create_task(pipe_logs(process, session_id))

    except Exception as e:
        logger.error(f"❌ Failed to start bot process: {e}")
        raise


async def pipe_logs(process: subprocess.Popen, session_id: str):
    """Pipe subprocess logs to main logger."""
    try:
        while True:
            line = process.stdout.readline()
            if not line:
                if process.poll() is not None:
                    break
                await asyncio.sleep(0.1)
                continue

            # Log subprocess output with session prefix
            logger.info(f"[{session_id[:8]}] {line.strip()}")

    except Exception as e:
        logger.error(f"❌ Error piping logs for {session_id}: {e}")


# Twilio WebSocket endpoint for Media Streams
@app.websocket("/twilio")
async def twilio_websocket(websocket: WebSocket):
    """
    Twilio Media Streams WebSocket endpoint.

    This handles incoming calls via Twilio's <Stream> TwiML directive.
    """
    try:
        from bot import twilio_websocket_handler
        await twilio_websocket_handler(websocket)
    except ImportError as e:
        logger.error(f"❌ Failed to import bot: {e}")
        await websocket.close(code=1011, reason="Server configuration error")


if __name__ == "__main__":
    host = "127.0.0.1"
    port = 8000

    logger.info("🚀 Starting Local Pipecat Server")
    logger.info(f"📍 Server will run at: http://{host}:{port}")
    logger.info("📋 Available endpoints:")
    logger.info("  POST /          - Start bot session (Daily.co/WebRTC)")
    logger.info("  WS   /twilio    - Twilio Media Streams WebSocket")

    uvicorn.run(app, host=host, port=port, log_level="info")