import os import logging from typing import Annotated from fastapi import FastAPI, Request, WebSocket, Depends, HTTPException, status from fastapi.security import HTTPBasic, HTTPBasicCredentials from fastapi.templating import Jinja2Templates from fastapi.staticfiles import StaticFiles from main import generate logging.basicConfig( level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", datefmt="%Y-%m-%d %H:%M:%S", ) logger = logging.getLogger(__name__) STATIC_DIR = "static" TEMPLATES_DIR = "templates" security = HTTPBasic() app = FastAPI() app.mount(f"/{STATIC_DIR}", StaticFiles(directory=STATIC_DIR), name="static") templates = Jinja2Templates(directory=os.path.join(STATIC_DIR, TEMPLATES_DIR)) def verify_credentials(credentials: HTTPBasicCredentials = Depends(security)): correct_username = os.getenv("BASIC_AUTH_USERNAME") correct_password = os.getenv("BASIC_AUTH_PASSWORD") if not ( credentials.username == correct_username and credentials.password == correct_password ): raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect username or password", headers={"WWW-Authenticate": "Basic"}, ) return credentials.username @app.get("/") async def home(request: Request, username: Annotated[str, Depends(verify_credentials)]): return templates.TemplateResponse("index.html", {"request": request}) @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() while True: data = await websocket.receive_text() for chunk in generate(data): await websocket.send_text(chunk) await websocket.send_text("<>")