103 lines
2.8 KiB
Python
103 lines
2.8 KiB
Python
import importlib
|
|
import sys
|
|
sys.path.insert(0, ".")
|
|
import logging
|
|
import asyncio
|
|
from typing import Any
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from lyric_search.sources import redis_cache
|
|
|
|
logger = logging.getLogger()
|
|
logger.setLevel(logging.INFO)
|
|
|
|
loop = asyncio.get_event_loop()
|
|
app = FastAPI(title="codey.lol API",
|
|
version="1.0",
|
|
contact={
|
|
'name': 'codey'
|
|
},
|
|
redirect_slashes=False,
|
|
loop=loop)
|
|
|
|
|
|
constants = importlib.import_module("constants").Constants()
|
|
util = importlib.import_module("util").Utilities(app, constants)
|
|
|
|
origins = [
|
|
"https://codey.lol",
|
|
"https://api.codey.lol"
|
|
]
|
|
|
|
app.add_middleware(CORSMiddleware, # type: ignore
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=["POST", "GET", "HEAD"],
|
|
allow_headers=["*"]) # type: ignore
|
|
|
|
"""
|
|
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']
|
|
if not (
|
|
isinstance(path, str)
|
|
and
|
|
path.split("/", maxsplit=1) == "widget"
|
|
):
|
|
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: dict = {
|
|
'randmsg': importlib.import_module("endpoints.rand_msg").RandMsg(app, util, constants),
|
|
'transcriptions': importlib.import_module("endpoints.transcriptions").Transcriptions(app, util, constants),
|
|
'ai': importlib.import_module("endpoints.ai").AI(app, util, constants),
|
|
'lyrics': importlib.import_module("endpoints.lyric_search").LyricSearch(app, util, constants),
|
|
'lastfm': importlib.import_module("endpoints.lastfm").LastFM(app, util, constants),
|
|
'yt': importlib.import_module("endpoints.yt").YT(app, util, constants),
|
|
'xc': importlib.import_module("endpoints.xc").XC(app, util, constants),
|
|
'karma': importlib.import_module("endpoints.karma").Karma(app, util, constants),
|
|
'radio': importlib.import_module("endpoints.radio").Radio(app, util, constants),
|
|
'mgr': importlib.import_module("endpoints.mgr.mgr_test").Mgr(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()) |