codey/api
1
0
Fork 0
Code Issues 1 Pull Requests Activity

Compare commits

base: codey:ae419a2cbf2387cea290f17530ed736a1da4efdb
Branches Tags
codey:main
...
compare: codey:a61970d2989c25c9cb60ddc44541d65e97d705d1
Branches Tags
codey:main

2 Commits
ae419a2cbf ... a61970d298

Author SHA1 Message Date
codey
a61970d298 . 2025-10-03 11:34:48 -04:00
codey
fb94750b46 - Removed unnecessary/unrelated files 
- Added Scalar
 2025-10-03 11:34:33 -04:00
7 changed files with 20 additions and 1825 deletions
Expand all files Collapse all files

3
.gitignore vendored
View File

@@ -30,7 +30,8 @@ job_review.py
check_missing.py
**/auth/*
**/radio_api/*
**/test/*
test/db_stats.py
test/report/*
.gitignore
.env
.env

5
README.md
View File

@@ -4,7 +4,10 @@ A modern FastAPI-based backend providing various endpoints for media, authentica
## Overview
This server is built with [FastAPI](https://fastapi.tiangolo.com/) and provides a comprehensive API for multiple services. The interactive API documentation is available at the [Swagger UI](https://api.codey.lol/docs).
This server is built with [FastAPI](https://fastapi.tiangolo.com/) and provides a comprehensive API for multiple services. API documentation is available in three formats:
- **Swagger UI**: [https://api.codey.lol/docs](https://api.codey.lol/docs) - Classic interactive API explorer with "Try it out" functionality
- **Scalar**: [https://api.codey.lol/scalar](https://api.codey.lol/scalar) - Modern, fast interactive API documentation (recommended)
- **ReDoc**: [https://api.codey.lol/redoc](https://api.codey.lol/redoc) - Clean, read-only documentation with better visual design
## API Endpoints

16
base.py
View File

@@ -7,6 +7,7 @@ import asyncio
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)
@@ -21,6 +22,8 @@ app = FastAPI(
contact={"name": "codey"},
redirect_slashes=False,
loop=loop,
docs_url="/docs", # Swagger UI (default)
redoc_url="/redoc", # ReDoc UI (default, but explicitly set)
)
constants = importlib.import_module("constants").Constants()
@@ -43,6 +46,13 @@ app.add_middleware(
allow_headers=["*"],
) # type: ignore
# Add Scalar API documentation endpoint (before blacklist routes)
@app.get("/scalar", include_in_schema=False)
def scalar_docs():
return get_scalar_api_reference(
openapi_url="/openapi.json",
title="codey.lol API"
)
"""
Blacklisted routes
@@ -62,10 +72,13 @@ def base_head():
@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", "scalar", "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) == "widget" or path == "misc/no")
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)
@@ -117,7 +130,6 @@ if radio_endpoint:
End Actionable Routes
"""
"""
Startup
"""

1299
test/add_cover_art.py
View File

File diff suppressed because it is too large Load Diff

252
test/liquidsoap.liq
View File

@@ -1,252 +0,0 @@
#!/usr/bin/env liquidsoap
set("log.file.path","/home/kyle/.lsl.txt")
set("log.stdout",true)
set("harbor.bind_addrs", ["127.0.0.1"])
# Buffer and timing settings
set("frame.duration",0.02)
set("root.max_latency",2.)
set("audio.converter.samplerate.libsamplerate.quality","best")
set("clock.allow_streaming_errors",false)
# Get next track dynamically [Each station]
def get_next_main() =
uri = list.hd(default="", process.read.lines("uv run get_next_track.py main"))
[request.create(uri)]
end
def get_next_rock() =
uri = list.hd(default="", process.read.lines("uv run get_next_track.py rock"))
[request.create(uri)]
end
def get_next_electronic() =
uri = list.hd(default="", process.read.lines("uv run get_next_track.py electronic"))
[request.create(uri)]
end
def get_next_rap() =
uri = list.hd(default="", process.read.lines("uv run get_next_track.py rap"))
[request.create(uri)]
end
#def get_next_classical() =
# uri = list.hd(default="", process.read.lines("uv run get_next_track.py classical"))
# [request.create(uri)]
#end
def get_next_pop() =
uri = list.hd(default="", process.read.lines("uv run get_next_track.py pop"))
[request.create(uri)]
end
# Set up queues [Each station]
main_list = request.dynamic(
id="requests",
get_next_main,
retry_delay=1.0,
timeout=20.0
)
rock_list = request.dynamic(
id="rock_requests",
get_next_rock,
retry_delay=1.0,
timeout=20.0
)
electronic_list = request.dynamic(
id="electronic_requests",
get_next_electronic,
retry_delay=1.0,
timeout=20.0
)
rap_list = request.dynamic(
id="rap_requests",
get_next_rap,
retry_delay=1.0,
timeout=20.0
)
#classical_list = request.dynamic.list(
# id="classical_requests",
# get_next_classical,
# prefetch=0
#)
pop_list = request.dynamic(
id="pop_requests",
get_next_pop,
retry_delay=1.0,
timeout=20.0
)
# Standard
silence = single("/home/kyle/ls/silence.ogg")
# Queue [Each station]
def main_queue(remaining, _) =
log("MAIN: Queueing with #{remaining} seconds remaining")
if not main_list.fetch() then
log("Fetching next query failed")
end
end
def rock_queue(remaining, _) =
log("ROCK: Queueing with #{remaining} seconds remaining")
if not rock_list.fetch() then
log("Fetching next query failed")
end
end
def electronic_queue(remaining, _) =
log("ELECTRONIC: Queueing with #{remaining} seconds remaining")
if not electronic_list.fetch() then
log("Fetching next query failed")
end
end
def rap_queue(remaining, _) =
log("RAP: Queueing with #{remaining} seconds remaining")
if not rap_list.fetch() then
log("Fetching next query failed")
end
end
#def classical_queue(remaining, _) =
# log("CLASSICAL: Queueing with #{remaining} seconds remaining")
# if not classical_list.fetch() then
# log("Fetching next query failed")
# end
#end
def pop_queue(remaining, _) =
log("POP: Queueing with #{remaining} seconds remaining")
if not pop_list.fetch() then
log("Fetching next query failed")
end
end
# Initial fetch [Each station]
main_list.fetch()
rock_list.fetch()
electronic_list.fetch()
rap_list.fetch()
#classical_list.fetch()
pop_list.fetch()
# Source setup [Each station]
def create_source(s,q) =
source.dynamic(s, track_sensitive=true, {q()})
end
main_source = create_source(main_list, main_queue)
rock_source = create_source(rock_list, rock_queue)
electronic_source = create_source(electronic_list, electronic_queue)
rap_source = create_source(rap_list, rap_queue)
#classical_source = create_source(classical_list, classical_queue)
pop_source = create_source(pop_list, pop_queue)
all_tracks_main = fallback(track_sensitive=false, [main_source, silence])
all_tracks_rock = fallback(track_sensitive=false, [rock_source, silence])
all_tracks_electronic = fallback(track_sensitive=false, [electronic_source, silence])
all_tracks_rap = fallback(track_sensitive=false, [rap_source, silence])
#all_tracks_classical = fallback(track_sensitive=false, [classical_source, silence])
all_tracks_pop = fallback(track_sensitive=false, [pop_source, silence])
# HLS Setup [Standard]
aac_lofi = %ffmpeg(format="mpegts",
%audio(codec="aac",
channels=2,
ar=48000,
b="128k"))
aac_midfi = %ffmpeg(format="mpegts",
%audio(codec="aac",
channels=2,
ar=48000,
b="256k"))
aac_hifi = %ffmpeg(format="mpegts",
%audio(codec="aac",
channels=2,
ar=48000,
b="512k"))
streams =
[("aac_lofi", aac_lofi), ("aac_midfi", aac_midfi), ("aac_hifi", aac_hifi)]
# HLS Outputs [Each station]
def create_hls_output(~name, source) =
output.file.hls(
playlist="#{name}.m3u8",
segment_duration=0.5,
segments=10,
segments_overhead=5,
persist_at="/nvme/pub/hls/#{name}/state.config",
"/nvme/pub/hls/#{name}",
streams,
source
)
end
create_hls_output(name="main", mksafe(main_source))
create_hls_output(name="rock", mksafe(rock_source))
create_hls_output(name="electronic", mksafe(electronic_source))
create_hls_output(name="rap", mksafe(rap_source))
#output.file.hls(
# playlist="classical.m3u8",
# segment_duration=0.45,
# segments=9,
# segments_overhead=3,
# persist_at="/nvme/pub/hls/classical_state.config",
# "/nvme/pub/hls/classical",
# streams,
# mksafe(classical_source)
#)
create_hls_output(name="pop", mksafe(pop_source))
# HTTP Server
def get_next_http(~protocol,~data,~headers,uri) =
source =
if data == "main" then main_source
elsif data == "rock" then rock_source
elsif data == "electronic" then electronic_source
elsif data == "rap" then rap_source
elsif data == "pop" then pop_source
else null() end
if source != null() then
source.skip(source)
http.response(
protocol=protocol,
code=200,
data="OK #{data}"
)
end
harbor.http.register(port=29000, method="POST", "/next", get_next_http)
# EOF

270
test/liquidsoap.ls
View File

@@ -1,270 +0,0 @@
#!/usr/bin/liquidsoap
set("log.file.path", "/home/kyle/.lsl.txt")
set("log.stdout", true)
set("harbor.bind_addrs", ["127.0.0.1"])
# Get next track dynamically [Each station]
def get_next_main() =
uri = list.hd(default="", process.read.lines("uv run get_next_track.py main"))
[request.create(uri)]
end
def get_next_rock() =
uri = list.hd(default="", process.read.lines("uv run get_next_track.py rock"))
[request.create(uri)]
end
def get_next_electronic() =
uri = list.hd(default="", process.read.lines("uv run get_next_track.py electronic"))
[request.create(uri)]
end
def get_next_rap() =
uri = list.hd(default="", process.read.lines("uv run get_next_track.py rap"))
[request.create(uri)]
end
#def get_next_classical() =
# uri = list.hd(default="", process.read.lines("uv run get_next_track.py classical"))
# [request.create(uri)]
#end
def get_next_pop() =
uri = list.hd(default="", process.read.lines("uv run get_next_track.py pop"))
[request.create(uri)]
end
# Set up queues [Each station]
main_list = request.dynamic.list(
id="requests",
get_next_main,
prefetch=0
)
rock_list = request.dynamic.list(
id="rock_requests",
get_next_rock,
prefetch=0
)
electronic_list = request.dynamic.list(
id="electronic_requests",
get_next_electronic,
prefetch=0
)
rap_list = request.dynamic.list(
id="rap_requests",
get_next_rap,
prefetch=0
)
#classical_list = request.dynamic.list(
# id="classical_requests",
# get_next_classical,
# prefetch=0
#)
pop_list = request.dynamic.list(
id="pop_requests",
get_next_pop,
prefetch=0
)
# Standard
silence = single("/home/kyle/ls/silence.ogg")
# Queue [Each station]
def main_queue(remaining, _) =
log("MAIN: Queueing with #{remaining} seconds remaining")
if not main_list.fetch() then
log("Fetching next query failed")
end
end
def rock_queue(remaining, _) =
log("ROCK: Queueing with #{remaining} seconds remaining")
if not rock_list.fetch() then
log("Fetching next query failed")
end
end
def electronic_queue(remaining, _) =
log("ELECTRONIC: Queueing with #{remaining} seconds remaining")
if not electronic_list.fetch() then
log("Fetching next query failed")
end
end
def rap_queue(remaining, _) =
log("RAP: Queueing with #{remaining} seconds remaining")
if not rap_list.fetch() then
log("Fetching next query failed")
end
end
#def classical_queue(remaining, _) =
# log("CLASSICAL: Queueing with #{remaining} seconds remaining")
# if not classical_list.fetch() then
# log("Fetching next query failed")
# end
#end
def pop_queue(remaining, _) =
log("POP: Queueing with #{remaining} seconds remaining")
if not pop_list.fetch() then
log("Fetching next query failed")
end
end
# Initial fetch [Each station]
main_list.fetch()
rock_list.fetch()
electronic_list.fetch()
rap_list.fetch()
#classical_list.fetch()
pop_list.fetch()
# Source setup [Each station]
main_source = source.on_end(delay=1.0, main_list, main_queue)
rock_source = source.on_end(delay=1.0, rock_list, rock_queue)
electronic_source = source.on_end(delay=1.0, electronic_list, electronic_queue)
rap_source = source.on_end(delay=1.0, rap_list, rap_queue)
#classical_source = source.on_end(delay=1.0, classical_list, classical_queue)
pop_source = source.on_end(delay=1.0, pop_list, pop_queue)
all_tracks_main = fallback(track_sensitive=false, [main_source, silence])
all_tracks_rock = fallback(track_sensitive=false, [rock_source, silence])
all_tracks_electronic = fallback(track_sensitive=false, [electronic_source, silence])
all_tracks_rap = fallback(track_sensitive=false, [rap_source, silence])
#all_tracks_classical = fallback(track_sensitive=false, [classical_source, silence])
all_tracks_pop = fallback(track_sensitive=false, [pop_source, silence])
# HLS Setup [Standard]
aac_lofi =
%ffmpeg(format = "mpegts", %audio(codec = "aac", channels = 2, ar = 44100))
aac_midfi =
%ffmpeg(
format = "mpegts",
%audio(codec = "aac", channels = 2, ar = 44100, b = "96k")
)
aac_hifi =
%ffmpeg(
format = "mpegts",
%audio(codec = "aac", channels = 2, ar = 44100, b = "448k")
)
streams =
[("aac_lofi", aac_lofi), ("aac_midfi", aac_midfi), ("aac_hifi", aac_hifi)]
# HLS Outputs [Each station]
output.file.hls(
playlist="main.m3u8",
segment_duration=0.5,
segments=9,
segments_overhead=4,
persist_at="/nvme/pub/hls/state.config",
"/nvme/pub/hls/main",
streams,
mksafe(main_source)
)
output.file.hls(
playlist="rock.m3u8",
segment_duration=0.5,
segments=9,
segments_overhead=4,
persist_at="/nvme/pub/hls/rock/state.config",
"/nvme/pub/hls/rock",
streams,
mksafe(rock_source)
)
output.file.hls(
playlist="electronic.m3u8",
segment_duration=0.5,
segments=9,
segments_overhead=4,
persist_at="/nvme/pub/hls/electronic/state.config",
"/nvme/pub/hls/electronic",
streams,
mksafe(electronic_source)
)
output.file.hls(
playlist="rap.m3u8",
segment_duration=0.5,
segments=9,
segments_overhead=4,
persist_at="/nvme/pub/hls/rap_state.config",
"/nvme/pub/hls/rap",
streams,
mksafe(rap_source)
)
#output.file.hls(
# playlist="classical.m3u8",
# segment_duration=0.45,
# segments=9,
# segments_overhead=3,
# persist_at="/nvme/pub/hls/classical_state.config",
# "/nvme/pub/hls/classical",
# streams,
# mksafe(classical_source)
#)
output.file.hls(
playlist="pop.m3u8",
segment_duration=0.5,
segments=9,
segments_overhead=4,
persist_at="/nvme/pub/hls/pop_state.config",
"/nvme/pub/hls/pop",
streams,
mksafe(pop_source)
)
# HTTP Server
def get_next_http(~protocol,~data,~headers,uri) =
if data == "main" then
_req = source.skip(main_source)
elsif data == "rock" then
_req = source.skip(rock_source)
elsif data == "electronic" then
_req = source.skip(electronic_source)
elsif data == "rap" then
_req = source.skip(rap_source)
#elsif data == "classical" then
# _req = source.skip(classical_source)
elsif data == "pop" then
_req = source.skip(pop_source)
end
http.response(
protocol=protocol,
code=200,
data="OK #{data}"
)
end
harbor.http.register(port=29000, method="POST", "/next", get_next_http)
# EOF

0
test/minimal_test.liq
View File