conjurer/communication_subroutine.py

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 = "192.168.1.191"
PORT_ADDRESS = 5000
ICECAST_ADDRESS = "http://192.168.1.15:8000"
OUT_COMM_Q = Queue()
IN_COMM_Q = Queue()
SRCHTITLE = re.compile(rb"StreamTitle=\\*(?P<title>[^;]*);").search

awaiting_q = []
incoming_q = Queue()
app = Flask(__name__)
PREPPED_TRACKS = {
    "requests": "",
    "hit": "",
    "all": "",
    "priority": "",
    "jingles": "",
    "now_playing": "",
    "next": "",
    "meta": "",
}
logger = logging.getLogger("discord")

API_KEY = os.getenv("CONJURER_API_KEY")


def _authorize_request() -> None:
    if API_KEY and request.headers.get("X-Conjurer-Api-Key") != API_KEY:
        abort(401)


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():
    _authorize_request()
    """
    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".
    """
    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.

    :param _logger: The `_logger` parameter is typically an instance of a logging object that is used to
    record and store log messages. It is commonly used to track the flow of the program, record errors,
    and provide information for debugging purposes. In this code snippet, the `_logger` object is used
    to log the
    """
    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 _logger: The `_logger` parameter in the `scan_incoming` function is a logger object that is
    used to log messages or information during the execution of the function. It is typically used for
    debugging, monitoring, or tracking the flow of the program. In this case, the `_logger` is being
    used
    """
    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.

    :param logger: The `logger` parameter in the `comm_subroutine` function is an instance of a logger
    object that is used for logging messages at various levels (e.g., debug, info, warning, error). In
    the provided code snippet, the logger is used to log messages at the "info" level
    """
    # logger.setLevel(logging.DEBUG)
    logger = logging.getLogger("discord")
    logger.info("Started comms")
    threads = []
    # 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()