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 import re
   5
   6 def findall(obj, key):
   7     """
   8     given a list of dicts, where one dict contains a given key, return said key.
   9     """
  10     if obj is None: return []
  11     return [ obj[key] for obj in obj if key in obj.keys() ]
  12 def listget(obj, index, fallback=None):
  13     if obj is None: return fallback
  14     return next(iter(obj[index:]), fallback)
  15 flatten = lambda l: [item for sublist in l for item in sublist] # https://stackoverflow.com/a/952952
  16 first = lambda l: next(iter(l),{})
  17 listfind = lambda obj,key: first(findall(obj,key))
  18
  19 class G:
  20     """
  21     null-coalescing version of dict.get() that also works on lists.
  22
  23     the | operator is overloaded to achieve similar looking code to jq(1) filters.
  24     the first found key is used: dict(foo=1)|G('bar','foo') returns 1.
  25     """
  26     def __init__(self, *keys):
  27         self.keys = keys
  28     def __ror__(self, other):
  29         for key in self.keys:
  30             try:    return other[key]
  31             except: continue
  32         return None
  33     class _Text:
  34         """ parses youtube's .runs[].text and .simpleText variants """
  35         def __ror__(self, other): # Note: only returning runs[0], not concat'ing all!
  36             return other|G('simpleText') or other|G('runs')|G(0)|G('text')
  37     text = _Text()
  38 class Select:
  39     """ |Select('foo') returns the first foo in list, |Select(all='foo') returns all foos. """
  40     def __init__(self, key=None, *, all=None):
  41         self.key = key or all
  42         self.all = all
  43     def __ror__(self, other):
  44         try: items = [ other[self.key] for other in other if self.key in other.keys() ]
  45         except: items = []
  46         return items if self.all else items|G(0)
  47 class A:
  48     """ apply """
  49     def __init__(self, f, *args):
  50         self.f = f
  51         self.args = args
  52     def __ror__(self, other):
  53         return self.f(other, *self.args)
  54     class _Int:
  55         def __ror__(self, other):
  56             try:    return int(''.join(filter(str.isdigit, other)))
  57             except: return None
  58     int = _Int()
  59
  60
  61 def prepare_searchresults(yt_results):
  62     contents = listfind(yt_results, 'response') \
  63         .get('contents',{})\
  64         .get('twoColumnSearchResultsRenderer',{})\
  65         .get('primaryContents',{})\
  66         .get('sectionListRenderer',{})\
  67         .get('contents',[])
  68     contents = flatten([c.get('contents',[]) for c in contents|Select(all='itemSectionRenderer')])
  69
  70     return parse_result_items(contents)
  71
  72 def prepare_infocards(metadata):
  73     cards = metadata.get('cards',{}).get('cardCollectionRenderer',{}).get('cards',[])
  74     return list(filter(None, map(parse_infocard, cards)))
  75
  76 def prepare_endcards(metadata):
  77     endsc = metadata.get('endscreen',{}).get('endscreenRenderer',{}).get('elements',[])
  78     return list(filter(None, map(parse_endcard, endsc)))
  79
  80 def prepare_channel(response, channel_id):
  81     meta1 = response|G('metadata')|G('channelMetadataRenderer')
  82     meta2 = response|G('microformat')|G('microformatDataRenderer')
  83     title = meta1|G('title') or meta2|G('title')
  84     descr = meta1|G('description') or meta2|G('description') # meta2.description is capped at 160chars
  85     thumb = mkthumbs((meta2|G('thumbnail') or meta1|G('avatar'))|G('thumbnails') or {}) # .avatar ~ 900px
  86
  87     contents = response|G('continuationContents')
  88     if not contents: # overran end of list
  89         return title, descr, thumb, [], False
  90
  91     unparsed = contents|G('gridContinuation')|G('items') or \
  92             contents|G('sectionListContinuation')|G('contents') or []
  93     items, extra = parse_channel_items(unparsed, channel_id, title)
  94     more = (
  95         contents
  96         |G('gridContinuation', 'sectionListContinuation')
  97         |G('continuations')
  98         |Select('nextContinuationData')
  99         |G('continuation')
 100     )
 101
 102     return title, descr, thumb, items, more
 103
 104 def prepare_playlist(result):
 105     contents = result['continuationContents']
 106     unparsed = contents['playlistVideoListContinuation'].get('contents',[])
 107     more = (
 108         contents
 109         |G('playlistVideoListContinuation')
 110         |G('continuations')
 111         |Select('nextContinuationData')
 112         |G('continuation')
 113     )
 114
 115     return list(filter(None, map(parse_playlist, unparsed))), more
 116
 117 def mkthumbs(thumbs):
 118     output = {str(e['height']): e['url'] for e in thumbs}
 119     largest=next(iter(sorted(output.keys(),reverse=True,key=int)),None)
 120     return {**output, 'largest': largest}
 121
 122 def clean_url(url):
 123     # externals URLs are redirected through youtube.com/redirect, but we
 124     # may encounter internal URLs, too
 125     return parse_qs(urlparse(url).query).get('q',[url])[0]
 126
 127 def toInt(s, fallback=0):
 128     if s is None:
 129         return fallback
 130     try:
 131         return int(''.join(filter(str.isdigit, s)))
 132     except ValueError:
 133         return fallback
 134
 135 # Remove left-/rightmost word from string:
 136 delL = lambda s: s.partition(' ')[2]
 137
 138 def age(s):
 139     if s is None: # missing from autogen'd music, some livestreams
 140         return None
 141     # Some livestreams have "Streamed 7 hours ago"
 142     s = s.replace("Streamed ","")
 143     # Now, everything should be in the form "1 year ago"
 144     value, unit, _ = s.split(" ")
 145     suffix = dict(
 146         minute='min',
 147         minutes='min',
 148     ).get(unit, unit[0]) # first letter otherwise (e.g. year(s) => y)
 149
 150     return f"{value}{suffix}"
 151
 152 def log_unknown_card(data):
 153     import json
 154     try:
 155         from flask import request
 156         source = request.url
 157     except: source = "unknown"
 158     with open("/tmp/innertube.err", "a") as f:
 159         f.write(f"\n/***** {source} *****/\n")
 160         json.dump(data, f, indent=2)
 161
 162 def parse_result_items(items):
 163     # TODO: use .get() for most non-essential attributes
 164     """
 165     parses youtube search response into an easier to use format.
 166     """
 167     results = []
 168     extras = []
 169     for item in items:
 170         key = next(iter(item.keys()), None)
 171         content = item[key]
 172         if key == 'videoRenderer':
 173             results.append({'type': 'VIDEO', 'content': {
 174                 'video_id': content['videoId'],
 175                 'title': content['title']|G.text,
 176                 'author': content|G('longBylineText','shortBylineText')|G.text,
 177                 'channel_id': content|G('ownerText')|G('runs')|G(0) \
 178                     |G('navigationEndpoint')|G('browseEndpoint')|G('browseId'),
 179                 'length': content|G('lengthText')|G.text, # "44:07", "1:41:50"
 180                 'views': content|G('viewCountText')|G.text|A.int or 0, # "1,234 {views|watching}", absent on 0 views
 181                 'published': content|G('publishedTimeText')|G('simpleText')|A(age),
 182                 'live': content|G('badges')|Select('metadataBadgeRenderer')|G('style')=='BADGE_STYLE_TYPE_LIVE_NOW',
 183             }})
 184         elif key in ['playlistRenderer', 'radioRenderer', 'showRenderer']: # radio == "Mix" playlist, show == normal playlist, specially displayed
 185             results.append({'type': 'PLAYLIST', 'content': {
 186                 'playlist_id': content['navigationEndpoint']|G('watchEndpoint')|G('playlistId') or \
 187                     content|G('playlistId'), # COURSE/"learning playlist"
 188                 'video_id': content['navigationEndpoint']|G('watchEndpoint')|G('videoId') or \
 189                     videoid_from_thumbnail(content), # learning playlist
 190                 'title': content['title']|G.text,
 191                 # Note: learning playlists have no author/channel_id
 192                 'author': content|G('longBylineText','shortBylineText')|G.text,
 193                 'channel_id': content|G('longBylineText','shortBylineText')|G('runs')|G(0) \
 194                     |G('navigationEndpoint')|G('browseEndpoint')|G('browseId'),
 195                 'n_videos': content|G('videoCount')|A.int or \
 196                     content|G('videoCountShortText','videoCountText')|G.text, # "Mix" playlists
 197             }})
 198         elif key == 'channelRenderer':
 199             results.append({'type': 'CHANNEL', 'content': {
 200                 'channel_id': content['channelId'],
 201                 'title': content['title']|G.text,
 202                 'icons': content['thumbnail']['thumbnails']|A(mkthumbs),
 203                 'subscribers': content|G('subscriberCountText')|G('simpleText'), # "2.47K subscribers"
 204             }})
 205         elif key == 'shelfRenderer':
 206             subkey = next(iter(content['content'].keys()), None) #verticalListRenderer/horizontalMovieListRenderer
 207             r, e = parse_result_items(content['content'][subkey]['items'])
 208             results.extend(r)
 209             extras.extend(e)
 210         elif key in ['movieRenderer', 'gridMovieRenderer']: # movies to buy/rent
 211             pass # gMR.{videoId,title.runs[].text,lengthText.simpleText}
 212         elif key in ['carouselAdRenderer','searchPyvRenderer','promotedSparklesTextSearchRenderer']: # haha, no.
 213             pass
 214         elif key == 'horizontalCardListRenderer':
 215             # suggested searches: .cards[].searchRefinementCardRenderer.query.runs[].text
 216             pass
 217         elif key == 'emergencyOneboxRenderer': # suicide prevention hotline
 218             pass
 219         elif key in ['clarificationRenderer', 'infoPanelContainerRenderer']: # COVID-19/conspiracy theory infos
 220             pass
 221         elif key == 'webAnswerRenderer': # "Result from the web"
 222             pass
 223         elif key == 'didYouMeanRenderer' or key == 'showingResultsForRenderer':
 224             extras.append({
 225                 'type': 'spelling',
 226                 'query': content['correctedQueryEndpoint']['searchEndpoint']['query'], # non-misspelled query
 227                 'autocorrected': key == 'showingResultsForRenderer',
 228             })
 229         elif key == 'messageRenderer': # "No more results"
 230             extras.append({
 231                 'type': 'message',
 232                 'message': content|G('title','text')|G.text,
 233             })
 234         elif key == 'backgroundPromoRenderer': # e.g. "no results"
 235             extras.append({
 236                 'type': content['icon']['iconType'],
 237                 'message': content['title']|G.text,
 238             })
 239         else:
 240             log_unknown_card(item)
 241     return results, extras
 242
 243 def parse_infocard(card):
 244     """
 245     parses a single infocard into a format that's easier to handle.
 246     """
 247     card = card['cardRenderer']
 248     ctype = list(card['content'].keys())[0]
 249     content = card['content'][ctype]
 250     if ctype == "pollRenderer":
 251         return {'type': "POLL", 'content': {
 252             'question': content['question']['simpleText'],
 253             'answers': [(a['text']['simpleText'],a['numVotes']) \
 254                 for a in content['choices']],
 255         }}
 256     elif ctype == "videoInfoCardContentRenderer":
 257         is_live = content.get('badge',{}).get('liveBadgeRenderer') is not None
 258         return {'type': "VIDEO", 'content': {
 259             'video_id': content['action']['watchEndpoint']['videoId'],
 260             'title': content['videoTitle']['simpleText'],
 261             'author': delL(content['channelName']['simpleText']),
 262             'length': content.get('lengthString',{}).get('simpleText') \
 263                 if not is_live else "LIVE", # "23:03"
 264             'views': toInt(content.get('viewCountText',{}).get('simpleText')),
 265                 # XXX: views sometimes "Starts: July 31, 2020 at 1:30 PM"
 266         }}
 267     elif ctype == "playlistInfoCardContentRenderer":
 268         return {'type': "PLAYLIST", 'content': {
 269             'playlist_id': content['action']['watchEndpoint']['playlistId'],
 270             'video_id': content['action']['watchEndpoint']['videoId'],
 271             'title': content['playlistTitle']['simpleText'],
 272             'author': delL(content['channelName']['simpleText']),
 273             'n_videos': toInt(content['playlistVideoCount']['simpleText']),
 274         }}
 275     elif ctype == "simpleCardContentRenderer" and \
 276             'urlEndpoint' in content['command']:
 277         return {'type': "WEBSITE", 'content': {
 278             'url': clean_url(content['command']['urlEndpoint']['url']),
 279             'domain': content['displayDomain']['simpleText'],
 280             'title': content['title']['simpleText'],
 281             # XXX: no thumbnails for infocards
 282         }}
 283     elif ctype == "collaboratorInfoCardContentRenderer":
 284         return {'type': "CHANNEL", 'content': {
 285             'channel_id': content['endpoint']['browseEndpoint']['browseId'],
 286             'title': content['channelName']['simpleText'],
 287             'icons': mkthumbs(content['channelAvatar']['thumbnails']),
 288             'subscribers': content.get('subscriberCountText',{}).get('simpleText',''), # "545K subscribers"
 289         }}
 290     else:
 291         log_unknown_card(card)
 292         return None
 293
 294 def parse_endcard(card):
 295     """
 296     parses a single endcard into a format that's easier to handle.
 297     """
 298     card = card.get('endscreenElementRenderer', card) #only sometimes nested
 299     ctype = card['style']
 300     if ctype == "CHANNEL":
 301         return {'type': ctype, 'content': {
 302             'channel_id': card['endpoint']['browseEndpoint']['browseId'],
 303             'title': card['title']|G.text,
 304             'icons': mkthumbs(card['image']['thumbnails']),
 305         }}
 306     elif ctype == "VIDEO":
 307         if not 'endpoint' in card: return None # title == "This video is unavailable."
 308         return {'type': ctype, 'content': {
 309             'video_id': card['endpoint']['watchEndpoint']['videoId'],
 310             'title': card['title']|G.text,
 311             'length': card['videoDuration']|G.text,  # '12:21'
 312             'views': toInt(card['metadata']|G.text),
 313             # XXX: no channel name
 314         }}
 315     elif ctype == "PLAYLIST":
 316         return {'type': ctype, 'content': {
 317             'playlist_id': card['endpoint']['watchEndpoint']['playlistId'],
 318             'video_id': card['endpoint']['watchEndpoint']['videoId'],
 319             'title': card['title']|G.text,
 320             'author': delL(card['metadata']|G.text),
 321             'n_videos': toInt(card['playlistLength']|G.text),
 322         }}
 323     elif ctype == "WEBSITE" or ctype == "CREATOR_MERCHANDISE":
 324         url = clean_url(card['endpoint']['urlEndpoint']['url'])
 325         return {'type': "WEBSITE", 'content': {
 326             'url': url,
 327             'domain': urlparse(url).netloc,
 328             'title': card['title']|G.text,
 329             'icons': mkthumbs(card['image']['thumbnails']),
 330         }}
 331     else:
 332         log_unknown_card(card)
 333         return None
 334
 335 def videoid_from_thumbnail(content):
 336     # learning playlist; example: PL96C35uN7xGJu6skU4TBYrIWxggkZBrF5 (/user/enyay/playlists)
 337     return re.match(r"https?://i.ytimg.com/vi/([-_0-9a-zA-Z]{11})|()",
 338         listget(listget(content.get('thumbnails',[]),0,{}).get('thumbnails',[]),0,{}).get('url','')
 339     ).group(1)
 340
 341 def parse_channel_items(items, channel_id, author):
 342     result = []
 343     extra = []
 344     for item in items:
 345         key = next(iter(item.keys()), None)
 346         content = item[key]
 347         if key in ["gridVideoRenderer", "videoRenderer", "videoCardRenderer"]:
 348             # only videoCardRenderer (topic channels) has author and channel, others fall back to supplied ones.
 349             result.append({'type': 'VIDEO', 'content': {
 350                 'video_id': content['videoId'],
 351                 'title': content['title'].get('simpleText') or content['title'].get('runs',[{}])[0].get('text'),
 352                 'author': content.get('bylineText',{}).get('runs',[{}])[0].get('text') or author,
 353                 'channel_id': content.get('bylineText',{}).get('runs',[{}])[0] \
 354                     .get('navigationEndpoint',{}).get('browseEndpoint',{}).get('browseId') or channel_id,
 355                 'length': (content.get('lengthText',{}).get('simpleText') or # topic channel
 356                     listfind(content.get('thumbnailOverlays',[]),'thumbnailOverlayTimeStatusRenderer')
 357                         .get('text',{}).get('simpleText')),
 358                 # topic channel: .metadataText.simpleText = "22M views \u00b7 2 months ago"
 359                 'views': toInt(content.get('viewCountText',{}).get('simpleText')),
 360                 'published': age(content.get('publishedTimeText',{}).get('simpleText')),
 361             }})
 362         elif key == "gridPlaylistRenderer" or key == "playlistRenderer":
 363             result.append({'type': 'PLAYLIST', 'content': {
 364                 'playlist_id': content['navigationEndpoint'].get('watchEndpoint',{}).get('playlistId') or content.get('playlistId'),
 365                 'video_id': content['navigationEndpoint'].get('watchEndpoint',{}).get('videoId',{}) or videoid_from_thumbnail(content),
 366                 'title': (content['title'].get('simpleText') or # playlistRenderer
 367                     content['title']['runs'][0]['text']), # gridPlaylistRenderer
 368                 'author': author,
 369                 'channel_id': channel_id,
 370                 'n_videos': toInt(content.get('videoCount') or # playlistRenderer
 371                     content.get('videoCountShortText',{}).get('simpleText') or # grid(1)
 372                     content.get('videoCountText',{}).get('runs',[{}])[0].get('text')), # grid(2)
 373             }})
 374         elif key == "showRenderer":
 375             result.append({'type': 'PLAYLIST', 'content': {
 376                 'playlist_id': content['navigationEndpoint']['watchEndpoint']['playlistId'],
 377                 'video_id': content['navigationEndpoint']['watchEndpoint']['videoId'],
 378                 'title': content['title']['simpleText'],
 379                 'author': author,
 380                 'channel_id': channel_id,
 381                 'n_videos': None,
 382             }})
 383         elif key in ["gridShowRenderer"]:
 384             result.append({'type': 'PLAYLIST', 'content': {
 385                 'playlist_id': (content|G('navigationEndpoint')
 386                     |G('browseEndpoint')|G('browseId'))[2:],
 387                 #^: playlistId prefixed with 'VL', which must be removed
 388                 'video_id': None,
 389                 'title': content|G('title')|G.text,
 390                 'author': author,
 391                 'channel_id': channel_id,
 392                 'n_videos': content|G('thumbnailOverlays')|G(0)
 393                     |G('thumbnailOverlayBottomPanelRenderer')|G('text')|G.text,
 394             }})
 395         elif key in ["itemSectionRenderer", "gridRenderer", "horizontalCardListRenderer", "horizontalListRenderer"]:
 396                 newkey = {
 397                     "itemSectionRenderer": 'contents',
 398                     "gridRenderer": 'items',
 399                     "horizontalCardListRenderer": 'cards',
 400                     "horizontalListRenderer": 'items',
 401                 }.get(key)
 402                 r, e = parse_channel_items(content[newkey], channel_id, author)
 403                 result.extend(r)
 404                 extra.extend(e)
 405         elif key == "shelfRenderer":
 406                 r, e = parse_channel_items([content['content']], channel_id, author)
 407                 result.extend(r)
 408                 extra.extend(e)
 409         elif key == "messageRenderer":
 410             # e.g. {'messageRenderer': {'text': {'runs': [{'text': 'This channel has no playlists.'}]}}}
 411             pass
 412         elif key == "gameCardRenderer":
 413             pass
 414         elif key == "gridChannelRenderer":
 415             pass # don't care; related channels, e.g. on UCMsgXPD3wzzt8RxHJmXH7hQ
 416         else:
 417             log_unknown_card(item)
 418
 419     return result, extra
 420
 421 def parse_playlist(item):
 422     key = next(iter(item.keys()), None)
 423     content = item[key]
 424     if key == "playlistVideoRenderer":
 425         if not content.get('isPlayable', False):
 426             return None # private or deleted video
 427
 428         return {'type': 'VIDEO', 'content': {
 429             'video_id': content['videoId'],
 430             'title': (content['title'].get('simpleText') or # playable videos
 431                 content['title'].get('runs',[{}])[0].get('text')), # "[Private video]"
 432             'playlist_id': content['navigationEndpoint']['watchEndpoint']['playlistId'],
 433             'index': content['navigationEndpoint']['watchEndpoint'].get('index',0), #or int(content['index']['simpleText']) (absent on course intros; e.g. PL96C35uN7xGJu6skU4TBYrIWxggkZBrF5)
 434             # rest is missing from unplayable videos:
 435             'author': content.get('shortBylineText',{}).get('runs',[{}])[0].get('text'),
 436             'channel_id':content.get('shortBylineText',{}).get('runs',[{}])[0].get('navigationEndpoint',{}).get('browseEndpoint',{}).get('browseId'),
 437             'length': (content.get("lengthText",{}).get("simpleText") or # "8:51"
 438                 int(content.get("lengthSeconds", 0))), # "531"
 439             'starttime': content['navigationEndpoint']['watchEndpoint'].get('startTimeSeconds'),
 440         }}
 441     else:
 442         raise Exception(item) # XXX TODO