Create playlist from radio stream by querying stream titles.

write_playlist_for_radio_stream.py

#!/usr/bin/python3

# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
#
# Copyright (C) 2023 Henning Hollermann
#
# The script was mainly inspired by 
# https://stackoverflow.com/questions/35102278/python-3-get-song-name-from-internet-radio-stream
#
# usage: write_playlist_for_radio_stream.py [-h] [--console] [--debug]
#                                           [--log_file LOG_FILE]
#                                           [--sleep_time SLEEP_TIME]
#                                           [--url URL] [--exclude EXCLUDE]
#
# Create playlist from radio stream by querying stream titles.
#
# optional arguments:
#   -h, --help            show this help message and exit
#   --console             Log playlist also to the console
#   --debug               Verbose logging to the console.
#   --log_file LOG_FILE   File to log the stream titles (song names).
#   --sleep_time SLEEP_TIME
#                         Sleep time between stream queries (Default: 22).
#   --url URL             Radio stream URL (Default: WDR Maus).
#   --exclude EXCLUDE     Substring to exclude from playlist: If stream title
#                         contains this string it won't appear in the playlist
#                         log.

import argparse
import re
import struct
import sys
import urllib.request as urllib2
from datetime import datetime
from time import sleep
from locale import setlocale, LC_TIME

# set locale for times (deutsches Zeitformat)
setlocale(LC_TIME, "de_DE.UTF-8")

def log_stream_title_update(stream_title_str, old_stream_title_str):
    """
    If stream title was updated, check first, if we really want to log this new title.
    :return: True if new title should be logged (not skipped)
    """
    # do not log if the new title equals the old one
    if stream_title_str == old_stream_title_str:
        return False
    #
    for sub_str in args.exclude:
        if sub_str in stream_title_str:
            return False
    # Artist mit Song == Song von Artist
    if " mit " in stream_title_str and not " von " in stream_title_str:
        l = stream_title_str.split(" mit ")
        if " von ".join([l[1], l[0]]) == old_stream_title_str:
            return False
    elif " von " in stream_title_str and not " mit " in stream_title_str:
        l = stream_title_str.split(" von ")
        if " mit ".join([l[1], l[0]]) == old_stream_title_str:
            return False

    return True


# SCRIPT START
parser = argparse.ArgumentParser(description='Create Playlist from radio stream by querying stream titles.')
parser.add_argument('--console', dest='console', default=False, action='store_true', help='Log playlist also to the console')
parser.add_argument('--debug', dest='DEBUG', default=False, action='store_true', help='Verbose logging to the console.')
parser.add_argument('--log_file', dest='log_file', 
        default='/var/log/playlist-wdrmaus.txt', help='File to log the stream titles (song names).')
parser.add_argument('--sleep_time', dest='sleep_time', type=int, default=22,
        help='Sleep time between stream queries (Default: 22).')
parser.add_argument('--url', dest='url',
        default='https://wdr-diemaus-live.icecastssl.wdr.de/wdr/diemaus/live/mp3/128/stream.mp3', 
        help='Radio stream URL (Default: WDR Maus).')
parser.add_argument('--exclude', action='append', type=str,
        help="Substring to exclude from playlist: If stream title contains this string it won't appear in the playlist log.")

args = parser.parse_args()

old_stream_title_str = None
old_loop_time = None

log = {}
# main loop
while True:
    # set the firstrun variable
    firstrun = True if old_stream_title_str is None else False
    # start loop with sleeping so that a 'continue' statement sends the script to sleep
    if firstrun is False:
        sleep(args.sleep_time)
    # set up the request for getting the stream metadata
    request = urllib2.Request(args.url, headers={'Icy-MetaData': 1})  # request metadata
    # send an http request to the stream server
    response = urllib2.urlopen(request)

    if args.DEBUG:
        print(response.headers, file=sys.stderr)
    metaint = int(response.headers['icy-metaint'])
    # skip in response to metadata part and extract it
    response.read(metaint)
    metadata_length = struct.unpack('B', response.read(1))[0]*16
    metadata = response.read(metadata_length).rstrip(b'\0')
    if args.DEBUG:
        print(metadata, file=sys.stderr)
    # extract stream title from the metadata
    m = re.search(br"StreamTitle='([^']*)';", metadata)
    if m:
        stream_title = m.group(1)
        stream_title_str = stream_title.decode()
        loop_time = datetime.now()

        # add date-key to log if it does not exist yet
        if loop_time.date() not in log.keys():
            log[loop_time.date()] = []

        if firstrun or log_stream_title_update(stream_title_str, old_stream_title_str):
            log[loop_time.date()].append((loop_time, stream_title_str))
            log_line = "{0!s}; {1!s}".format(loop_time.strftime('%A, %x %R'),
                                             stream_title_str)
            if args.DEBUG or args.console:
                print(log_line)
            with open(args.log_file, 'a') as log_file:
                log_file.write("{0!s}\n".format(log_line))

            old_loop_time = loop_time
            old_stream_title_str = stream_title_str