implement playlist restore from embed

backend/download/src/thumbnails.py

@@ -234,38 +234,41 @@ class ThumbManager(ThumbManagerBase):
         banner_path = os.path.join(
             self.CHANNEL_DIR, f"{channel_id}_banner.jpg"
         )
-        if os.path.exists(banner_path):
-            with open(banner_path, "rb") as f:
-                banner_data = f.read()
-
-            video["----:com.tubearchivist:channel_banner"] = [
-                MP4Cover(banner_data, imageformat=MP4Cover.FORMAT_JPEG)
-            ]
+        self._embed_art_item(video, "channel_banner", art_path=banner_path)
 
         channel_icon_path = os.path.join(
             self.CHANNEL_DIR, f"{channel_id}_thumb.jpg"
         )
-        if os.path.exists(channel_icon_path):
-            with open(channel_icon_path, "rb") as f:
-                icon_data = f.read()
-
-            video["----:com.tubearchivist:channel_icon"] = [
-                MP4Cover(icon_data, imageformat=MP4Cover.FORMAT_JPEG)
-            ]
+        self._embed_art_item(video, "channel_icon", art_path=channel_icon_path)
 
         channel_tv_path = os.path.join(
             self.CHANNEL_DIR, f"{channel_id}_tvart.jpg"
         )
-        if os.path.exists(channel_tv_path):
-            with open(channel_tv_path, "rb") as f:
-                tv_data = f.read()
+        self._embed_art_item(video, "channel_tv", art_path=channel_tv_path)
 
-            video["----:com.tubearchivist:channel_tv"] = [
-                MP4Cover(tv_data, imageformat=MP4Cover.FORMAT_JPEG)
-            ]
+        playlist_ids = json_data.get("playlist", [])
+        for plalyist_id in playlist_ids:
+            playlist_path = os.path.join(
+                self.PLAYLIST_DIR, f"{plalyist_id}.jpg"
+            )
+            self._embed_art_item(
+                video, f"playlist_{plalyist_id}", art_path=playlist_path
+            )
 
         video.save()
 
+    def _embed_art_item(self, video, key, art_path):
+        """embed single item"""
+        if not os.path.exists(art_path):
+            return
+
+        with open(art_path, "rb") as f:
+            art_data = f.read()
+
+        video[f"----:com.tubearchivist:{key}"] = [
+            MP4Cover(art_data, imageformat=MP4Cover.FORMAT_JPEG)
+        ]
+
     def delete_video_thumb(self):
         """delete video thumbnail if exists"""
         thumb_path = self.vid_thumb_path()

backend/playlist/serializers.py

@@ -11,7 +11,7 @@ class PlaylistEntrySerializer(serializers.Serializer):
 
     youtube_id = serializers.CharField()
     title = serializers.CharField()
-    uploader = serializers.CharField()
+    uploader = serializers.CharField(allow_null=True)
     idx = serializers.IntegerField()
     downloaded = serializers.BooleanField()
 
@@ -22,7 +22,7 @@ class PlaylistSerializer(serializers.Serializer):
     playlist_active = serializers.BooleanField()
     playlist_channel = serializers.CharField()
     playlist_channel_id = serializers.CharField()
-    playlist_description = serializers.CharField()
+    playlist_description = serializers.CharField(allow_null=True)
     playlist_entries = PlaylistEntrySerializer(many=True)
     playlist_id = serializers.CharField()
     playlist_last_refresh = serializers.CharField()

backend/video/src/meta_embed.py

@@ -15,6 +15,8 @@ from common.src.env_settings import EnvironmentSettings
 from common.src.es_connect import ElasticWrap, IndexPaginate
 from download.src.thumbnails import ThumbManager
 from mutagen.mp4 import MP4, MP4FreeForm
+from playlist.serializers import PlaylistSerializer
+from playlist.src.index import YoutubePlaylist
 from video.serializers import (
     CommentsSerializer,
     SubtitleFragmentSerializer,
@@ -105,6 +107,7 @@ class IndexFromEmbed:
         self.index_subtitles(json_embed, video)
         self.index_comments(json_embed)
         self.restore_artwork(video)
+        self.index_playlists(json_embed, video)
 
         return video.json_data
 
@@ -208,6 +211,64 @@ class IndexFromEmbed:
         if self.HOST_UID and self.HOST_GID:
             os.chown(new_path, self.HOST_UID, self.HOST_GID)
 
+    def index_playlists(self, json_embed, video):
+        """index playlists"""
+        playlist_data = json_embed.get("playlists")
+        if not playlist_data or not isinstance(playlist_data, list):
+            return
+
+        serializer = PlaylistSerializer(data=playlist_data, many=True)
+        is_valid = serializer.is_valid()
+        if not is_valid:
+            err = serializer.errors
+            raise ValueError(
+                f"[{self.file_path}] playlist serializer failed: {err}"
+            )
+
+        expected = video.json_data.get("playlist", [])
+        video_mutagen = MP4(self.file_path)
+
+        for playlist_data in serializer.data:
+            playlist_id = playlist_data["playlist_id"]
+            if playlist_id not in expected:
+                continue
+
+            json_data = self._process_embedded_playlist(playlist_data)
+            if not json_data:
+                continue
+
+            playlist_art = os.path.join(
+                self.CACHE_DIR, "playlists", f"{playlist_id}.jpg"
+            )
+            self._restore_art_item(
+                video_mutagen,
+                "----:com.tubearchivist:playlist_{plalyist_id}",
+                playlist_art,
+            )
+
+    def _process_embedded_playlist(self, playlist_data) -> dict | None:
+        """process single embedded playlist"""
+        if not self.use_user_conf:
+            if playlist_data["playlist_type"] == "custom":
+                # custom playlist is user conf
+                return None
+
+        playlist = YoutubePlaylist(youtube_id=playlist_data["playlist_id"])
+        playlist.get_from_es()
+        if playlist.json_data:
+            # already indexed
+            return None
+
+        playlist_data_clean = dict(playlist_data)
+        if not self.use_user_conf:
+            playlist_data_clean["playlist_subscribed"] = False
+            playlist_data_clean["playlist_sort_order"] = "top"
+
+        playlist.json_data = playlist_data_clean
+        playlist.upload_to_es()
+
+        return playlist.json_data
+
     def restore_artwork(self, video):
         """restore artwork if needed"""
         video_mutagen = MP4(self.file_path)