らじる★らじるの再生や録音を自動化

#! /usr/bin/luajit

-- radiru-get-streamurl.lua
-- This script requires "lua-sec" and "lua-expat"
--
-- (C) 2019 kakurasan
-- Licensed under GPLv3+

local https = require 'ssl.https'
local ltn12 = require 'ltn12'
local lxp = require 'lxp'


local api = {
    xml = 'https://www.nhk.or.jp/radio/config/config_web.xml',
    areas = {'sapporo', 'sendai', 'tokyo', 'nagoya',
             'osaka', 'hiroshima', 'matsuyama', 'fukuoka'},
    stations = {'r1', 'r2', 'fm'},
}

local parser_data = {
    hls_elem_name = nil,
    is_area_elem = false,
    is_specified_area = false,
    is_hls_elem = false,
    area = nil,
    m3u8 = nil,
}

local callbacks = {
    StartElement = function(parser, name)
        if parser_data.is_specified_area then
            if name == parser_data.hls_elem_name then
                parser_data.is_hls_elem = true
            end
        else
            parser_data.is_area_elem = name == 'area'
        end
    end,
    EndElement = function(parser, name)
        if parser_data.is_specified_area and name == 'data' then
            parser_data.is_specified_area = false
        elseif parser_data.is_hls_elem and name == parser_data.hls_elem_name then
            parser_data.is_hls_elem = false
        elseif parser_data.is_area_elem and name == 'area' then
            parser_data.is_area_elem = false
        end
    end,
    CharacterData = function(parser, data)
        if parser_data.is_hls_elem then
            parser_data.m3u8 = data
        elseif parser_data.is_area_elem and data == parser_data.area then
            parser_data.is_specified_area = true
        end
    end
}

function gen_usage()
    usage = 'USAGE: %s [AREA] ["r1", "r2", or "fm"]\nAvailable AREA:\n '
    for i, area in ipairs(api.areas) do
      usage = usage .. ('%s%s'):format(area, (i < #api.areas) and ', ' or '')
    end
    return usage:format(arg[0]) .. '\n'
end

do
    -- Check for number of arguments
    if #arg ~= 2 then
        io.stderr:write(gen_usage())
        os.exit(false)
    end

    -- Check for area
    local area = arg[1]
    local area_is_valid = false
    for _, valid_area in ipairs(api.areas) do
        if area == valid_area then
            area_is_valid = true
        end
    end
    if not area_is_valid then
        io.stderr:write(('ERROR: Invalid area (%s)\n\n'):format(area))
        io.stderr:write(gen_usage())
        os.exit(false)
    end

    -- Check for station
    local station = arg[2]
    local station_is_valid = false
    for _, valid_station in ipairs(api.stations) do
        if station == valid_station then
            station_is_valid = true
        end
    end
    if not station_is_valid then
        io.stderr:write(('ERROR: Invalid station (%s)\n\n'):format(station))
        io.stderr:write(gen_usage())
        os.exit(false)
    end

    -- Download/parse XML
    local xmldata = {}
    local _, code, _, _ = https.request {
        url = api.xml,
        sink = ltn12.sink.table(xmldata),
    }
    if code ~= 200 then
        io.stderr:write(('ERROR: Could not download XML (status %s)\n'):format(code))
        os.exit(false)
    end
    local p = lxp.new(callbacks)
    parser_data.hls_elem_name = station .. 'hls'
    parser_data.area = area
    p:parse(table.concat(xmldata))
    p:close()
    if parser_data.m3u8 then
        io.stdout:write(parser_data.m3u8)
    else
        io.stderr:write('ERROR: Could not find m3u8 URL.')
        os.exit(false)
    end
end

#! /usr/bin/python

# radiru-get-streamurl.py
# (C) 2019 kakurasan
# Licensed under GPLv3+

from __future__ import print_function

import xml.parsers.expat as expat
import sys
try:
    from urllib.parse import urlparse as parse_url
    from http.client import HTTPSConnection
except ImportError:
    from urlparse import urlparse as parse_url
    from httplib import HTTPSConnection


class Api:
    xml = 'https://www.nhk.or.jp/radio/config/config_web.xml'
    areas = ('sapporo', 'sendai', 'tokyo', 'nagoya',
             'osaka', 'hiroshima', 'matsuyama', 'fukuoka')
    stations = ('r1', 'r2', 'fm')


class ConfigWebParser:
    xml = None
    p = expat.ParserCreate()
    hls_elem_name = None
    is_area_elem = False
    is_specified_area = False
    is_hls_elem = False
    area = None
    m3u8 = None

    def __init__(self, area, station):
        self.xml = get_xml()
        self.hls_elem_name = station + 'hls'
        self.area = area
        self.p.StartElementHandler = self._expat_start_elem
        self.p.EndElementHandler = self._expat_end_elem
        self.p.CharacterDataHandler = self._expat_char_data

    def _expat_start_elem(self, name, attrs):
        if self.is_specified_area:
            if name == self.hls_elem_name:
                self.is_hls_elem = True
        else:
            self.is_area_elem = name == 'area'

    def _expat_end_elem(self, name):
        if self.is_specified_area and name == 'data':
            self.is_specified_area = False
        elif self.is_hls_elem and name == self.hls_elem_name:
            self.is_hls_elem = False
        elif self.is_area_elem and name == 'area':
            self.is_area_elem = False

    def _expat_char_data(self, data):
        if self.is_hls_elem:
            self.m3u8 = data
        elif self.is_area_elem and data == self.area:
            self.is_specified_area = True

    def get_m3u8(self):
        self.p.Parse(self.xml)
        if self.m3u8:
            return self.m3u8
        else:
            return None


def get_xml():
    o = parse_url(Api.xml)
    conn = HTTPSConnection(o.netloc)
    conn.request('GET', o.path)
    res = conn.getresponse()
    try:
        if res.status != 200:
            sys.exit(
              'ERROR: Could not download XML (status {})'.format(res.status))
        xmldata = res.read()
    finally:
        conn.close()
    return xmldata


def gen_usage():
    usage = 'USAGE: {} [AREA] ["r1", "r2", or "fm"]\nAvailable AREA:\n '
    for i, area in enumerate(Api.areas):
        usage += '{}{}'.format(area, ', ' if i < len(Api.areas) - 1 else '')
    return usage.format(__file__)


def main():
    # Check for number of arguments
    if len(sys.argv) != 3:
        sys.exit(gen_usage())

    # Check for area
    area = sys.argv[1]
    area_is_valid = False
    for valid_area in Api.areas:
        if area == valid_area:
            area_is_valid = True
    if not area_is_valid:
        print('ERROR: Invalid area ({})\n'.format(area), file=sys.stderr)
        sys.exit(gen_usage())

    # Check for station
    station = sys.argv[2]
    station_is_valid = False
    for valid_station in Api.stations:
        if station == valid_station:
            station_is_valid = True
    if not station_is_valid:
        print('ERROR: Invalid station ({})\n'.format(station), file=sys.stderr)
        sys.exit(gen_usage())

    # Download/parse XML
    streamurl = ConfigWebParser(area, station).get_m3u8()
    if streamurl:
        print(streamurl, end='')
    else:
        sys.exit('ERROR: Could not find m3u8 URL.')


if __name__ == '__main__':
    main()

#! /usr/bin/luajit

-- play-radiru.lua
-- This script requires "radiru-get-streamurl.lua"
--
-- (C) 2019 kakurasan
-- Licensed under GPLv3+

local helper_name = 'radiru-get-streamurl.lua'

function gen_usage()
    usage = 'USAGE: %s [AREA] ["r1", "r2", or "fm"]\n[DURATION]\n'
    return usage:format(arg[0])
end

do
    -- Check for number of arguments
    if #arg ~= 3 then
        io.stderr:write(gen_usage())
        os.exit(false)
    end

    local area, station, duration = arg[1], arg[2], arg[3]
    local f = io.popen(('realpath "%s"'):format(arg[0]))
    local scriptpath = f:read()
    f:close()
    f = io.popen(('dirname "%s"'):format(scriptpath))
    local scriptdir = f:read()
    f:close()
    local streamurl_helper = ('%s/%s'):format(scriptdir, helper_name)

    -- Get stream URL
    f = io.popen(('"%s" %s %s'):format(streamurl_helper, area, station))
    local streamurl = f:read('*all')
    f:close()

    -- Play
    os.execute(('mpv --really-quiet --end %s %s'):format(duration, streamurl))
end

#! /usr/bin/python

# play-radiru.py
# This script requires "radiru-get-streamurl.py"
#
# (C) 2019 kakurasan
# Licensed under GPLv3+

from __future__ import print_function

import subprocess
import sys
import os


helper_name = 'radiru-get-streamurl.py'


def gen_usage():
    usage = 'USAGE: {} [AREA] ["r1", "r2", or "fm"] [DURATION]'
    return usage.format(__file__)


def main():
    # Check for number of arguments
    if len(sys.argv) != 4:
        sys.exit(gen_usage())

    area, station, duration = sys.argv[1:]
    streamurl_helper = os.path.join(os.path.dirname(__file__), helper_name)

    # Get stream URL
    try:
        streamurl = subprocess.check_output((streamurl_helper, area, station))
    except subprocess.CalledProcessError as e:
        sys.exit(e.output)

    # Play
    subprocess.call(('mpv', '--really-quiet', '--end', duration, streamurl))


if __name__ == '__main__':
    main()

#! /usr/bin/luajit

-- rec-radiru.lua
-- This script requires "radiru-get-streamurl.lua"
--
-- (C) 2019 kakurasan
-- Licensed under GPLv3+

local helper_name = 'radiru-get-streamurl.lua'

function gen_usage()
    usage = 'USAGE: %s [AREA] ["r1", "r2", or "fm"]\n[DURATION] [OUTPUT]\n'
    return usage:format(arg[0])
end

do
    -- Check for number of arguments
    if #arg ~= 4 then
        io.stderr:write(gen_usage())
        os.exit(false)
    end

    local area, station, duration, outfile = arg[1], arg[2], arg[3], arg[4]
    local f = io.popen(('realpath "%s"'):format(arg[0]))
    local scriptpath = f:read()
    f:close()
    f = io.popen(('dirname "%s"'):format(scriptpath))
    local scriptdir = f:read()
    f:close()
    local streamurl_helper = ('%s/%s'):format(scriptdir, helper_name)

    -- Get stream URL
    f = io.popen(('"%s" %s %s'):format(streamurl_helper, area, station))
    local streamurl = f:read('*all')
    f:close()

    -- Record
    os.execute(
      ('ffmpeg -v quiet -i %s -t %s -acodec copy -y "%s"'):format(
       streamurl, duration, outfile))
end

#! /usr/bin/python

# rec-radiru.py
# This script requires "radiru-get-streamurl.py"
#
# (C) 2019 kakurasan
# Licensed under GPLv3+

from __future__ import print_function

import subprocess
import sys
import os


helper_name = 'radiru-get-streamurl.py'


def gen_usage():
    usage = 'USAGE: {} [AREA] ["r1", "r2", or "fm"] [DURATION] [OUTPUT]'
    return usage.format(__file__)


def main():
    # Check for number of arguments
    if len(sys.argv) != 5:
        sys.exit(gen_usage())

    area, station, duration, outfile = sys.argv[1:]
    streamurl_helper = os.path.join(os.path.dirname(__file__), helper_name)

    # Get stream URL
    try:
        streamurl = subprocess.check_output((streamurl_helper, area, station))
    except subprocess.CalledProcessError as e:
        sys.exit(e.output)

    # Record
    subprocess.call(('ffmpeg', '-v', 'quiet', '-i', streamurl,
                     '-t', duration, '-acodec', 'copy', '-y', outfile))


if __name__ == '__main__':
    main()