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