api/base.py

import importlib
import sys

sys.path.insert(0, ".")
import logging
import asyncio

# Install uvloop for better async performance (2-4x speedup on I/O)
try:
    import uvloop

    uvloop.install()
    logging.info("uvloop installed successfully")
except ImportError:
    logging.warning("uvloop not available, using default asyncio event loop")

from contextlib import asynccontextmanager
from typing import Any
from fastapi import FastAPI, Request
from fastapi.responses import RedirectResponse, HTMLResponse
from fastapi.middleware.cors import CORSMiddleware
from scalar_fastapi import get_scalar_api_reference
from lyric_search.sources import redis_cache
import shared  # Shared connection pools

logging.basicConfig(level=logging.INFO)
logging.getLogger("aiosqlite").setLevel(logging.WARNING)
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("python_multipart.multipart").setLevel(logging.WARNING)
logging.getLogger("streamrip").setLevel(logging.WARNING)
logging.getLogger("utils.sr_wrapper").setLevel(logging.WARNING)
logger = logging.getLogger()

loop = asyncio.get_event_loop()

# Pre-import endpoint modules so we can wire up lifespan
constants = importlib.import_module("constants").Constants()

# Will be set after app creation
_routes: dict = {}


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Lifespan context manager for startup/shutdown events."""
    # Startup
    uvicorn_access_logger = logging.getLogger("uvicorn.access")
    uvicorn_access_logger.disabled = True

    # Initialize shared infrastructure (Redis pool, aiohttp session, SQLite pool)
    await shared.startup()

    # Start Radio playlists
    if "radio" in _routes and hasattr(_routes["radio"], "on_start"):
        await _routes["radio"].on_start()

    # Start endpoint background tasks
    if "trip" in _routes and hasattr(_routes["trip"], "startup"):
        await _routes["trip"].startup()
    if "lighting" in _routes and hasattr(_routes["lighting"], "startup"):
        await _routes["lighting"].startup()

    logger.info("Application startup complete")

    yield

    # Shutdown
    if "lighting" in _routes and hasattr(_routes["lighting"], "shutdown"):
        await _routes["lighting"].shutdown()
    if "trip" in _routes and hasattr(_routes["trip"], "shutdown"):
        await _routes["trip"].shutdown()

    # Clean up shared infrastructure
    await shared.shutdown()

    logger.info("Application shutdown complete")


app = FastAPI(
    title="codey.lol API",
    version="1.0",
    contact={"name": "codey"},
    redirect_slashes=False,
    loop=loop,
    docs_url=None,  # Disabled - using Scalar at /docs instead
    redoc_url="/redoc",
    lifespan=lifespan,
)

util = importlib.import_module("util").Utilities(app, constants)

origins = [
    "https://codey.lol",
    "https://old.codey.lol",
    "https://api.codey.lol",
    "https://status.boatson.boats",
    "https://_new.codey.lol",
    "http://localhost:4321",
    "https://local.codey.lol:4321",
]

app.add_middleware(
    CORSMiddleware,  # type: ignore
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["POST", "GET", "HEAD", "OPTIONS"],
    allow_headers=["*"],
)  # type: ignore


# Scalar API documentation at /docs (replaces default Swagger UI)
@app.get("/docs", include_in_schema=False)
def scalar_docs():
    # Replace default FastAPI favicon with site favicon
    html_response = get_scalar_api_reference(
        openapi_url="/openapi.json", title="codey.lol API"
    )
    try:
        body = (
            html_response.body.decode("utf-8")
            if isinstance(html_response.body, (bytes, bytearray))
            else str(html_response.body)
        )
        body = body.replace(
            "https://fastapi.tiangolo.com/img/favicon.png",
            "https://codey.lol/images/favicon.png",
        )
        # Build fresh response so Content-Length matches modified body
        return HTMLResponse(content=body, status_code=html_response.status_code)
    except Exception:
        # Fallback to original if anything goes wrong
        return html_response


@app.get("/favicon.ico", include_in_schema=False)
async def favicon():
    """Redirect favicon requests to the site icon."""
    return RedirectResponse("https://codey.lol/images/favicon.png")


"""
Blacklisted routes
"""


@app.get("/", include_in_schema=False)
def disallow_get():
    return util.get_blocked_response()


@app.head("/", include_in_schema=False)
def base_head():
    return


@app.get("/{path}", include_in_schema=False)
def disallow_get_any(request: Request, var: Any = None):
    path = request.path_params["path"]
    allowed_paths = ["widget", "misc/no", "docs", "redoc", "openapi.json"]
    logging.info(
        f"Checking path: {path}, allowed: {path in allowed_paths or path.split('/', maxsplit=1)[0] in allowed_paths}"
    )
    if not (
        isinstance(path, str)
        and (path.split("/", maxsplit=1)[0] in allowed_paths or path in allowed_paths)
    ):
        logging.error(f"BLOCKED path: {path}")
        return util.get_blocked_response()
    else:
        logging.info("OK, %s", path)


@app.post("/", include_in_schema=False)
def disallow_base_post():
    return util.get_blocked_response()


"""
End Blacklisted Routes
"""

"""
Actionable Routes
"""

_routes.update(
    {
        "randmsg": importlib.import_module("endpoints.rand_msg").RandMsg(
            app, util, constants
        ),
        "lyrics": importlib.import_module("endpoints.lyric_search").LyricSearch(
            app, util, constants
        ),
        "yt": importlib.import_module("endpoints.yt").YT(app, util, constants),
        "radio": importlib.import_module("endpoints.radio").Radio(
            app, util, constants, loop
        ),
        "meme": importlib.import_module("endpoints.meme").Meme(app, util, constants),
        "trip": importlib.import_module("endpoints.rip").RIP(app, util, constants),
        "auth": importlib.import_module("endpoints.auth").Auth(app),
        "lighting": importlib.import_module("endpoints.lighting").Lighting(
            app, util, constants
        ),
    }
)

# Misc endpoint depends on radio endpoint instance
radio_endpoint = _routes.get("radio")
if radio_endpoint:
    _routes["misc"] = importlib.import_module("endpoints.misc").Misc(
        app, util, constants, radio_endpoint
    )

"""
End Actionable Routes
"""

"""
Startup
"""

redis = redis_cache.RedisCache()
loop.create_task(redis.create_index())