Simplify caching and fetching of updated subscriptions for My Subscriptions #1395

- Subscription contentDetails cannot be solely relied upon to determine if subscriptions have been updated as feeds may not be available
- Use more naive approach that initially uses cached feed data even if YouTube indicates the subscription has been updated
- This allows cached data to be used if the feed update fails

Author: MoojMidge

resources/lib/youtube_plugin/youtube/client/data_client.py

@@ -2323,10 +2323,10 @@ def _sort_by_date_time(item, limits):
             return timestamp
 
         threaded_output = {
-            'channel_ids': [],
-            'playlist_ids': [],
+            'channel_ids': set(),
+            'playlist_ids': set(),
             'feeds': {},
-            'to_refresh': [],
+            'to_refresh': set(),
         }
 
         (use_subscriptions,
@@ -2337,21 +2337,25 @@ def _sort_by_date_time(item, limits):
             use_subscriptions = False
             use_saved_playlists = False
 
-        bookmarks = context.get_bookmarks_list().get_items()
-        if bookmarks:
-            channel_ids = threaded_output['channel_ids']
-            playlist_ids = threaded_output['playlist_ids']
-            for item_id, item in bookmarks.items():
-                if isinstance(item, DirectoryItem):
-                    item_id = getattr(item, PLAYLIST_ID, None)
-                    if item_id and use_bookmarked_playlists:
-                        playlist_ids.append(item_id)
-                        continue
-                    item_id = getattr(item, CHANNEL_ID, None)
-                elif not isinstance(item, float):
-                    continue
-                if item_id and use_bookmarked_channels:
-                    channel_ids.append(item_id)
+        if use_bookmarked_channels or use_bookmarked_playlists:
+            bookmarks = context.get_bookmarks_list().get_items()
+            if bookmarks:
+                channel_ids = threaded_output['channel_ids']
+                playlist_ids = threaded_output['playlist_ids']
+                for item_id, item in bookmarks.items():
+                    if isinstance(item, DirectoryItem):
+                        if use_bookmarked_playlists:
+                            item_id = getattr(item, PLAYLIST_ID, None)
+                            if item_id:
+                                playlist_ids.add(item_id)
+                                continue
+                        if use_bookmarked_channels:
+                            item_id = getattr(item, CHANNEL_ID, None)
+                            if item_id:
+                                channel_ids.add(item_id)
+                                continue
+                    elif use_bookmarked_channels and isinstance(item, float):
+                        channel_ids.add(item_id)
 
         headers = {
             'Host': 'www.youtube.com',
@@ -2378,6 +2382,7 @@ def _get_cached_feed(output,
                              ttl=feed_history.ONE_HOUR):
             feeds = output['feeds']
             to_refresh = output['to_refresh']
+
             if item_type == 'channel_id':
                 channel_prefix = (
                     'UUSH' if feed_type == 'shorts' else
@@ -2386,7 +2391,9 @@ def _get_cached_feed(output,
                 )
             else:
                 channel_prefix = False
-            for item_id in inputs:
+
+            batch = inputs.copy()
+            for item_id in batch:
                 if channel_prefix:
                     channel_id = item_id
                     item_id = item_id.replace('UC', channel_prefix, 1)
@@ -2396,44 +2403,46 @@ def _get_cached_feed(output,
                 cached = feed_history.get_item(item_id)
                 if cached:
                     feed_details = cached['value']
-                    _refresh = refresh or cached['age'] > ttl
-                    feed_details['refresh'] = _refresh
+
                     if channel_id:
                         feed_details.setdefault('channel_id', channel_id)
-                        if _refresh:
-                            to_refresh.append({item_type: channel_id})
-                    elif _refresh:
-                        to_refresh.append({item_type: item_id})
+
+                    _refresh = refresh or cached['age'] > ttl
+                    feed_details['refresh'] = _refresh
+                    if _refresh:
+                        to_refresh.add(channel_id)
+
                     if item_id in feeds:
                         feeds[item_id].update(feed_details)
                     else:
                         feeds[item_id] = feed_details
-                elif channel_id:
-                    to_refresh.append({item_type: channel_id})
                 else:
-                    to_refresh.append({item_type: item_id})
-            del inputs[:]
+                    to_refresh.add(channel_id)
+
+            inputs -= batch
             return True, False
 
         def _get_feed(output,
-                      channel_id=None,
-                      playlist_id=None,
+                      input=None,
                       feed_type=feed_type,
                       headers=headers,
                       stream=True,
                       cache=False):
-            if channel_id:
-                item_id = channel_id.replace(
+            if not input:
+                return True, False
+
+            if input.startswith('UC'):
+                channel_id = input
+                item_id = input.replace(
                     'UC',
                     'UUSH' if feed_type == 'shorts' else
                     'UULV' if feed_type == 'live' else
                     'UULF',
                     1,
                 )
-            elif playlist_id:
-                item_id = playlist_id
             else:
-                return True, False
+                channel_id = None
+                item_id = input
 
             response = self.request(
                 ''.join((self.BASE_URL,
@@ -2615,6 +2624,7 @@ def _parse_feeds(feeds,
 
         def _threaded_fetch(kwargs,
                             do_batch,
+                            unpack,
                             output,
                             worker,
                             threads,
@@ -2631,7 +2641,12 @@ def _threaded_fetch(kwargs,
                 if kwargs is True:
                     _kwargs = {}
                 elif kwargs:
-                    _kwargs = {'inputs': kwargs} if do_batch else kwargs.pop()
+                    if do_batch:
+                        _kwargs = {'inputs': kwargs}
+                    elif unpack:
+                        _kwargs = {'input': kwargs.pop()}
+                    else:
+                        _kwargs = kwargs.pop()
                 elif check_inputs:
                     if check_inputs.wait(0.1) and kwargs:
                         continue
@@ -2686,51 +2701,44 @@ def _threaded_fetch(kwargs,
             }
 
             def _get_updated_subscriptions(new_data, old_data):
-                items = new_data and new_data.get('items')
-                if not items:
+                new_items = new_data and new_data.get('items')
+                if not new_items:
                     new_data['_abort'] = True
                     return new_data
 
-                _items = old_data and old_data.get('items')
-                if _items:
-                    _items = {
+                old_items = old_data and old_data.get('items')
+                if old_items:
+                    old_items = {
                         item['snippet']['resourceId']['channelId']:
                             item['contentDetails']
-                        for item in _items
+                        for item in old_items
                     }
 
-                    updated_subscriptions = []
-                    old_subscriptions = []
-
-                    for item in items:
+                    old_subscriptions = False
+                    updated_subscriptions = set()
+                    for item in new_items:
                         channel_id = item['snippet']['resourceId']['channelId']
-                        counts = item['contentDetails']
-
-                        if (counts['newItemCount']
-                                or counts['totalItemCount']
-                                > _items.get(channel_id, {})['totalItemCount']):
-                            updated_subscriptions.append(
-                                {
-                                    'channel_id': channel_id,
-                                }
-                            )
+                        if channel_id in updated_subscriptions:
+                            continue
+
+                        item_counts = item['contentDetails']
+                        if (item_counts['newItemCount']
+                                or (channel_id in old_items
+                                    and item_counts['totalItemCount']
+                                    > old_items[channel_id]['totalItemCount'])):
+                            updated_subscriptions.add(channel_id)
                         else:
-                            old_subscriptions.append(channel_id)
+                            old_subscriptions = True
 
                     if old_subscriptions:
                         new_data['nextPageToken'] = None
                 else:
-                    updated_subscriptions = [
-                        {
-                            'channel_id':
-                                item['snippet']['resourceId']['channelId'],
-                        }
-                        for item in items
-                    ]
-                    old_subscriptions = []
+                    updated_subscriptions = {
+                        item['snippet']['resourceId']['channelId']
+                        for item in new_items
+                    }
 
                 new_data['_updated_subscriptions'] = updated_subscriptions
-                new_data['_old_subscriptions'] = old_subscriptions
                 return new_data
 
             def _get_channels(output,
@@ -2753,11 +2761,14 @@ def _get_channels(output,
 
                 updated_subscriptions = json_data.get('_updated_subscriptions')
                 if updated_subscriptions:
-                    output['to_refresh'].extend(updated_subscriptions)
+                    output['to_refresh'] |= updated_subscriptions
 
-                old_subscriptions = json_data.get('_old_subscriptions')
-                if old_subscriptions:
-                    output['channel_ids'].extend(old_subscriptions)
+                all_subscriptions = json_data.get('items')
+                if all_subscriptions:
+                    output['channel_ids'].update([
+                        item['snippet']['resourceId']['channelId']
+                        for item in all_subscriptions
+                    ])
 
                 page_token = json_data.get('nextPageToken')
                 if page_token:
@@ -2771,6 +2782,7 @@ def _get_channels(output,
                 'worker': _get_channels,
                 'kwargs': True,
                 'do_batch': False,
+                'unpack': False,
                 'output': threaded_output,
                 'threads': threads,
                 'limit': 1,
@@ -2812,10 +2824,12 @@ def _get_playlists(output,
                 if not json_data:
                     return False, True
 
-                output['playlist_ids'].extend([
-                    item['id']
-                    for item in json_data.get('items', [])
-                ])
+                saved_playlists = json_data.get('items')
+                if saved_playlists:
+                    output['playlist_ids'].update([
+                        item['id']
+                        for item in saved_playlists
+                    ])
 
                 subs_page_token = json_data.get('nextPageToken')
                 if subs_page_token:
@@ -2827,6 +2841,7 @@ def _get_playlists(output,
                 'worker': _get_playlists,
                 'kwargs': True,
                 'do_batch': False,
+                'unpack': False,
                 'output': threaded_output,
                 'threads': threads,
                 'limit': 1,
@@ -2838,6 +2853,7 @@ def _get_playlists(output,
             'worker': partial(_get_cached_feed, item_type='channel_id'),
             'kwargs': threaded_output['channel_ids'],
             'do_batch': True,
+            'unpack': False,
             'output': threaded_output,
             'threads': threads,
             'limit': None,
@@ -2849,6 +2865,7 @@ def _get_playlists(output,
             'worker': partial(_get_cached_feed, item_type='playlist_id'),
             'kwargs': threaded_output['playlist_ids'],
             'do_batch': True,
+            'unpack': False,
             'output': threaded_output,
             'threads': threads,
             'limit': None,
@@ -2860,6 +2877,7 @@ def _get_playlists(output,
             'worker': _get_feed,
             'kwargs': threaded_output['to_refresh'],
             'do_batch': False,
+            'unpack': True,
             'output': threaded_output,
             'threads': threads,
             'limit': None,