api/base.py

import importlib
import sys

sys.path.insert(0, ".")
import logging
import asyncio
from contextlib import asynccontextmanager
from typing import Any
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from scalar_fastapi import get_scalar_api_reference
from lyric_search.sources import redis_cache

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
    
    # 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()
    
    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():
    return get_scalar_api_reference(openapi_url="/openapi.json", title="codey.lol API")


"""
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())