icy_titler.py

import re
import time
from io import BytesIO
from threading import Thread

import requests
from sanic import Sanic, response
from sanic.request import Request

global_stream_dict = {}


def _icy_monitor(url):
    try:
        r = requests.get(url, headers={'Icy-MetaData': '1'}, stream=True)
    except ConnectionError:
        return
    r.encoding = r.encoding or 'utf-8'

    byte_counter = 0
    meta_counter = 0
    metadata_buffer = BytesIO()

    metadata_size = int(r.headers['icy-metaint']) + 255
    data_is_meta = False

    for byte in r.iter_content(1):
        byte_counter += 1

        if byte_counter > 2048:
            if meta_counter == 0:
                meta_counter += 1

            elif meta_counter <= int(metadata_size + 1):
                metadata_buffer.write(byte)
                meta_counter += 1
            else:
                data_is_meta = True

        if byte_counter > 2048 + metadata_size:
            byte_counter = 0

        if data_is_meta:
            metadata_buffer.seek(0)
            meta = metadata_buffer.read().rstrip(b'\0')

            if m := re.search(br"StreamTitle='([^']*)';", bytes(meta)):
                title = m.group(1).decode(r.encoding, errors='replace')
                if title != global_stream_dict.get(url):
                    print(f'{url} :: {title}')
                    global_stream_dict[url] = title

            byte_counter = 0
            meta_counter = 0
            metadata_buffer = BytesIO()
            data_is_meta = False


class IcyTitler:
    def __init__(self):
        self.app = Sanic(name="icytit")
        self.threads = []

        async def home(request: Request):
            get_args = request.get_args()
            if get_args.get("all"):
                return response.json(global_stream_dict)

            if url := get_args.get('url'):
                if not global_stream_dict.get(url):
                    t = Thread(target=_icy_monitor, args=(url,))
                    t.start()
                    self.threads += [t]
                    time.sleep(2)
                return response.json({'title':global_stream_dict.get(url)})

            return response.text("♫ icy tit is up and running")

        self.app.add_route(home, "/")

        # def stop_threads():
        # self.app.register_listener(stop_threads, "before_server_stop")


def main():
    icyTitler = IcyTitler()
    # if config.get("unix-socket"):
    #     import os
    #     import socket
    #
    #     with socket.socket(socket.AF_UNIX) as sock:
    #         sock.bind("icyTitler.socket")
    #         icyTitler.app.run(sock=sock)
    #     os.remove("icyTitler.socket")
    # else:
    icyTitler.app.run(host="0.0.0.0", port=9282)


if __name__ == "__main__":
    main()

dependencies:

sanic = "*"
requests = "*"