Merge pull request #1230 from tcely/patch-10

Replace `get_channel_image_info` with `get_image_info`

tubesync/sync/models/source.py

@@ -5,7 +5,6 @@ from collections import deque as queue
 from pathlib import Path
 from django import db
 from django.conf import settings
-from django.core.exceptions import SuspiciousOperation
 from django.core.validators import RegexValidator
 from django.utils import timezone
 from django.utils.text import slugify
@@ -20,7 +19,7 @@ from ..choices import (Val,
 from ..fields import CommaSepChoiceField
 from ..youtube import (
     get_media_info as get_youtube_media_info,
-    get_channel_image_info as get_youtube_channel_image_info,
+    get_image_info as get_youtube_image_info,
 )
 from ._migrations import media_file_storage
 from ._private import _srctype_dict
@@ -484,10 +483,7 @@ class Source(db.models.Model):
 
     @property
     def get_image_url(self):
-        if self.is_playlist:
-            raise SuspiciousOperation('This source is a playlist so it doesn\'t have thumbnail.')
-
-        return get_youtube_channel_image_info(self.url)
+        return get_youtube_image_info(self.url)
 
 
     def directory_exists(self):

tubesync/sync/signals.py

@@ -36,7 +36,7 @@ def source_pre_save(sender, instance, **kwargs):
     check_source_directory_exists.call_local(*args)
     existing_copy_channel_images = existing_source.copy_channel_images
     new_copy_channel_images = instance.copy_channel_images
-    if new_copy_channel_images and not (existing_copy_channel_images or instance.is_playlist):
+    if new_copy_channel_images and not existing_copy_channel_images:
         download_source_images(str(instance.pk))
     existing_dirpath = existing_source.directory_path.resolve(strict=True)
     new_dirpath = instance.directory_path.resolve(strict=False)
@@ -110,7 +110,7 @@ def source_post_save(sender, instance, created, **kwargs):
     # Check directory exists and create an indexing task for newly created sources
     if created:
         check_source_directory_exists(str(source.pk))
-        if source.copy_channel_images and not source.is_playlist:
+        if source.copy_channel_images:
             download_source_images(str(source.pk))
         if source.is_active:
             log.info(f'Scheduling first media indexing for source: {source.name}')

tubesync/sync/tasks.py

@@ -648,10 +648,27 @@ def download_source_images(source_id):
         log.error(f'Task download_source_images(pk={source_id}) called but no '
                   f'source exists with ID: {source_id}')
         raise CancelExecution(_('no such source'), retry=False) from e
-    avatar, banner = source.get_image_url
+    avatar, banner, thumbnail = source.get_image_url
     log.info(f'Thumbnail URL for source with ID: {source_id} / {source} '
         f'Avatar: {avatar} '
-        f'Banner: {banner}')
+        f'Banner: {banner} '
+        f'Thumbnail: {thumbnail}')
+    if thumbnail is not None:
+        url = thumbnail
+        i = get_remote_image(url)
+        image_file = BytesIO()
+        i.save(image_file, 'JPEG', quality=85, optimize=True, progressive=True)
+
+        for file_name in ["thumbnail.jpg",]:
+            # Reset file pointer to the beginning for the next save
+            image_file.seek(0)
+            # Create a Django ContentFile from BytesIO stream
+            django_file = ContentFile(image_file.read())
+            file_path = source.directory_path / file_name
+            with open(file_path, 'wb') as f:
+                f.write(django_file.read())
+        i = image_file = None
+
     if banner is not None:
         url = banner
         i = get_remote_image(url)

tubesync/sync/youtube.py

@@ -81,7 +81,10 @@ def get_channel_id(url):
             else:
                 return channel_id
 
-def get_channel_image_info(url):
+def get_image_info(url):
+    avatar_url = None
+    banner_url = None
+    thumbnail_url = None
     opts = get_yt_opts()
     opts.update({
         'skip_download': True,
@@ -94,20 +97,53 @@ def get_channel_image_info(url):
     with yt_dlp.YoutubeDL(opts) as y:
         try:
             response = y.extract_info(url, download=False)
-
-            avatar_url = None
-            banner_url = None
+        except yt_dlp.utils.DownloadError as e:
+            raise YouTubeError(f'Failed to extract info for "{url}": {e}') from e
+        else:
+            max_height = 0
             for thumbnail in response['thumbnails']:
+                thumbnail_height = thumbnail.get('height')
+                try:
+                    thumbnail_height = int(thumbnail_height)
+                except (TypeError, ValueError,):
+                    thumbnail_height = int()
                 if thumbnail['id'] == 'avatar_uncropped':
                     avatar_url = thumbnail['url']
-                if thumbnail['id'] == 'banner_uncropped':
+                elif thumbnail['id'] == 'banner_uncropped':
                     banner_url = thumbnail['url']
-                if banner_url is not None and avatar_url is not None:
-                    break
-
-            return avatar_url, banner_url
-        except yt_dlp.utils.DownloadError as e:
-            raise YouTubeError(f'Failed to extract channel info for "{url}": {e}') from e
+                elif thumbnail_height > max_height:
+                    max_height = thumbnail_height
+                    thumbnail_url = thumbnail['url']
+            try:
+                entry_type = response['entries'][0].get('_type')
+            except IndexError:
+                # an empty entries list
+                pass
+            else:
+                if 'url' == entry_type:
+                    del response['entries']
+                elif 'playlist' == entry_type:
+                    for playlist in response['entries']:
+                        del playlist['entries']
+            from .models import Metadata
+            t = Metadata.objects.defer('value').filter(
+                source__isnull=True,
+                media__isnull=True,
+            ).get_or_create(
+                key=response['id'],
+                site=response['extractor_key'],
+            )
+            md = t[0]
+            field_defaults = {
+                f.attname: f.get_default()
+                for f in md._meta.fields
+                if f.has_default()
+            }
+            if 'retrieved' in field_defaults:
+                md.retrieved = field_defaults['retrieved']
+            md.value = response
+            md.save()
+    return avatar_url, banner_url, thumbnail_url
 
 
 def _subscriber_only(msg='', response=None):