mirror of
https://github.com/migatu/conjurer.git
synced 2026-07-14 21:38:38 +00:00
278 lines
9.4 KiB
Python
278 lines
9.4 KiB
Python
"""
|
|
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))
|
|
threads.append(threading.Thread(target=flask_debug))
|
|
|
|
for worker in threads:
|
|
worker.start()
|
|
|
|
for worker in threads:
|
|
worker.join()
|
|
|
|
#TODO: Inne pliki - sadox, dokumenty naukowe itp.
|