media/radio.py

   1 import re
   2 import logging
   3 import struct
   4 import requests
   5 import traceback
   6 import hashlib
   7
   8 from media.item import BaseItem
   9 from media.item import item_builders, item_loaders, item_id_generators
  10 import constants
  11
  12 log = logging.getLogger("bot")
  13
  14
  15 def get_radio_server_description(url):
  16     global log
  17
  18     log.debug("radio: fetching radio server description")
  19     p = re.compile('(https?://[^/]*)', re.IGNORECASE)
  20     res = re.search(p, url)
  21     base_url = res.group(1)
  22     url_icecast = base_url + '/status-json.xsl'
  23     url_shoutcast = base_url + '/stats?json=1'
  24     try:
  25         r = requests.get(url_shoutcast, timeout=10)
  26         data = r.json()
  27         title_server = data['servertitle']
  28         return title_server
  29         # logging.info("TITLE FOUND SHOUTCAST: " + title_server)
  30     except (requests.exceptions.ConnectionError,
  31             requests.exceptions.HTTPError,
  32             requests.exceptions.ReadTimeout,
  33             requests.exceptions.Timeout):
  34         error_traceback = traceback.format_exc()
  35         error = error_traceback.rstrip().split("\n")[-1]
  36         log.debug("radio: unsuccessful attempts on fetching radio description (shoutcast): " + error)
  37     except ValueError:
  38         return url
  39
  40     try:
  41         r = requests.get(url_icecast, timeout=10)
  42         data = r.json()
  43         source = data['icestats']['source']
  44         if type(source) is list:
  45             source = source[0]
  46         title_server = source['server_name']
  47         if 'server_description' in source:
  48             title_server += ' - ' + source['server_description']
  49         # logging.info("TITLE FOUND ICECAST: " + title_server)
  50         return title_server
  51     except (requests.exceptions.ConnectionError,
  52             requests.exceptions.HTTPError,
  53             requests.exceptions.ReadTimeout,
  54             requests.exceptions.Timeout):
  55         error_traceback = traceback.format_exc()
  56         error = error_traceback.rstrip().split("\n")[-1]
  57         log.debug("radio: unsuccessful attempts on fetching radio description (icecast): " + error)
  58
  59     return url
  60
  61
  62 def get_radio_title(url):
  63     global log
  64
  65     log.debug("radio: fetching radio server description")
  66     try:
  67         r = requests.get(url, headers={'Icy-MetaData': '1'}, stream=True, timeout=10)
  68         icy_metaint_header = int(r.headers['icy-metaint'])
  69         r.raw.read(icy_metaint_header)
  70
  71         metadata_length = struct.unpack('B', r.raw.read(1))[0] * 16  # length byte
  72         metadata = r.raw.read(metadata_length).rstrip(b'\0')
  73         logging.info(metadata)
  74         # extract title from the metadata
  75         m = re.search(br"StreamTitle='([^']*)';", metadata)
  76         if m:
  77             title = m.group(1)
  78             if title:
  79                 return title.decode()
  80     except (requests.exceptions.ConnectionError,
  81             requests.exceptions.HTTPError,
  82             requests.exceptions.ReadTimeout,
  83             requests.exceptions.Timeout,
  84             KeyError):
  85         log.debug("radio: unsuccessful attempts on fetching radio title (icy)")
  86     return url
  87
  88
  89 def radio_item_builder(bot, **kwargs):
  90     if 'name' in kwargs:
  91         return RadioItem(bot, kwargs['url'], kwargs['name'])
  92     else:
  93         return RadioItem(bot, kwargs['url'], '')
  94
  95
  96 def radio_item_loader(bot, _dict):
  97     return RadioItem(bot, "", "", _dict)
  98
  99
 100 def radio_item_id_generator(**kwargs):
 101     return hashlib.md5(kwargs['url'].encode()).hexdigest()
 102
 103
 104 item_builders['radio'] = radio_item_builder
 105 item_loaders['radio'] = radio_item_loader
 106 item_id_generators['radio'] = radio_item_id_generator
 107
 108
 109 class RadioItem(BaseItem):
 110     def __init__(self, bot, url, name="", from_dict=None):
 111         if from_dict is None:
 112             super().__init__(bot)
 113             self.url = url
 114             if not name:
 115                 self.title = get_radio_server_description(self.url)  # The title of the radio station
 116             else:
 117                 self.title = name
 118             self.id = hashlib.md5(url.encode()).hexdigest()
 119         else:
 120             super().__init__(bot, from_dict)
 121             self.url = from_dict['url']
 122             self.title = from_dict['title']
 123
 124         self.type = "radio"
 125
 126     def validate(self):
 127         self.version += 1  # 0 -> 1, notify the wrapper to save me when validate() is visited the first time
 128         return True
 129
 130     def is_ready(self):
 131         return True
 132
 133     def uri(self):
 134         return self.url
 135
 136     def to_dict(self):
 137         dict = super().to_dict()
 138         dict['url'] = self.url
 139         dict['title'] = self.title
 140
 141         return dict
 142
 143     def format_debug_string(self):
 144         return "[radio] {name} ({url})".format(
 145             name=self.title,
 146             url=self.url
 147         )
 148
 149     def format_song_string(self, user):
 150         return constants.strings("radio_item",
 151                                  url=self.url,
 152                                  title=get_radio_title(self.url),  # the title of current song
 153                                  name=self.title,  # the title of radio station
 154                                  user=user
 155                                  )
 156
 157     def format_current_playing(self, user):
 158         return constants.strings("now_playing", item=self.format_song_string(user))
 159
 160     def format_short_string(self):
 161         return self.title if self.title else self.url
 162
 163     def display_type(self):
 164         return constants.strings("radio")