""" 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 re from pathlib import Path from platform import uname import threading import time from datetime import datetime from sys import platform import logging from flask import Flask, jsonify, request from waitress import serve HOST_ADDRESS = "192.168.1.15" PORT_ADDRESS = 5000 if platform in ("linux", "linux2"): SEPARATOR_FILE_PATH = "/" if "microsoft-standard" in uname().release: LOGFILE = "/home/mtuszowski/conjurer/discord_mus_service.log" MUSIC_FOLDER = "/mnt/g/Muzyka/" NETRC_FILE = "/home/mtuszowski/.netrc" LOGSTORE = "/home/mtuszowski/conjurer/logs/" ENCODING = "utf-8" GRAPHICS_PATH = "/home/mtuszowski/conjurer/Conjurer_graphics/" else: LOGFILE = "/home/pi/Conjurer/discord_mus_service.log" MEMORY_FIVE_SIARA = "/home/pi/Conjurer/pamiec.json" SYSTEM_GPT_SETTINGS = "/home/pi/Conjurer/system_gpt_settings.json" MEMORY_FIVE_MUZYKA = "/home/pi/Conjurer/pamiec_muzyki.json" MUSIC_FOLDER = "/home/pi/RetroPie/mp3/" SETTINGS_FILE = "/home/pi/Conjurer/settings.json" NETRC_FILE = "/home/pi/.netrc" LOGSTORE = "/home/pi/RetroPie/logs/" ACCIDENT_LOG = "/home/pi/Conjurer/accident_log.json" ENCODING = "utf-8" GRAPHICS_PATH = "/home/pi/RetroPie/Conjurer_graphics/" DIR_PATH_SADOX = "/home/pi/RetroPie/Fansadox/" music_file_list = [] 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. """ logging.info("Rescan triggered") for mp3_item in Path.glob(Path(MUSIC_FOLDER), "**/*.mp3"): temp_music_file = mp3_item.as_posix() if platform == "win32": temp_music_file = temp_music_file.replace("/", "\\") music_file_list.append(temp_music_file) def thread_rescan(): """ The `thread_rescan` function periodically triggers a rescan operation after a specified time interval. """ logging.info("Starting filesystemupdater") while True: time.sleep(60*60*24) logging.info("Rescan triggered") rescan() app = Flask(__name__) def wyszukaj(word_list, how_many): logger = logging.getLogger() 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) 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 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 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 None not_found = True return_list = [] if int(how_many) <= 0: logger.info("Jeden plik do zagrania") while not_found: if search_weight[itr][0] == item_to_search: return_list.append(search_weight[itr]) return return_list itr += 1 else: 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)]) 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("/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. """ 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 `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. """ 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"]) return_data = ( jsonify(isError=False, message="Success", statusCode=200, data=return_data), 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 "0.0.0.0" and port 5000 using the Waitress WSGI server. """ serve(app, host=HOST_ADDRESS, port=PORT_ADDRESS) if __name__ == "__main__": logger = logging.getLogger('conjurer_musician') logger.setLevel(logging.DEBUG) rescan() logger.info("Started") threads = [] threads.append(threading.Thread(target=waitress_run)) threads.append(threading.Thread(target=thread_rescan)) for worker in threads: worker.start() for worker in threads: worker.join() #TODO: Inne pliki - sadox, dokumenty naukowe itp.