""" The provided Python script sets up a Flask web server to manage a list of music files, with functions for rescanning the music folder, updating the music list, and serving the music list via API endpoints. """ import json import logging import os import random import re import threading import time from datetime import datetime from logging import handlers from pathlib import Path from typing import Dict, List import requests from flask import ( Flask, abort, jsonify, redirect, render_template, request, send_from_directory, ) from waitress import serve import media_search_functions def _env(name: str, default: str) -> str: return os.getenv(name, default) def _env_path(name: str, default: str) -> Path: value = os.getenv(name, default) return Path(value).expanduser().resolve() 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("CONJURER_MUSICIAN_HOST", "0.0.0.0") PORT_ADDRESS = int(_env("CONJURER_MUSICIAN_PORT", "5000")) BASE_DIR = Path( os.getenv("CONJURER_MUSICIAN_BASE", str(Path(__file__).resolve().parent)) ) LOGFILE = _env_path( "CONJURER_MUSICIAN_LOG", str(BASE_DIR / "discord_mus_service.log") ) LOGSTORE = _env_path("CONJURER_LOGSTORE", str(BASE_DIR / "logs")) MUSIC_FOLDER = _env_path( "CONJURER_MUSIC_FOLDER", str(BASE_DIR / "music") ) PRIORITY_FOLDER = _env_path( "CONJURER_PRIORITY_FOLDER", str(MUSIC_FOLDER / "priority") ) RADIOLOG_PATH = _env_path( "CONJURER_RADIO_LOG", str(BASE_DIR / "radio_log.log") ) PERSISTENCE_PATH = _env_path( "CONJURER_PERSISTENCE_LOG", str(BASE_DIR / "persistence.log") ) ALL_PLAYLIST_PATH = _env_path( "CONJURER_ALL_PLAYLIST", str(BASE_DIR / "all_playlist.playlist") ) HIT_PLAYLIST_PATH = _env_path( "CONJURER_HIT_PLAYLIST", str(BASE_DIR / "hit.playlist") ) REQUEST_PLAYLIST_PATH = _env_path( "CONJURER_REQUEST_PLAYLIST", str(BASE_DIR / "request.playlist") ) PRIORITY_PLAYLIST_PATH = _env_path( "CONJURER_PRIORITY_PLAYLIST", str(BASE_DIR / "priority_queue.playlist") ) STREAM_TEMPLATE = _env_path( "CONJURER_STREAM_TEMPLATE", str(BASE_DIR / "stream.html") ) ENCODING = _env("CONJURER_ENCODING", "utf-8") SEPARATOR_FILE_PATH = os.sep for playlist_path in ( ALL_PLAYLIST_PATH, HIT_PLAYLIST_PATH, REQUEST_PLAYLIST_PATH, PRIORITY_PLAYLIST_PATH, ): playlist_path.parent.mkdir(parents=True, exist_ok=True) random.seed() music_file_list: List[str] = [] priority_list: List[str] = [] def _build_headers() -> Dict[str, str]: headers: Dict[str, str] = {} if API_KEY: headers["X-Conjurer-Api-Key"] = API_KEY return headers 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: response = requests.post( f"{MAIN_BOT_ADDRESS}{MUSIC_TRACKER}", json=payload, headers=_build_headers(), timeout=60, ) logger = logging.getLogger("conjurer_musician") logger.info("SENT") logger.info(response.status_code) logger.info("SEND CONFIRMED") def rescan(): """ The `rescan` function logs a message, scans for mp3 files in a specified folder, and adds them to a list of music files. """ logger = logging.getLogger("conjurer_musician") logger.info("Rescan triggered") music_file_list.clear() priority_list.clear() for mp3_item in MUSIC_FOLDER.glob("**/*.mp3"): temp_music_file = mp3_item.as_posix() if os.name == "nt": temp_music_file = temp_music_file.replace("/", "\\") music_file_list.append(temp_music_file) for mp3_item in PRIORITY_FOLDER.glob("**/*.mp3"): temp_music_file = mp3_item.as_posix() if os.name == "nt": temp_music_file = temp_music_file.replace("/", "\\") priority_list.append(temp_music_file) with ALL_PLAYLIST_PATH.open("w", encoding=ENCODING) as w_file: try: for item in music_file_list: w_file.write(item) w_file.write("\n") except json.JSONDecodeError: pass with HIT_PLAYLIST_PATH.open("w", encoding=ENCODING) as w_file: try: for item in priority_list: w_file.write(item) w_file.write("\n") except json.JSONDecodeError: pass def thread_rescan(): """ The `thread_rescan` function periodically triggers a rescan operation after a specified time interval. """ logger = logging.getLogger("conjurer_musician") logger.info("Starting filesystemupdater") while True: time.sleep(60 * 60 * 24) logger.info("Rescan triggered") rescan() def scan_tracks(): # Set the filename and open the file logger = logging.getLogger("conjurer_musician") 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) app = Flask(__name__) # AutoIndex(app, browse_root="/") # TODO: Odpalić wyszukiwarki w wątkach i dopiero po wszystkim zsumować wyszukiwanie. def wyszukaj(word_list, how_many, _logger=None, return_to_bot=True): """ Searches for music files based on a list of keywords and returns a list of matching files. Args: word_list (list): A list of keywords to search for. how_many (int): The maximum number of matching files to return. Returns: list: A list of matching files, sorted by their search weight. """ if _logger is None: fun_logger = logging.getLogger() else: fun_logger = _logger return_list = [] search_weight = [] for _ in range(len(music_file_list)): search_weight.append((0, "")) time_start = datetime.now() if int(how_many) > 0: skip_start = 2 else: skip_start = 1 for word in word_list[skip_start:]: token_weight = len(word) fun_logger.info("Słowo kluczowe: %s", word) itr = 0 for file in music_file_list: file = file.split(SEPARATOR_FILE_PATH) char_remove = [ ".", "^", "$", "*", "+", "?", "{", "}", "[", "]", "\\", "/", "|", "(", ")", "!", ",", "-", ":", "mp3", ] all_words = [] for f_iter in file: for char in char_remove: f_iter = remove_characters(f_iter, char) tmp = f_iter.split() all_words.extend(tmp) pingu = 1 pattern_len = len(all_words) if platform == "win32": skip = 2 else: skip = 4 matched_times = 1 # TODO: To do wątków albo for asynchroniczny for itm in all_words[skip:]: pingu += 1 pattern = ".*" + word + ".*" if re.match(pattern, 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 itr += 1 fun_logger.info( "Stworzylem tablice wag zajęło mi to %s", datetime.now() - time_start ) time_start = datetime.now() item_to_search = max_weight(search_weight) itr = 0 if item_to_search == 0: return [] not_found = True return_list = [] if int(how_many) <= 0: fun_logger.info("Jeden plik do zagrania") while not_found: if search_weight[itr][0] == item_to_search: return_list.append(search_weight[itr]) if not return_to_bot: with PRIORITY_PLAYLIST_PATH.open("a", encoding=ENCODING) as s_file: s_file.write(search_weight[itr][1] + "\n") break itr += 1 else: fun_logger.info("Wiele plików do zagrania") search_weight.sort(key=lambda x: x[0], reverse=True) return_list.extend(search_weight[: int(how_many)]) fun_logger.info("Done: %s", return_list) return return_list def max_weight(lista): """ Take a list of weights and return the maximum weight. :param lista: It seems like the parameter `lista` is a list of items, possibly representing weights. The function name `max_weight` suggests that the function is intended to find the maximum weight from the list. However, without more context or information about the problem, it's difficult to say for sure what the """ maximum_weight = 0 for iterator in lista: if iterator[0] > maximum_weight: maximum_weight = iterator[0] return maximum_weight def remove_characters(string, character): """ The `remove_characters` function removes all occurrences of a specified character from a given string. :param string: The string parameter is the input string from which characters will be removed :param character: The character parameter is the character that you want to remove from the string :return: a new string where all occurrences of the specified character have been removed. """ return string.replace(character, "") @app.route('/get_share_list', methods=['POST']) def get_share_list(): _authorize_request() data = request.get_json() entries = data.get('entries') keywords = data.get('keywords') # Validate entries if not isinstance(entries, int) or not (1 <= entries <= 10): return jsonify({'error': '"entries" must be an integer between 1 and 10.'}), 400 # Validate keywords list if not isinstance(keywords, list) or not all(isinstance(k, str) for k in keywords): return jsonify({'error': '"keywords" must be a list of strings.'}), 400 # Call external search files = media_search_functions.find_matches(entries, keywords) # Return each file path as a JSON list return jsonify({'files': files}), 200 @app.route('/get_share_links', methods=['POST']) def get_share_links(): _authorize_request() data = request.get_json() file_paths = data.get('file_paths') # Validate file_paths list if not isinstance(file_paths, list) or not all(isinstance(p, str) for p in file_paths): return jsonify({'error': '"file_paths" must be a list of strings.'}), 400 # Call external publish links = media_search_functions.publish(file_paths) # Return list of published links return jsonify({'links': links}), 200 @app.route("/stream", methods=["GET"]) def stream_music(): """ The function `stream_music` is a route handler for the "/stream" endpoint. It returns a JSON response containing a key "music_file_list" with the value of the variable `music_file_list`. :return: A JSON response containing a key "music_file_list" with the value of the variable `music_file_list`. """ # return send_from_directory("/tmp/hls", "stream.m3u8") return render_template(str(STREAM_TEMPLATE)) @app.route("/") def stream(file_name): """ Stream the specified file from the video directory. Args: file_name (str): The name of the file to be streamed. Returns: Response: The response containing the streamed file. """ # trunk-ignore(bandit/B108) video_dir = "/tmp/hls" return send_from_directory(video_dir, file_name) @app.route("/stream_mp3", methods=["GET"]) def stream_music_mp3(): """ The function `stream_music_mp3` is a route handler for the "/stream_mp3" endpoint. It redirects the user to the URL "http://www.example.com" with a status code of 302. :return: A redirect response to "http://www.example.com" with a status code of 302. """ return redirect("http://www.example.com", code=302) @app.route("/clear_pr_pls", methods=["GET"]) def clear_pr_pls(): _authorize_request() """ The function `clear_pr_pls` clears the contents of the priority queue playlist file. :return: A JSON response indicating the success of the operation. """ app.logger.info("CLEARING PLAYLIST") with PRIORITY_PLAYLIST_PATH.open("w", encoding=ENCODING) as cleared_pl: cleared_pl.write("") return_data = jsonify(isError=False, message="Success", statusCode=200, data=[]) return return_data, 200 @app.route("/mp3", methods=["GET"]) def get_music_list(): """ The function `get_music_list` returns a JSON object containing a list of music files. :return: A JSON response containing a key "music_file_list" with the value of the variable `music_file_list`. """ return jsonify({"music_file_list": music_file_list}) @app.route("/update_mp3", methods=["POST"]) def update_music_list(): """ The function `update_music_list` receives a POST request with a JSON payload, logs the received item, adds it to a music file list, and returns a success message along with the updated record. :return: The function `update_music_list` is returning a tuple containing a JSON response and a status code. The JSON response includes keys `isError`, `message`, `statusCode`, and `data`, with values indicating the success of the operation and the data that was received and added to the `music_file_list`. The status code returned is 200, indicating a successful response. """ _authorize_request() record = json.loads(request.data) app.logger.info(record["item"]) music_file_list.append(record["item"]) return_data = ( jsonify(isError=False, message="Success", statusCode=200, data=record), 200, ) return return_data @app.route("/get_music", methods=["POST"]) def look_for_playlist(): """ The function `look_for_playlist` receives a POST request with a JSON payload, logs the received item, adds it to a music file list, and returns a success message along with the updated record. :return: A tuple containing a JSON response and a status code. The JSON response includes keys `isError`, `message`, `statusCode`, and `data`, with values indicating the success of the operation and the data that was received and added to the `music_file_list`. The status code returned is 200, indicating a successful response. """ _authorize_request() record = json.loads(request.data) app.logger.info(record) app.logger.info(record["lista_slow"]) app.logger.info(record["UUID"]) app.logger.info(record["dlugosc_plejlisty"]) return_data = wyszukaj( record["lista_slow"], record["dlugosc_plejlisty"], app.logger, True ) return_data = ( jsonify(isError=False, message="Success", statusCode=200, data=return_data), 200, ) return return_data @app.route("/request_radio_file", methods=["POST"]) def add_request(): _authorize_request() record = json.loads(request.data) app.logger.info(record) app.logger.info(record["lista_slow"]) app.logger.info(record["UUID"]) return_data = wyszukaj(record["lista_slow"], 0, app.logger, False) with REQUEST_PLAYLIST_PATH.open("a", encoding=ENCODING) as s_file: for item in return_data: s_file.write(item[1] + "\n") return_data = ( jsonify( isError=False, message="Success", statusCode=200, data={"status": "OK"} ), 200, ) return return_data @app.route("/create_priority_playlist", methods=["POST"]) def create_priority_playlist(): """ The function `create_priority_playlist` receives a POST request with a JSON payload, logs the received item, adds it to a music file list, and returns a success message along with the updated record. :return: A tuple containing a JSON response and a status code. The JSON response includes keys `isError`, `message`, `statusCode`, and `data`, with values indicating the success of the operation and the data that was received and added to the `music_file_list`. The status code returned is 200,indicating a successful response. """ _authorize_request() record = json.loads(request.data) app.logger.info(record) app.logger.info(record["lista_slow"]) app.logger.info(record["UUID"]) app.logger.info(record["dlugosc_plejlisty"]) return_data = wyszukaj( record["lista_slow"], record["dlugosc_plejlisty"], app.logger, False ) random.shuffle(return_data) with REQUEST_PLAYLIST_PATH.open("a", encoding=ENCODING) as s_file: for item in return_data: s_file.write(item[1] + "\n") return_data = ( jsonify( isError=False, message="Success", statusCode=200, data={"status": "OK"} ), 200, ) return return_data @app.route("/add_to_priority", methods=["POST"]) def add_to_priority(): """ The function `add_to_priority` receives a POST request with a JSON payload, logs the received item, adds it to a music file list, and returns a success message along with the updated record. :return: A tuple containing a JSON response and a status code. The JSON response includes keys `isError`, `message`, `statusCode`, and `data`, with values indicating the success of the operation and the data that was received and added to the `music_file_list`. The status code returned is 200,indicating a successful response. """ _authorize_request() record = json.loads(request.data) app.logger.info(record) app.logger.info(record["lista_slow"]) app.logger.info(record["UUID"]) app.logger.info(record["dlugosc_plejlisty"]) return_data = wyszukaj( record["lista_slow"], record["dlugosc_plejlisty"], app.logger, False ) with PRIORITY_PLAYLIST_PATH.open("a", encoding=ENCODING) as s_file: for item in return_data: s_file.write(item[1] + "\n") return_data = ( jsonify( isError=False, message="Success", statusCode=200, data={"status": "OK"} ), 200, ) return return_data def flask_debug(): """ The `flask_debug` function starts a Flask application in debug mode without using the reloader. Do not use for production for fucks sake. """ # trunk-ignore(bandit/B201) app.run(debug=True, use_reloader=False, host=HOST_ADDRESS, port=PORT_ADDRESS) def waitress_run(): """ The `waitress_run` function serves the `app` on host HOST_ADDRESS and port 5000 using the Waitress WSGI server. This function starts the Waitress server and listens for incoming requests on the specified host and port. It uses the `serve` function from the Waitress library to handle the serving process. Parameters: None Returns: None """ serve(app, host=HOST_ADDRESS, port=PORT_ADDRESS) if __name__ == "__main__": logger = logging.getLogger("conjurer_musician") logger.setLevel(logging.DEBUG) formatter = logging.Formatter( "%(asctime)s - %(name)s - %(levelname)s - %(message)s" ) handler = handlers.RotatingFileHandler( filename=LOGFILE, encoding=ENCODING, mode="a", maxBytes=6 * 1024 * 1024, backupCount=6, ) handler.setFormatter(formatter) logger.addHandler(handler) rescan() logger.info("Started") threads = [] # threads.append(threading.Thread(target=flask_debug)) threads.append(threading.Thread(target=waitress_run, daemon=True)) threads.append(threading.Thread(target=thread_rescan, daemon=True)) for worker in threads: worker.start() time.sleep(60) 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 musician service")