116 lines
4.2 KiB
Python
Executable file
116 lines
4.2 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
# vim: tabstop=8 expandtab shiftwidth=4 softtabstop=4
|
|
|
|
"""
|
|
This script records internet radio streams. It can be used in conjunction
|
|
with "at" or "crontab".
|
|
"""
|
|
|
|
import argparse
|
|
import configparser
|
|
import datetime
|
|
import os
|
|
import sys
|
|
import threading
|
|
import urllib.request
|
|
|
|
def check_duration(value):
|
|
try:
|
|
value = int(value)
|
|
except ValueError:
|
|
raise argparse.ArgumentTypeError('Duration must be a positive integer.')
|
|
|
|
if value < 1:
|
|
raise argparse.ArgumentTypeError('Duration must be a positive integer.')
|
|
else:
|
|
return value
|
|
|
|
def read_settings():
|
|
settings_base_dir = ''
|
|
if sys.platform == 'linux':
|
|
settings_base_dir = os.getenv('HOME') + os.sep + '.config' + os.sep + 'radiorec'
|
|
elif sys.platform == 'win32':
|
|
settings_base_dir = os.getenv('APPDATA') + os.sep + 'radiorec'
|
|
settings_base_dir += os.sep
|
|
config = configparser.ConfigParser()
|
|
config.read(settings_base_dir + 'settings.ini')
|
|
return dict(config.items())
|
|
|
|
def record_worker(stoprec, streamurl, target_dir, name=None):
|
|
conn = urllib.request.urlopen(streamurl)
|
|
filename = target_dir + os.sep + datetime.datetime.now().isoformat()
|
|
if name:
|
|
filename += '_' + name
|
|
content_type = conn.getheader('Content-Type')
|
|
if(content_type == 'audio/mpeg'):
|
|
filename += '.mp3'
|
|
elif(content_type == 'application/ogg' or content_type == 'audio/ogg'):
|
|
filename += '.ogg'
|
|
elif(content_type == 'audio/x-mpegurl'):
|
|
print('Sorry, M3U playlists are currently not supported')
|
|
sys.exit()
|
|
else:
|
|
print('Unknown content type "' + content_type + '". Assuming mp3.')
|
|
filename += 'mp3'
|
|
with open(filename, "wb") as target:
|
|
while(not stoprec.is_set() and not conn.closed):
|
|
target.write(conn.read(1024))
|
|
|
|
def record(args):
|
|
settings = read_settings()
|
|
streamurl = ''
|
|
global verboseprint
|
|
verboseprint = print if args.verbose else lambda *a, **k: None
|
|
|
|
try:
|
|
streamurl = settings['STATIONS'][args.station]
|
|
except KeyError:
|
|
print('Unkown station name: ' + args.station)
|
|
sys.exit()
|
|
if streamurl.endswith('.m3u'):
|
|
verboseprint('Seems to be an M3U playlist. Trying to parse…')
|
|
with urllib.request.urlopen(streamurl) as remotefile:
|
|
for line in remotefile:
|
|
if not line.decode('utf-8').startswith('#'):
|
|
tmpstr = line.decode('utf-8')
|
|
break
|
|
streamurl = tmpstr
|
|
verboseprint('stream url: ' + streamurl)
|
|
target_dir = os.path.expandvars(settings['GLOBAL']['target_dir'])
|
|
stoprec = threading.Event()
|
|
|
|
verboseprint('Recording ' + args.station + '…')
|
|
recthread = threading.Thread(target = record_worker,
|
|
args = (stoprec, streamurl, target_dir, args.name), daemon = True)
|
|
recthread.start()
|
|
recthread.join(args.duration * 60)
|
|
|
|
if(recthread.is_alive):
|
|
stoprec.set()
|
|
|
|
def list(args):
|
|
settings = read_settings()
|
|
for key in settings['STATIONS']:
|
|
print(key)
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(prog='radiorec', description='This program records internet radio streams')
|
|
subparsers = parser.add_subparsers(help='sub-command help')
|
|
parser_record = subparsers.add_parser('record', help='Record a station')
|
|
parser_record.add_argument('station', type=str, help='Name of the radio station '
|
|
'(see config file for a list)')
|
|
parser_record.add_argument('duration', type=check_duration,
|
|
help='Recording time in minutes')
|
|
parser_record.add_argument('name', nargs='?', type=str,
|
|
help='A name for the recording')
|
|
parser_record.add_argument('-v', '--verbose', action='store_true', help="Verbose output")
|
|
parser_record.set_defaults(func=record)
|
|
parser_list = subparsers.add_parser('list', help='List all known stations')
|
|
parser_list.set_defaults(func=list)
|
|
#parser_list.add_argument('-l', '--list', action='store_true',
|
|
# help='Get a list of all known radio stations')
|
|
args = parser.parse_args()
|
|
args.func(args)
|
|
|
|
if __name__ == '__main__':
|
|
main()
|