media/radio.py

   1 import re
   2 import logging
   3 import json
   4 import http.client
   5 import struct
   6 import requests
   7 import traceback
   8
   9 log = logging.getLogger("bot")
  10
  11 def get_radio_server_description(url):
  12     global log
  13
  14     p = re.compile('(https?\:\/\/[^\/]*)', re.IGNORECASE)
  15     res = re.search(p, url)
  16     base_url = res.group(1)
  17     url_icecast = base_url + '/status-json.xsl'
  18     url_shoutcast = base_url + '/stats?json=1'
  19     title_server = None
  20     try:
  21         r = requests.get(url_shoutcast, timeout=5)
  22         data = r.json()
  23         title_server = data['servertitle']
  24         return title_server
  25         # logging.info("TITLE FOUND SHOUTCAST: " + title_server)
  26     except (requests.exceptions.ConnectionError, requests.exceptions.HTTPError, requests.exceptions.Timeout) as e:
  27         error_traceback = traceback.format_exc()
  28         error = error_traceback.rstrip().split("\n")[-1]
  29         log.debug("radio: unsuccessful attempts on fetching radio description (shoutcast): " + error)
  30     except ValueError:
  31         return False # ?
  32
  33     try:
  34         r = requests.get(url_icecast, timeout=5)
  35         data = r.json()
  36         source = data['icestats']['source']
  37         if type(source) is list:
  38             source = source[0]
  39         title_server = source['server_name']
  40         if 'server_description' in source:
  41             title_server += ' - ' + source['server_description']
  42         # logging.info("TITLE FOUND ICECAST: " + title_server)
  43         return title_server
  44     except (requests.exceptions.ConnectionError, requests.exceptions.HTTPError, requests.exceptions.Timeout) as e:
  45         error_traceback = traceback.format_exc()
  46         error = error_traceback.rstrip().split("\n")[-1]
  47         log.debug("radio: unsuccessful attempts on fetching radio description (icecast): " + error)
  48
  49     return url
  50
  51
  52 def get_radio_title(url):
  53     try:
  54         r = requests.get(url, headers={'Icy-MetaData': '1'}, stream=True, timeout=5)
  55         icy_metaint_header = int(r.headers['icy-metaint'])
  56         r.raw.read(icy_metaint_header)
  57
  58         metadata_length = struct.unpack('B', r.raw.read(1))[0] * 16  # length byte
  59         metadata = r.raw.read(metadata_length).rstrip(b'\0')
  60         logging.info(metadata)
  61         # extract title from the metadata
  62         m = re.search(br"StreamTitle='([^']*)';", metadata)
  63         if m:
  64             title = m.group(1)
  65             if title:
  66                 return title.decode()
  67     except (requests.exceptions.ConnectionError, requests.exceptions.HTTPError) as e:
  68         pass
  69     return url