diff --git a/tubesync/sync/models/source.py b/tubesync/sync/models/source.py index 3dd7f1bbb..6301377e0 100644 --- a/tubesync/sync/models/source.py +++ b/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 diff --git a/tubesync/sync/tasks.py b/tubesync/sync/tasks.py index 6f171d203..5721362e4 100644 --- a/tubesync/sync/tasks.py +++ b/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() diff --git a/tubesync/sync/tests/test_index_coverage.py b/tubesync/sync/tests/test_index_coverage.py new file mode 100644 index 000000000..b3d352077 --- /dev/null +++ b/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"})