Spaces:
Sleeping
Sleeping
File size: 1,964 Bytes
24352e2 8362005 24352e2 8362005 24352e2 8362005 24352e2 8362005 24352e2 8362005 24352e2 8362005 24352e2 8362005 24352e2 8362005 24352e2 8362005 24352e2 8362005 24352e2 8362005 24352e2 8362005 24352e2 8362005 24352e2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 |
#
# Copyright (c) 2025, Daily
#
# SPDX-License-Identifier: BSD 2-Clause License
#
import asyncio
import os
import sys
from contextlib import asynccontextmanager
from typing import Any, Dict
import uvicorn
from dotenv import load_dotenv
from fastapi import FastAPI, Request, WebSocket
from fastapi.middleware.cors import CORSMiddleware
# from bot.bot_fast_api import run_bot
from bot.bot_websocket_server import run_bot_websocket_server
# Load environment variables
load_dotenv(override=True)
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Handles FastAPI startup and shutdown."""
yield # Run app
# Initialize FastAPI app with lifespan manager
app = FastAPI(lifespan=lifespan)
# Configure CORS to allow requests from any origin
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.websocket("/ws")
async def websocket_endpoint(websocket: WebSocket):
await websocket.accept()
print("WebSocket connection accepted")
try:
await run_bot_websocket_server(websocket)
except Exception as e:
print(f"Exception in run_bot: {e}")
@app.post("/connect")
async def bot_connect(request: Request) -> Dict[Any, Any]:
return {"ws_url": "ws://localhost:7860/ws"}
async def main():
config = uvicorn.Config(app, host="0.0.0.0", port=7860)
server = uvicorn.Server(config)
await server.serve()
if __name__ == "__main__":
import signal
async def serve():
config = uvicorn.Config(app, host="0.0.0.0", port=7860)
server = uvicorn.Server(config)
await server.serve()
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(serve())
except KeyboardInterrupt:
print("Received exit signal (Ctrl+C), shutting down...")
finally:
loop.run_until_complete(loop.shutdown_asyncgens())
loop.close()
|