mirror of
https://github.com/migatu/conjurer.git
synced 2026-07-14 21:38:38 +00:00
split: betoniarka (radio operator) colocated with liquidsoap; musician goes Discord-only
Permissions post-mortem that motivated this: the musician wrote radio playlists AS ROOT onto a ROOT-OWNED network share which liquidsoap then read AS USER 'radio' - chown fails on such shares by design (root squash / uid mapping), so the radio came up and died on the playlists. The fix is structural: the playlist WRITER now lives in the same container as the READER, as the same user, on a local volume. No shared partition, no chown, no uid mapping. New: conjurer_betoniarka/betoniarka.py - runs inside the radio container (started by the entrypoint as user 'radio', port 5005): - library scan -> all_playlist/hit playlists with LOCAL container paths (start + every 24h + authenticated GET /rescan) - bot-facing radio API moved from the musician: /add_to_priority, /create_priority_playlist, /request_radio_file, /clear_pr_pls, plus GET /ping (health) and /stream (web page) - radio_log/persistence tailer forwarding play events to the bot's /prepped_tracks with the shared API key (bot-unreachable = logged, not fatal) Musician: pure Discord music player now - keeps /mp3, /update_mp3, /get_music and the file-share endpoints; all radio playlist writing, radio paths/env and the tailer removed. Bot: new CONJURER_RADIO_SERVICE (defaults to CONJURER_FILE_SERVICE so un-split deployments keep working); radio_commands targets it; separate 'radio' health-gate group on betoniarka /ping (musician group now covers music_commands + file_search_commands only). Docker: betoniarka baked into the radio image (python3 + flask/waitress/ requests from Debian debs), port 5005 exposed, entrypoint starts it via setpriv as 'radio'; data-volume chown is now best-effort with a loud warning (keep the volume local); docs get the post-mortem + wiring. Verified: py_compile everything; functional stub tests - rescan writes local-path playlists, wyszukaj scores and appends to priority, auth 401/ok, tailer forward carries the API key. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Michal Tuszowski
@@ -0,0 +1,353 @@
|
||||
"""Betoniarka - the radio operator service.
|
||||
|
||||
Lives in the SAME container as Liquidsoap and runs as the SAME unprivileged
|
||||
user ('radio'), which is the whole point: the process that writes the radio
|
||||
playlists is colocated with the process that watches them, so there is no
|
||||
cross-host permission juggling (root-owned network shares, failing chowns)
|
||||
anymore.
|
||||
|
||||
Responsibilities (extracted from conjurer_musician, which is now a pure
|
||||
Discord music player):
|
||||
- scan the local music library into all_playlist.playlist / hit.playlist
|
||||
- serve the radio-management HTTP API the bot calls
|
||||
(/add_to_priority, /create_priority_playlist, /request_radio_file,
|
||||
/clear_pr_pls) plus /ping for health checks and /stream for the web page
|
||||
- tail radio_log.log / persistence.log and forward "now playing" events to
|
||||
the bot's /prepped_tracks
|
||||
"""
|
||||
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import random
|
||||
import re
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
from typing import Dict, List
|
||||
|
||||
import requests
|
||||
from flask import Flask, abort, jsonify, request, send_file
|
||||
from waitress import serve
|
||||
|
||||
|
||||
def _env(name: str, default: str) -> str:
|
||||
return os.getenv(name, default)
|
||||
|
||||
|
||||
API_KEY = os.getenv("CONJURER_API_KEY")
|
||||
MAIN_BOT_ADDRESS = _env("CONJURER_MAIN_BOT", "http://127.0.0.1:5000")
|
||||
MUSIC_TRACKER = _env("CONJURER_MUSIC_TRACKER_ENDPOINT", "/prepped_tracks")
|
||||
HOST_ADDRESS = _env("BETONIARKA_HOST", "0.0.0.0")
|
||||
PORT_ADDRESS = int(_env("BETONIARKA_PORT", "5005"))
|
||||
|
||||
DATA_DIR = Path(_env("BETONIARKA_DATA", "/srv/betoniarka/data"))
|
||||
MUSIC_FOLDER = Path(_env("BETONIARKA_MUSIC", "/srv/betoniarka/music"))
|
||||
PRIORITY_FOLDER = Path(_env("BETONIARKA_PRIORITY_FOLDER", str(MUSIC_FOLDER / "priority")))
|
||||
STREAM_TEMPLATE = _env("BETONIARKA_STREAM_TEMPLATE", "/app/stream.html")
|
||||
RESCAN_SECONDS = int(_env("BETONIARKA_RESCAN_SECONDS", str(24 * 60 * 60)))
|
||||
# How many leading path tokens to ignore when keyword-matching
|
||||
# (/srv/betoniarka/music/... -> '', 'srv', 'betoniarka', 'music').
|
||||
PATH_SKIP = int(_env("BETONIARKA_PATH_SKIP", "4"))
|
||||
|
||||
ALL_PLAYLIST_PATH = DATA_DIR / "all_playlist.playlist"
|
||||
HIT_PLAYLIST_PATH = DATA_DIR / "hit.playlist"
|
||||
REQUEST_PLAYLIST_PATH = DATA_DIR / "request.playlist"
|
||||
PRIORITY_PLAYLIST_PATH = DATA_DIR / "priority_queue.playlist"
|
||||
RADIOLOG_PATH = DATA_DIR / "radio_log.log"
|
||||
PERSISTENCE_PATH = DATA_DIR / "persistence.log"
|
||||
|
||||
ENCODING = _env("CONJURER_ENCODING", "utf-8")
|
||||
|
||||
logger = logging.getLogger("betoniarka")
|
||||
|
||||
music_file_list: List[str] = []
|
||||
priority_list: List[str] = []
|
||||
|
||||
app = Flask(__name__)
|
||||
|
||||
|
||||
def _build_headers() -> Dict[str, str]:
|
||||
if API_KEY:
|
||||
return {"X-Conjurer-Api-Key": API_KEY}
|
||||
return {}
|
||||
|
||||
|
||||
def _authorize_request() -> None:
|
||||
if API_KEY and request.headers.get("X-Conjurer-Api-Key") != API_KEY:
|
||||
abort(401)
|
||||
|
||||
|
||||
def _post_to_bot(payload: List[str]) -> None:
|
||||
try:
|
||||
response = requests.post(
|
||||
f"{MAIN_BOT_ADDRESS}{MUSIC_TRACKER}",
|
||||
json=payload,
|
||||
headers=_build_headers(),
|
||||
timeout=60,
|
||||
)
|
||||
logger.info("Forwarded to bot (%s): %s", response.status_code, payload[0])
|
||||
except requests.exceptions.RequestException as exc:
|
||||
logger.warning("Bot unreachable, dropping %s: %s", payload[0], exc)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- library
|
||||
def rescan():
|
||||
"""Scan the local library into the playlists Liquidsoap watches.
|
||||
|
||||
Paths written here are LOCAL container paths, the same ones Liquidsoap
|
||||
resolves - no shared network filesystem involved.
|
||||
"""
|
||||
logger.info("Rescan triggered")
|
||||
music_file_list.clear()
|
||||
priority_list.clear()
|
||||
|
||||
for mp3_item in MUSIC_FOLDER.glob("**/*.mp3"):
|
||||
music_file_list.append(mp3_item.as_posix())
|
||||
for mp3_item in PRIORITY_FOLDER.glob("**/*.mp3"):
|
||||
priority_list.append(mp3_item.as_posix())
|
||||
|
||||
with ALL_PLAYLIST_PATH.open("w", encoding=ENCODING) as w_file:
|
||||
for item in music_file_list:
|
||||
w_file.write(item + "\n")
|
||||
with HIT_PLAYLIST_PATH.open("w", encoding=ENCODING) as w_file:
|
||||
for item in priority_list:
|
||||
w_file.write(item + "\n")
|
||||
logger.info("Rescan done: %d tracks, %d hits", len(music_file_list), len(priority_list))
|
||||
|
||||
|
||||
def thread_rescan():
|
||||
while True:
|
||||
time.sleep(RESCAN_SECONDS)
|
||||
rescan()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- search
|
||||
def remove_characters(string, character):
|
||||
return string.replace(character, "")
|
||||
|
||||
|
||||
def max_weight(lista):
|
||||
maximum_weight = 0
|
||||
for iterator in lista:
|
||||
if iterator[0] > maximum_weight:
|
||||
maximum_weight = iterator[0]
|
||||
return maximum_weight
|
||||
|
||||
|
||||
_CHAR_REMOVE = [
|
||||
".", "^", "$", "*", "+", "?", "{", "}", "[", "]",
|
||||
"\\", "/", "|", "(", ")", "!", ",", "-", ":", "mp3",
|
||||
]
|
||||
|
||||
|
||||
def wyszukaj(word_list, how_many, _logger=None, write_to=None):
|
||||
"""Keyword-score the library; optionally append hits to a playlist file.
|
||||
|
||||
Ported unchanged from the musician (same scoring), minus the win32
|
||||
branches. ``write_to`` replaces the old ``return_to_bot`` flag: pass a
|
||||
playlist Path to append the result, or None to just return it.
|
||||
"""
|
||||
fun_logger = _logger or logger
|
||||
search_weight = [(0, "") for _ in range(len(music_file_list))]
|
||||
|
||||
time_start = datetime.now()
|
||||
skip_start = 2 if int(how_many) > 0 else 1
|
||||
for word in word_list[skip_start:]:
|
||||
token_weight = len(word)
|
||||
fun_logger.info("Słowo kluczowe: %s", word)
|
||||
for itr, file in enumerate(music_file_list):
|
||||
parts = file.split("/")
|
||||
all_words = []
|
||||
for f_iter in parts:
|
||||
for char in _CHAR_REMOVE:
|
||||
f_iter = remove_characters(f_iter, char)
|
||||
all_words.extend(f_iter.split())
|
||||
pingu = 1
|
||||
pattern_len = len(all_words)
|
||||
matched_times = 1
|
||||
for itm in all_words[PATH_SKIP:]:
|
||||
pingu += 1
|
||||
if re.match(".*" + word + ".*", itm, re.IGNORECASE):
|
||||
temp_weight = (
|
||||
search_weight[itr][0]
|
||||
+ (token_weight + (pingu**1.5) / pattern_len) / matched_times
|
||||
)
|
||||
search_weight[itr] = (temp_weight, music_file_list[itr])
|
||||
matched_times += 1
|
||||
|
||||
fun_logger.info("Stworzylem tablice wag zajęło mi to %s", datetime.now() - time_start)
|
||||
best = max_weight(search_weight)
|
||||
if best == 0:
|
||||
return []
|
||||
|
||||
return_list = []
|
||||
if int(how_many) <= 0:
|
||||
for weight, path in search_weight:
|
||||
if weight == best:
|
||||
return_list.append((weight, path))
|
||||
break
|
||||
else:
|
||||
search_weight.sort(key=lambda x: x[0], reverse=True)
|
||||
return_list.extend(search_weight[: int(how_many)])
|
||||
|
||||
if write_to is not None:
|
||||
with write_to.open("a", encoding=ENCODING) as s_file:
|
||||
for item in return_list:
|
||||
s_file.write(item[1] + "\n")
|
||||
fun_logger.info("Done: %s", return_list)
|
||||
return return_list
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- tailer
|
||||
def scan_tracks():
|
||||
"""Tail the radio logs and forward play events to the bot."""
|
||||
with open(RADIOLOG_PATH, "r", encoding=ENCODING) as log_file:
|
||||
log_file.seek(os.stat(RADIOLOG_PATH).st_size)
|
||||
prev_size = os.stat(PERSISTENCE_PATH).st_size
|
||||
|
||||
while True:
|
||||
current_size = os.stat(PERSISTENCE_PATH).st_size
|
||||
if prev_size != current_size:
|
||||
while prev_size != current_size:
|
||||
prev_size = current_size
|
||||
time.sleep(0.1)
|
||||
current_size = os.stat(PERSISTENCE_PATH).st_size
|
||||
with open(PERSISTENCE_PATH, "r", encoding=ENCODING) as persistence:
|
||||
lines = persistence.readlines()
|
||||
if len(lines) >= 3:
|
||||
_post_to_bot(["next", lines[2]])
|
||||
|
||||
position = log_file.tell()
|
||||
line = log_file.readline()
|
||||
if not line:
|
||||
time.sleep(1)
|
||||
log_file.seek(position)
|
||||
continue
|
||||
|
||||
if not re.match(r".*Prepared.*", line):
|
||||
time.sleep(0.1)
|
||||
continue
|
||||
|
||||
result = None
|
||||
if re.match(r".*jingles.*", line):
|
||||
result = ["jingles", line]
|
||||
elif re.match(r".*priority.*", line):
|
||||
result = ["priority", line]
|
||||
elif re.match(r".*hit.*", line):
|
||||
result = ["hit", line]
|
||||
elif re.match(r".*all_playlist.*", line):
|
||||
result = ["all", line]
|
||||
elif re.match(r".*request.*", line):
|
||||
result = ["requests", line]
|
||||
|
||||
if result:
|
||||
logger.info("Forwarding radio log entry: %s", result[0])
|
||||
_post_to_bot(result)
|
||||
|
||||
time.sleep(0.1)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------- routes
|
||||
@app.route("/ping", methods=["GET"])
|
||||
def ping():
|
||||
"""Health check - the bot gates radio_commands on this answering."""
|
||||
return jsonify("ALIVE")
|
||||
|
||||
|
||||
@app.route("/stream", methods=["GET"])
|
||||
def stream_page():
|
||||
return send_file(STREAM_TEMPLATE)
|
||||
|
||||
|
||||
@app.route("/clear_pr_pls", methods=["GET"])
|
||||
def clear_pr_pls():
|
||||
_authorize_request()
|
||||
app.logger.info("CLEARING PLAYLIST")
|
||||
with PRIORITY_PLAYLIST_PATH.open("w", encoding=ENCODING) as cleared_pl:
|
||||
cleared_pl.write("")
|
||||
return jsonify(isError=False, message="Success", statusCode=200, data=[]), 200
|
||||
|
||||
|
||||
@app.route("/rescan", methods=["GET"])
|
||||
def manual_rescan():
|
||||
_authorize_request()
|
||||
rescan()
|
||||
return jsonify(isError=False, message="Success", statusCode=200,
|
||||
data={"tracks": len(music_file_list)}), 200
|
||||
|
||||
|
||||
@app.route("/request_radio_file", methods=["POST"])
|
||||
def add_request():
|
||||
_authorize_request()
|
||||
record = json.loads(request.data)
|
||||
app.logger.info(record)
|
||||
wyszukaj(record["lista_slow"], 0, app.logger, write_to=REQUEST_PLAYLIST_PATH)
|
||||
return jsonify(isError=False, message="Success", statusCode=200,
|
||||
data={"status": "OK"}), 200
|
||||
|
||||
|
||||
@app.route("/create_priority_playlist", methods=["POST"])
|
||||
def create_priority_playlist():
|
||||
_authorize_request()
|
||||
record = json.loads(request.data)
|
||||
app.logger.info(record)
|
||||
return_data = wyszukaj(
|
||||
record["lista_slow"], record["dlugosc_plejlisty"], app.logger, write_to=None
|
||||
)
|
||||
random.shuffle(return_data)
|
||||
# NOTE: appends to the REQUEST playlist - behaviour inherited verbatim
|
||||
# from the musician implementation (the request queue picks it up).
|
||||
with REQUEST_PLAYLIST_PATH.open("a", encoding=ENCODING) as s_file:
|
||||
for item in return_data:
|
||||
s_file.write(item[1] + "\n")
|
||||
return jsonify(isError=False, message="Success", statusCode=200,
|
||||
data={"status": "OK"}), 200
|
||||
|
||||
|
||||
@app.route("/add_to_priority", methods=["POST"])
|
||||
def add_to_priority():
|
||||
_authorize_request()
|
||||
record = json.loads(request.data)
|
||||
app.logger.info(record)
|
||||
wyszukaj(
|
||||
record["lista_slow"], record["dlugosc_plejlisty"], app.logger,
|
||||
write_to=PRIORITY_PLAYLIST_PATH,
|
||||
)
|
||||
return jsonify(isError=False, message="Success", statusCode=200,
|
||||
data={"status": "OK"}), 200
|
||||
|
||||
|
||||
def waitress_run():
|
||||
serve(app, host=HOST_ADDRESS, port=PORT_ADDRESS)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
logger.setLevel(logging.DEBUG)
|
||||
console = logging.StreamHandler()
|
||||
console.setFormatter(
|
||||
logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
|
||||
)
|
||||
logger.addHandler(console)
|
||||
|
||||
rescan()
|
||||
logger.info("Betoniarka started on %s:%s", HOST_ADDRESS, PORT_ADDRESS)
|
||||
|
||||
threads = [
|
||||
threading.Thread(target=waitress_run, daemon=True),
|
||||
threading.Thread(target=thread_rescan, daemon=True),
|
||||
]
|
||||
for worker in threads:
|
||||
worker.start()
|
||||
|
||||
time.sleep(5)
|
||||
track_thread = threading.Thread(target=scan_tracks, daemon=True)
|
||||
track_thread.start()
|
||||
|
||||
try:
|
||||
for worker in threads:
|
||||
worker.join()
|
||||
track_thread.join()
|
||||
except KeyboardInterrupt:
|
||||
logger.info("Shutdown requested - exiting betoniarka")
|
||||
Reference in New Issue
Block a user