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 = os.getenv("CONJURER_DISCORD_HOST", "192.168.1.31")
PORT_ADDRESS = int(os.getenv("CONJURER_DISCORD_PORT", "5000"))
ICECAST_ADDRESS = os.getenv("CONJURER_ICECAST", "http://192.168.1.15:8000")
API_KEY = os.getenv("CONJURER_API_KEY")
OUT_COMM_Q = Queue()
IN_COMM_Q = Queue()
SRCHTITLE = re.compile(rb"StreamTitle=\\*(?P<title>[^;]*);").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.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()