feat: track published and processed media keys

tubesync/sync/models/source.py

@@ -1,16 +1,20 @@
 import os
 import re
 import uuid
+
 from collections import deque as queue
 from pathlib import Path
 from typing import ClassVar
+
 from django import db
 from django.conf import settings
 from django.core.validators import RegexValidator
 from django.utils import timezone
 from django.utils.text import slugify
 from django.utils.translation import gettext_lazy as _
+
 from yt_dlp import DownloadError
+
 from ..choices import (Val,
     AudioTrack, SponsorBlock_Category, YouTube_SourceType, IndexSchedule,
     CapChoices, Fallback, FileExtension, FilterSeconds,
@@ -24,14 +28,11 @@
     get_image_urls as get_youtube_image_urls,
     merge_image_thumbnails,
 )
+
 from ._migrations import media_file_storage
 from ._private import _srctype_dict
 
 
-def _has_index_metadata_value(value):
-    return bool(value) or value in (0, False,)
-
-
 class Source(db.models.Model):
     '''
         A Source is a source of media. Currently, this is either a YouTube channel
@@ -619,6 +620,10 @@ def get_index(self, url_type, /):
             if not isinstance(response, dict):
                 return entries
             entries = response.pop('entries', list())
+
+            def _has_index_metadata_value(value):
+                return bool(value) or value in (0, False,)
+
             default_site = self.videos.model._meta.get_field('site').get_default()
             site = response.get('extractor_key') or default_site
             metadata, _ = self.videos.filter(
@@ -633,13 +638,15 @@ def get_index(self, url_type, /):
                 previous.get('thumbnails'),
                 response.get('thumbnails'),
             )
+
             for key, value in previous.items():
                 if 'thumbnails' == key:
                     continue
-                if (
+                write_to_key = (
                     _has_index_metadata_value(value) and
                     not _has_index_metadata_value(response.get(key))
-                ):
+                )
+                if write_to_key:
                     response[key] = value
             update_fields = {'retrieved', 'value'}
             if site and site not in (metadata.site, default_site):
@@ -651,27 +658,35 @@ def get_index(self, url_type, /):
         return entries
 
     def index_media(self):
-        '''
-            Index the media source returning a queue of media metadata as dicts.
-        '''
-        entries = queue(list(), getattr(settings, 'MAX_ENTRIES_PROCESSING', 0) or None)
+        """
+        Index the media source returning a queue of media metadata as dicts and a set of media keys.
+        """
+        entries = queue(maxlen=getattr(settings, 'MAX_ENTRIES_PROCESSING', 0) or None)
+        seen_keys = set()
+        def register_key(item):
+            key = item.get(self.key_field, None)
+            if key is not None:
+                seen_keys.add(key)
+            return item
+
         if self.index_videos:
             videos = self.get_index('videos')
-            entries.extend(reversed(videos))
+            entries.extend(map(register_key, reversed(videos)))
+            del videos
 
         # Playlists do something different that I have yet to figure out
-        # ruff: ignore[SIM102]
-        if not self.is_playlist:
-            if self.index_streams:
-                streams = self.get_index('streams')
-                if entries.maxlen is None or 0 == len(entries):
-                    entries.extend(reversed(streams))
-                else:
-                    # share the queue between streams and videos
-                    allowed_streams = max(
-                        entries.maxlen // 2,
-                        entries.maxlen - len(entries),
-                    )
-                    entries.extend(reversed(streams[: allowed_streams]))
-
-        return entries
+        if not self.is_playlist and self.index_streams:
+            streams = self.get_index('streams')
+            if entries.maxlen is None or 0 == len(entries):
+                entries.extend(map(register_key, reversed(streams)))
+            else:
+                # share the queue between streams and videos
+                allowed_streams = max(
+                    entries.maxlen // 2,
+                    entries.maxlen - len(entries),
+                )
+                streams = tuple(map(register_key, streams))[: allowed_streams]
+                entries.extend(reversed(streams))
+            del streams
+
+        return entries, seen_keys

tubesync/sync/tasks.py

@@ -553,7 +553,7 @@ def index_source(source_id):
     # Reset any errors
     source.has_failed = False
     # Index the source
-    videos = source.index_media()
+    videos, published_keys = source.index_media()
     if not videos:
         source.has_failed = True
         update_model(source, has_failed=source.has_failed)
@@ -593,7 +593,7 @@ def index_source(source_id):
             end=task.verbose_name.find('Index'),
         )
     vn = 0
-    video_keys = set()
+    processed_keys = set()
     while len(videos) > 0:
         vn += 1
         video = videos.popleft()
@@ -602,7 +602,7 @@ def index_source(source_id):
         if not key:
             # Video has no unique key (ID), it can't be indexed
             continue
-        video_keys.add(key)
+        processed_keys.add(key)
         if len(db_batch_data) == db_batch_data.maxlen:
             save_db_batch(Metadata.objects, db_batch_data, db_fields_data)
         if len(db_batch_media) == db_batch_media.maxlen:
@@ -685,7 +685,10 @@ def index_source(source_id):
     save_db_batch(Metadata.objects, db_batch_data, db_fields_data)
     save_db_batch(Media.objects, db_batch_media, db_fields_media)
     # Cleanup of media no longer available from the source
-    cleanup_removed_media(str(source.pk), video_keys)
+    use_published = 'published' == str(
+        getattr(settings, 'INDEXING_CLEANUP_UNAVAILABLE_POLICY', 'processed')
+    ).strip().lower()
+    cleanup_removed_media(str(source.pk), published_keys if use_published else processed_keys)
     # Clear references to indexed data
     videos = video = None
     db_batch_data.clear()

tubesync/sync/tests/test_index_coverage.py

@@ -0,0 +1,22 @@
+"""The processing cap must not make upstream media look removed."""
+from unittest.mock import patch
+
+from django.test import SimpleTestCase, override_settings
+
+from sync.models import Source
+
+
+@override_settings(MAX_ENTRIES_PROCESSING=2, INDEXING_CLEANUP_UNAVAILABLE_POLICY='Published ')
+class IndexCoverageTests(SimpleTestCase):
+    def test_capped_queue_retains_complete_removal_keys(self):
+        videos = [{"id": key} for key in ("video-a", "video-b", "video-c")]
+        streams = [{"id": key} for key in ("stream-a", "stream-b", "stream-c")]
+        for mixed in (False, True):
+            with self.subTest(mixed=mixed):
+                source = Source(index_videos=True, index_streams=mixed)
+                with patch.object(source, "get_index", side_effect=lambda kind: videos if kind == "videos" else streams):
+                    processing, upstream = source.index_media()
+                self.assertEqual(len(processing), 2)
+                expected = {row["id"] for row in videos + (streams if mixed else [])}
+                self.assertEqual(upstream, expected)
+                self.assertEqual((expected | {"genuinely-removed"}) - upstream, {"genuinely-removed"})