app/common/innertube.py

   1 # functions that deal with parsing data from youtube's internal API ("innertube")
   2
   3 from urllib.parse import parse_qs, urlparse
   4
   5 def listfind(obj, key):
   6     """
   7     given a list of dicts, where one dict contains a given key, return said key.
   8     """
   9     return next(iter([ obj[key] for obj in obj if key in obj.keys() ]),{})
  10 def listget(obj, index, fallback=None):
  11     return next(iter(obj[index:]), fallback)
  12
  13 def prepare_searchresults(yt_results):
  14     contents = listfind(yt_results, 'response') \
  15         .get('contents',{})\
  16         .get('twoColumnSearchResultsRenderer',{})\
  17         .get('primaryContents',{})\
  18         .get('sectionListRenderer',{})\
  19         .get('contents',[])
  20     contents = listfind(contents, 'itemSectionRenderer').get('contents',[])
  21
  22     return parse_result_items(contents)
  23 def prepare_infocards(metadata):
  24     cards = metadata.get('cards',{}).get('cardCollectionRenderer',{}).get('cards',[])
  25     return [parse_infocard(card) for card in cards if card is not None]
  26
  27 def prepare_endcards(metadata):
  28     endsc = metadata.get('endscreen',{}).get('endscreenRenderer',{}).get('elements',[])
  29     return [parse_endcard(card) for card in endsc if card is not None]
  30
  31 def mkthumbs(thumbs):
  32     output = {str(e['height']): e['url'] for e in thumbs}
  33     largest=next(iter(sorted(output.keys(),reverse=True,key=int)),None)
  34     return {**output, 'largest': largest}
  35
  36 def clean_url(url):
  37     # externals URLs are redirected through youtube.com/redirect, but we
  38     # may encounter internal URLs, too
  39     return parse_qs(urlparse(url).query).get('q',[url])[0]
  40
  41 def toInt(s, fallback=0):
  42     if s is None:
  43         return fallback
  44     try:
  45         return int(''.join(filter(str.isdigit, s)))
  46     except ValueError:
  47         return fallback
  48
  49 # Remove left-/rightmost word from string:
  50 delL = lambda s: s.partition(' ')[2]
  51
  52 def age(s):
  53     if s is None: # missing from autogen'd music, some livestreams
  54         return None
  55     # Some livestreams have "Streamed 7 hours ago"
  56     s = s.replace("Streamed ","")
  57     # Now, everything should be in the form "1 year ago"
  58     value, unit, _ = s.split(" ")
  59     suffix = dict(
  60         month='mn',
  61         months='mn',
  62     ).get(unit, unit[0]) # first letter otherwise (e.g. year(s) => y)
  63
  64     return f"{value}{suffix}"
  65
  66 def parse_result_items(items):
  67     # TODO: use .get() for most non-essential attributes
  68     """
  69     parses youtube search response into an easier to use format.
  70     """
  71     results = []
  72     for item in items:
  73         key = next(iter(item.keys()), None)
  74         content = item[key]
  75         if key == 'videoRenderer':
  76             is_live = listfind(content.get('badges',[]), 'metadataBadgeRenderer').get('style') == 'BADGE_STYLE_TYPE_LIVE_NOW'
  77             results.append({'type': 'VIDEO', 'content': {
  78                 'video_id': content['videoId'],
  79                 'title': content['title']['runs'][0]['text'],
  80                 'author': content['longBylineText']['runs'][0]['text'] or \
  81                           content['shortBylineText']['runs'][0]['text'],
  82                 'channel_id': content['ownerText']['runs'][0] \
  83                     ['navigationEndpoint']['browseEndpoint']['browseId'],
  84                 'length': content.get('lengthText',{}).get('simpleText') \
  85                     if not is_live else 'LIVE', # "44:07", "1:41:50"
  86                 'views': toInt(content.get('viewCountText',{}).get('simpleText') or # "123,456 views"
  87                     listget(content.get('viewCountText',{}).get('runs'),0,{}).get('text')), # "1,234 watching"
  88                 'published': age(content.get('publishedTimeText',{}).get('simpleText')),
  89             }})
  90         elif key == 'playlistRenderer':
  91             results.append({'type': 'PLAYLIST', 'content': {
  92                 'playlist_id': content['navigationEndpoint']['watchEndpoint']['playlistId'],
  93                 'video_id': content['navigationEndpoint']['watchEndpoint']['videoId'],
  94                 'title': content['title']['simpleText'],
  95                 'author': content['longBylineText']['runs'][0]['text'] or
  96                      content['shortBylineText']['runs'][0]['text'],
  97                 'channel_id': content['longBylineText']['runs'][0]['navigationEndpoint']['browseEndpoint']['browseId'], # OR .shortBylineText
  98                 'n_videos': toInt(content['videoCount']),
  99             }})
 100         elif key == 'radioRenderer': # "Mix" playlists
 101             results.append({'type': 'PLAYLIST', 'content': {
 102                 'playlist_id': content['playlistId'],
 103                 'video_id': content['navigationEndpoint']['watchEndpoint']['videoId'],
 104                 'title': content['title']['simpleText'],
 105                 'author': content['longBylineText']['simpleText'] or \
 106                     content['shortBylineText']['simpleText'] , # always "YouTube"
 107                 'channel_id': None,
 108                 'n_videos': content['videoCountShortText']['runs'][0]['text'] or \
 109                     content['videoCountText']['runs'][0]['text'],
 110                     # videoCountShortText: "50+"; videoCountText: "50+ videos"
 111             }})
 112         elif key == 'channelRenderer':
 113             results.append({'type': 'CHANNEL', 'content': {
 114                 'channel_id': content['channelId'],
 115                 'title': content['title']['simpleText'],
 116                 'icons': mkthumbs(content['thumbnail']['thumbnails']),
 117                 'subscribers': content['subscriberCountText']['simpleText'], # "2.47K subscribers"
 118             }})
 119         elif key == 'shelfRenderer':
 120             results.extend([item for item in
 121                 parse_result_items(content['content']['verticalListRenderer']['items'])
 122             ])
 123         elif key == 'movieRenderer': # movies to buy/rent
 124             pass
 125         elif key == 'horizontalCardListRenderer':
 126             # suggested searches: .cards[].searchRefinementCardRenderer.query.runs[].text
 127             pass
 128         else:
 129             import pprint
 130             content = {'error': f"{key} is not implemented; <pre>{pprint.pformat(item)}</pre>"}
 131             results.append({'type': key, 'content': content})
 132     return results
 133
 134 def parse_infocard(card):
 135     """
 136     parses a single infocard into a format that's easier to handle.
 137     """
 138     card = card['cardRenderer']
 139     ctype = list(card['content'].keys())[0]
 140     content = card['content'][ctype]
 141     if ctype == "pollRenderer":
 142         return {'type': "POLL", 'content': {
 143             'question': content['question']['simpleText'],
 144             'answers': [(a['text']['simpleText'],a['numVotes']) \
 145                 for a in content['choices']],
 146         }}
 147     elif ctype == "videoInfoCardContentRenderer":
 148         is_live = content.get('badge',{}).get('liveBadgeRenderer') is not None
 149         return {'type': "VIDEO", 'content': {
 150             'video_id': content['action']['watchEndpoint']['videoId'],
 151             'title': content['videoTitle']['simpleText'],
 152             'author': delL(content['channelName']['simpleText']),
 153             'length': content.get('lengthString',{}).get('simpleText') \
 154                 if not is_live else "LIVE", # "23:03"
 155             'views': toInt(content.get('viewCountText',{}).get('simpleText')),
 156                 # XXX: views sometimes "Starts: July 31, 2020 at 1:30 PM"
 157         }}
 158     elif ctype == "playlistInfoCardContentRenderer":
 159         return {'type': "PLAYLIST", 'content': {
 160             'playlist_id': content['action']['watchEndpoint']['playlistId'],
 161             'video_id': content['action']['watchEndpoint']['videoId'],
 162             'title': content['playlistTitle']['simpleText'],
 163             'author': delL(content['channelName']['simpleText']),
 164             'n_videos': toInt(content['playlistVideoCount']['simpleText']),
 165         }}
 166     elif ctype == "simpleCardContentRenderer" and \
 167             'urlEndpoint' in content['command']:
 168         return {'type': "WEBSITE", 'content': {
 169             'url': clean_url(content['command']['urlEndpoint']['url']),
 170             'domain': content['displayDomain']['simpleText'],
 171             'title': content['title']['simpleText'],
 172             # XXX: no thumbnails for infocards
 173         }}
 174     elif ctype == "collaboratorInfoCardContentRenderer":
 175         return {'type': "CHANNEL", 'content': {
 176             'channel_id': content['endpoint']['browseEndpoint']['browseId'],
 177             'title': content['channelName']['simpleText'],
 178             'icons': mkthumbs(content['channelAvatar']['thumbnails']),
 179             'subscribers': content.get('subscriberCountText',{}).get('simpleText',''), # "545K subscribers"
 180         }}
 181     else:
 182         import pprint
 183         content = {'error': f"{ctype} is not implemented; <pre>{pprint.pformat(card)}</pre>"}
 184         # TODO!!!
 185         return None
 186
 187 def parse_endcard(card):
 188     """
 189     parses a single endcard into a format that's easier to handle.
 190     """
 191     card = card.get('endscreenElementRenderer', card) #only sometimes nested
 192     ctype = card['style']
 193     if ctype == "CHANNEL":
 194         return {'type': ctype, 'content': {
 195             'channel_id': card['endpoint']['browseEndpoint']['browseId'],
 196             'title': card['title']['simpleText'],
 197             'icons': mkthumbs(card['image']['thumbnails']),
 198         }}
 199     elif ctype == "VIDEO":
 200         return {'type': ctype, 'content': {
 201             'video_id': card['endpoint']['watchEndpoint']['videoId'],
 202             'title': card['title']['simpleText'],
 203             'length': card['videoDuration']['simpleText'],  # '12:21'
 204             'views': toInt(card['metadata']['simpleText']),
 205             # XXX: no channel name
 206         }}
 207     elif ctype == "PLAYLIST":
 208         return {'type': ctype, 'content': {
 209             'playlist_id': card['endpoint']['watchEndpoint']['playlistId'],
 210             'video_id': card['endpoint']['watchEndpoint']['videoId'],
 211             'title': card['title']['simpleText'],
 212             'author': delL(card['metadata']['simpleText']),
 213             'n_videos': toInt(card['playlistLength']['simpleText']),
 214         }}
 215     elif ctype == "WEBSITE" or ctype == "CREATOR_MERCHANDISE":
 216         url = clean_url(card['endpoint']['urlEndpoint']['url'])
 217         return {'type': "WEBSITE", 'content': {
 218             'url': url,
 219             'domain': urlparse(url).netloc,
 220             'title': card['title']['simpleText'],
 221             'icons': mkthumbs(card['image']['thumbnails']),
 222         }}
 223     else:
 224         import pprint
 225         content = {'error': f"{ctype} is not implemented; <pre>{pprint.pformat(card)}</pre>"}
 226         # TODO!!!
 227         return None