This commit is contained in:
2024-04-22 20:11:50 +02:00
committed by migatu
parent 8f006b4323
commit e44bc7f64a
4 changed files with 40 additions and 29 deletions
+5 -18
View File
@@ -44,7 +44,7 @@ import yt_dlp
from spotify_dl import spotify from spotify_dl import spotify
from spotify_dl import youtube as youtube_download from spotify_dl import youtube as youtube_download
from communication_subroutine import comm_subroutine, IN_COMM_Q, OUT_COMM_Q from communication_subroutine import comm_subroutine, IN_COMM_Q, OUT_COMM_Q, QueryControl
# *=========================== Classes definition # *=========================== Classes definition
@@ -62,22 +62,6 @@ Music_Config = TypedDict(
) )
class QueryControl:
"""
This class `QueryControl` is used to manage queries with information about the author, UUID,
content, logger, context, and replies.
"""
def __init__(self, query_author, query_uuid, query_content, uplogger, ctx) -> None:
self.author = query_author
self.uuid = query_uuid
self.content = query_content
self.logger = uplogger
self.stop = False
self.ctx = ctx
self.logger.info(
f"Created Query control for {self.author}, {self.uuid}: {self.content}"
)
self.replies = []
class MusicFileList(object): class MusicFileList(object):
""" """
@@ -695,7 +679,10 @@ async def check_data_q():
entries = [] entries = []
if fresh_data.stop: if fresh_data.stop:
searcher = fresh_data.author searcher = fresh_data.author
ctx = fresh_data.ctx if fresh_data.ctx is not None:
ctx = fresh_data.ctx
else:
ctx = client.get_channel(1062047367337095268)
query = fresh_data.content query = fresh_data.content
l_p = 1 l_p = 1
for doi in fresh_data.entries: for doi in fresh_data.entries:
+23
View File
@@ -13,6 +13,22 @@ IN_COMM_Q = Queue()
awaiting_q = [] awaiting_q = []
incoming_q = Queue() incoming_q = Queue()
app = Flask(__name__) app = Flask(__name__)
class QueryControl:
"""
This class `QueryControl` is used to manage queries with information about the author, UUID,
content, logger, context, and replies.
"""
def __init__(self, query_author, query_uuid, query_content, uplogger, ctx) -> None:
self.author = query_author
self.uuid = query_uuid
self.content = query_content
self.logger = uplogger
self.stop = False
self.ctx = ctx
self.logger.info(
f"Created Query control for {self.author}, {self.uuid}: {self.content}"
)
self.replies = []
@app.route("/conjurer", methods=["POST"]) @app.route("/conjurer", methods=["POST"])
def answer_external_command(): def answer_external_command():
@@ -91,6 +107,13 @@ def scan_incoming(_logger):
record.stop = True record.stop = True
record.entries = answer[record.uuid] record.entries = answer[record.uuid]
IN_COMM_Q.put(record) IN_COMM_Q.put(record)
else:
for key in answer.keys():
record = QueryControl("Orphaned", key, "Orphan", _logger, None)
record.stop = True
record.entries = answer[record.uuid]
record.uuid = "000000"
IN_COMM_Q.put(record)
except Empty: except Empty:
time.sleep(1) time.sleep(1)
+10 -9
View File
@@ -68,7 +68,7 @@ def producer(out_q, control_q, filename, _logger):
# IMPORTANT!!! ONLY ONE CONSUMER THREAD AS WE ARE NOT PUTTING SENTINELS BACK # IMPORTANT!!! ONLY ONE CONSUMER THREAD AS WE ARE NOT PUTTING SENTINELS BACK
def consumer(in_q, control_q, doi, live_results, result_list, _logger, ): def consumer(in_q, control_q, doi, live_results, result_list, control_dict, _logger):
""" """
Consumes items from an input queue and checks if DOI exists in the live results list. Consumes items from an input queue and checks if DOI exists in the live results list.
@@ -79,7 +79,6 @@ def consumer(in_q, control_q, doi, live_results, result_list, _logger, ):
live_results (list): List to store the search results. live_results (list): List to store the search results.
_logger: Logger object for logging. _logger: Logger object for logging.
""" """
sentinels = 0
for item in doi: for item in doi:
result_list.append({"DOI": item[0], "exists": False, "data": item[1]}) result_list.append({"DOI": item[0], "exists": False, "data": item[1]})
while True: while True:
@@ -87,7 +86,7 @@ def consumer(in_q, control_q, doi, live_results, result_list, _logger, ):
data = in_q.get() data = in_q.get()
if data is _sentinel: if data is _sentinel:
_logger.info("Worker finished") _logger.info("Worker finished")
sentinels += 1 control_dict["sentinels"] += 1
else: else:
for item in result_list: for item in result_list:
if item["DOI"] in data and not item["exists"]: if item["DOI"] in data and not item["exists"]:
@@ -98,7 +97,7 @@ def consumer(in_q, control_q, doi, live_results, result_list, _logger, ):
done_check = done_check and item["exists"] done_check = done_check and item["exists"]
if done_check: if done_check:
control_q.put(_sentinel) control_q.put(_sentinel)
if sentinels >= MAXTHREADS: if control_dict["sentinels"] >= MAXTHREADS:
_logger.info("All workers finished") _logger.info("All workers finished")
break break
@@ -112,15 +111,17 @@ def search_for_doi(doi, live_results, _logger):
live_results (list): List to store the search results. live_results (list): List to store the search results.
_logger: Logger object for logging. _logger: Logger object for logging.
""" """
control_dict = {"sentinels":0}
result_list = [] result_list = []
threads = [] threads = []
work_q = Queue(maxsize=WORK_Q_SIZE) work_q = Queue(maxsize=WORK_Q_SIZE)
control_q = Queue() control_q = Queue()
t_cons = Thread( for _ in range (1, len(doi)//1000+1):
target=consumer, args=(work_q, control_q, doi, live_results, result_list, _logger) t_cons = Thread(
) target=consumer, args=(work_q, control_q, doi, live_results, result_list, control_dict, _logger)
_logger.info("Consumer thread created") )
threads.append(t_cons) _logger.info("Consumer thread created")
threads.append(t_cons)
for i in range(0, MAXTHREADS): for i in range(0, MAXTHREADS):
# TEST DATA # TEST DATA
# filename = "test" + str(i) + CHUNk # filename = "test" + str(i) + CHUNk
+2 -2
View File
@@ -24,8 +24,8 @@ let json.parse credentials = file.contents("/home/pi/Conjurer/icecast_credentia
s3 = replaygain(playlist(reload_mode="watch", check_next=check, "/home/pi/Conjurer/hit.playlist")) s3 = replaygain(playlist(reload_mode="watch", check_next=check, "/home/pi/Conjurer/hit.playlist"))
requests_queue = request.queue() requests_queue = request.queue()
def queue_processing() def queue_processing()
requests_queue.push.uri("/home/pi/Conjurer/request.playlist"), requests_queue.push.uri("/home/pi/Conjurer/request.playlist")
file.remove("/home/pi/Conjurer/requests_queue.playlist"), file.remove("/home/pi/Conjurer/requests_queue.playlist")
file.touch("/home/pi/Conjurer/requests_queue.playlist") file.touch("/home/pi/Conjurer/requests_queue.playlist")
end end
thread.run(every=60., {queue_processing()}) thread.run(every=60., {queue_processing()})