import json import logging import os import re import threading import time from queue import Empty, Queue from typing import Optional from urllib import request as urequest from flask import Flask, abort, jsonify, request from waitress import serve HOST_ADDRESS = os.getenv("CONJURER_DISCORD_HOST", "192.168.1.92") PORT_ADDRESS = int(os.getenv("CONJURER_DISCORD_PORT", "5000")) ICECAST_ADDRESS = os.getenv("CONJURER_ICECAST", "http://192.168.1.12:8000") API_KEY = os.getenv("CONJURER_API_KEY") OUT_COMM_Q = Queue() IN_COMM_Q = Queue() SRCHTITLE = re.compile(rb"StreamTitle=\\*(?P[^;]*);").search awaiting_q = [] incoming_q = Queue() app = Flask(__name__) def _authorize_request() -> None: """Reject inbound calls lacking the shared key (no-op if key is unset).""" if API_KEY and request.headers.get("X-Conjurer-Api-Key") != API_KEY: abort(401) PREPPED_TRACKS = { "requests": "", "hit": "", "all": "", "priority": "", "jingles": "", "now_playing": "", "next": "", "meta": "", } logger = logging.getLogger("discord") 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, ctx) -> None: self.author = query_author self.uuid = query_uuid self.content = query_content self.logger = logging.getLogger("discord") self.stop = False self.ctx = ctx self.logger.info( f"Created Query control for {self.author}, {self.uuid}: {self.content}" ) self.replies = [] @app.route("/prepped_tracks", methods=["POST"]) def log_radio_tracks(): _authorize_request() app.logger = logging.getLogger("discord") app.logger.info(request) record = json.loads(request.data) app.logger.info(record) if "next" in record[0]: metadata = id3(ICECAST_ADDRESS) PREPPED_TRACKS["now_playing"] = PREPPED_TRACKS["next"] if metadata: PREPPED_TRACKS["meta"] = ( metadata["name"] + " - " + metadata["title"] + "(" + metadata["genre"] + ")" ) else: PREPPED_TRACKS["meta"] = "Nie znaju" PREPPED_TRACKS[record[0]] = record[1] app.logger.info("DATA RECEIVED") return jsonify("SUCCESS") @app.route("/conjurer", methods=["POST"]) def answer_external_command(): """ The function `answer_external_command` logs the request data, loads the data as JSON, logs the record, and then puts the record into an incoming queue before returning a success message. :return: The function `answer_external_command()` is returning a JSON response with the message "SUCCESS". """ _authorize_request() logger = logging.getLogger("discord") logger.info(request) record = json.loads(request.data) logger.info(record) logger.info("DATA RECEIVED") incoming_q.put(record) return jsonify("SUCCESS") @app.route("/conjurer", methods=["GET"]) def check_alive(): """ The function "check_alive" returns a JSON response with the message "ALIVE" when the "/conjurer" route is accessed via a GET request. :return: The code snippet is a Flask route that listens for GET requests to the "/conjurer" endpoint. When a GET request is made to this endpoint, the function check_alive() is called, which returns a JSON response with the message "ALIVE". """ return jsonify("ALIVE") 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. """ logger = logging.getLogger("discord") logger.info("Attempt debug") # 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. """ logger = logging.getLogger("discord") logger.info("Attempt waitress") serve(app, host=HOST_ADDRESS, port=PORT_ADDRESS) def scan_queue(stop_event: Optional[threading.Event] = None): """ The function `scan_queue` reads data from a queue, logs it, and appends it to another queue. A bounded ``get(timeout=1)`` is used instead of a blocking ``get()`` so the worker can observe ``stop_event`` and exit cleanly during shutdown. :param stop_event: optional :class:`threading.Event`; when set the loop stops at the next iteration. """ logger = logging.getLogger("discord") while True: if stop_event and stop_event.is_set(): logger.info("scan_queue: stop requested") break try: data = OUT_COMM_Q.get(timeout=1) except Empty: continue logger.info(data) awaiting_q.append(data) def scan_incoming(stop_event: Optional[threading.Event] = None): """ The `scan_incoming` function continuously checks for incoming data, processes it, and logs when data is found. :param stop_event: optional :class:`threading.Event`; when set the loop stops at the next iteration. """ logger = logging.getLogger("discord") while True: if stop_event and stop_event.is_set(): logger.info("scan_incoming: stop requested") break try: answer = incoming_q.get(block=False) logger.info("DATA FOUND") record_stored = False for record in awaiting_q: if record.uuid in answer.keys(): record_stored = True record.stop = True record.entries = answer[record.uuid] IN_COMM_Q.put(record) if not record_stored: for key in answer.keys(): record = QueryControl("Orphaned", key, "Orphan", None) record.stop = True record.entries = answer[record.uuid] IN_COMM_Q.put(record) except Empty: time.sleep(1) def get_stream_title(tag: bytes) -> str: title = "" if m := SRCHTITLE(tag): # decode, strip, unescape and remove surrounding quotes (may not even be the same type of quote) title = m.group("title").decode("utf-8").strip().replace("\\", "")[1:-1] return title def id3(url: str) -> dict: request = urequest.Request(url, headers={"Icy-MetaData": 1}) with urequest.urlopen(request) as resp: metaint = int(resp.headers.get("icy-metaint", "-1")) if metaint < 0: return False resp.read( metaint ) # this isn't seekable so, arbitrarily read to the point we want tagdata = dict( site_url=resp.headers.get("icy-url"), name=resp.headers.get("icy-name").title(), genre=resp.headers.get("icy-genre").title(), title=get_stream_title(resp.read(255)), ) return tagdata def comm_subroutine(stop_event: Optional[threading.Event] = None): """ The `comm_subroutine` function starts multiple threads to run different tasks concurrently. Workers run as daemon threads and honour an optional ``stop_event`` so the bot can shut the communication layer down cleanly instead of blocking forever on ``join()``. :param stop_event: optional :class:`threading.Event` shared with the caller to coordinate a cooperative shutdown. """ logger = logging.getLogger("discord") logger.setLevel(logging.DEBUG) logger.info("Started comms") threads = [] # NOTE: flask_debug is the dev server bound to the SAME host:port as # waitress - running both kills the comm layer with 'address in use'. # Enable it only INSTEAD of waitress_run, never alongside. # threads.append(threading.Thread(target=flask_debug)) threads.append(threading.Thread(target=waitress_run, daemon=True)) threads.append( threading.Thread( target=scan_queue, kwargs={"stop_event": stop_event}, daemon=True ) ) threads.append( threading.Thread( target=scan_incoming, kwargs={"stop_event": stop_event}, daemon=True ) ) for worker in threads: worker.start() try: while any(thread.is_alive() for thread in threads): if stop_event and stop_event.is_set(): break time.sleep(0.5) finally: if stop_event: stop_event.set() if __name__ == "__main__": comm_subroutine()