From c9d4a45e39d9848889c066b381085896b36f8736 Mon Sep 17 00:00:00 2001 From: Simon Date: Thu, 25 Dec 2025 12:08:03 +0700 Subject: [PATCH] implement channel artwork embed in video --- backend/download/src/thumbnails.py | 88 ++++++++++++++++---------- backend/download/src/yt_dlp_handler.py | 14 ++-- backend/video/src/index.py | 20 +++++- 3 files changed, 75 insertions(+), 47 deletions(-) diff --git a/backend/download/src/thumbnails.py b/backend/download/src/thumbnails.py index f9617973..f387f2da 100644 --- a/backend/download/src/thumbnails.py +++ b/backend/download/src/thumbnails.py @@ -4,9 +4,7 @@ functionality: - check for missing thumbnails """ -import base64 import os -from io import BytesIO from time import sleep import requests @@ -14,7 +12,7 @@ from common.src.env_settings import EnvironmentSettings from common.src.es_connect import ElasticWrap, IndexPaginate from common.src.helper import is_missing from mutagen.mp4 import MP4, MP4Cover -from PIL import Image, ImageFile, ImageFilter, UnidentifiedImageError +from PIL import Image, ImageFile, UnidentifiedImageError ImageFile.LOAD_TRUNCATED_IMAGES = True @@ -23,6 +21,7 @@ class ThumbManagerBase: """base class for thumbnail management""" CACHE_DIR = EnvironmentSettings.CACHE_DIR + MEDIA_DIR = EnvironmentSettings.MEDIA_DIR VIDEO_DIR = os.path.join(CACHE_DIR, "videos") CHANNEL_DIR = os.path.join(CACHE_DIR, "channels") PLAYLIST_DIR = os.path.join(CACHE_DIR, "playlists") @@ -217,6 +216,56 @@ class ThumbManager(ThumbManagerBase): img_raw = img_raw.resize((336, 189)) img_raw.convert("RGB").save(thumb_path) + def embed_video_art(self, json_data: dict): + """embed video artwork""" + file_path = os.path.join(self.MEDIA_DIR, json_data["media_url"]) + video = MP4(file_path) + + thumb_path = os.path.join(self.CACHE_DIR, self.vid_thumb_path()) + if os.path.exists(thumb_path): + with open(thumb_path, "rb") as f: + cover_data = f.read() + + video["covr"] = [ + MP4Cover(cover_data, imageformat=MP4Cover.FORMAT_JPEG) + ] + + channel_id = json_data["channel"]["channel_id"] + 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) + ] + + 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) + ] + + 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() + + video["----:com.tubearchivist:channel_tv"] = [ + MP4Cover(tv_data, imageformat=MP4Cover.FORMAT_JPEG) + ] + + video.save() + def delete_video_thumb(self): """delete video thumbnail if exists""" thumb_path = self.vid_thumb_path() @@ -242,20 +291,6 @@ class ThumbManager(ThumbManagerBase): if os.path.exists(thumb_path): os.remove(thumb_path) - def get_vid_base64_blur(self): - """return base64 encoded placeholder""" - file_path = os.path.join(self.CACHE_DIR, self.vid_thumb_path()) - img_raw = Image.open(file_path) - img_raw.thumbnail((img_raw.width // 20, img_raw.height // 20)) - img_blur = img_raw.filter(ImageFilter.BLUR) - buffer = BytesIO() - img_blur.save(buffer, format="JPEG") - img_data = buffer.getvalue() - img_base64 = base64.b64encode(img_data).decode() - data_url = f"data:image/jpg;base64,{img_base64}" - - return data_url - class ValidatorCallback: """handle callback validate thumbnails page by page""" @@ -484,9 +519,7 @@ class ThumbFilesystem: class EmbedCallback: """callback class to embed thumbnails""" - CACHE_DIR = EnvironmentSettings.CACHE_DIR MEDIA_DIR = EnvironmentSettings.MEDIA_DIR - FORMAT = MP4Cover.FORMAT_JPEG def __init__(self, source, index_name, counter=0): self.source = source @@ -497,19 +530,4 @@ class EmbedCallback: """run embed""" for video in self.source: video_id = video["_source"]["youtube_id"] - media_url = os.path.join( - self.MEDIA_DIR, video["_source"]["media_url"] - ) - thumb_path = os.path.join( - self.CACHE_DIR, ThumbManager(video_id).vid_thumb_path() - ) - if os.path.exists(thumb_path): - self.embed(media_url, thumb_path) - - def embed(self, media_url, thumb_path): - """embed thumb in single media file""" - video = MP4(media_url) - with open(thumb_path, "rb") as f: - video["covr"] = [MP4Cover(f.read(), imageformat=self.FORMAT)] - - video.save() + ThumbManager(video_id).embed_video_art(video["_source"]) diff --git a/backend/download/src/yt_dlp_handler.py b/backend/download/src/yt_dlp_handler.py index ad95abe4..08017a3c 100644 --- a/backend/download/src/yt_dlp_handler.py +++ b/backend/download/src/yt_dlp_handler.py @@ -196,15 +196,6 @@ class VideoDownloader(DownloaderBase): } ) - if self.config["downloads"]["add_thumbnail"]: - postprocessors.append( - { - "key": "EmbedThumbnail", - "already_have_thumbnail": True, - } - ) - self.obs["writethumbnail"] = True - self.obs["postprocessors"] = postprocessors def _set_overwrites(self, obs: dict, channel_id: str) -> None: @@ -491,7 +482,10 @@ class DownloadPostProcess(DownloaderBase): def embed_metadata(self): """embed metadata in media file""" - if not self.config["downloads"].get("add_metadata"): + meta = self.config["downloads"].get("add_metadata") + thumb = self.config["downloads"].get("add_thumbnail") + + if not meta and not thumb: return queue = RedisQueue(self.VIDEO_QUEUE) diff --git a/backend/video/src/index.py b/backend/video/src/index.py index 1d5cf839..11a5775f 100644 --- a/backend/video/src/index.py +++ b/backend/video/src/index.py @@ -424,9 +424,13 @@ class YoutubeVideo(YouTubeItem, YoutubeSubtitle): def embed_metadata(self): """embed metadata for video""" - if not self.config["downloads"].get("add_metadata"): - return + if self.config["downloads"].get("add_metadata"): + self._embed_text_data() + if self.config["downloads"].get("add_thumbnail"): + self._embed_artwork() + + def _embed_text_data(self): print(f"{self.youtube_id}: embed metadata") if not self.json_data: self.get_from_es() @@ -484,6 +488,18 @@ class YoutubeVideo(YouTubeItem, YoutubeSubtitle): return to_embed + def _embed_artwork(self): + """embed artwork""" + print(f"{self.youtube_id}: embed artwork") + if not self.json_data: + self.get_from_es() + + if not self.json_data: + print(f"{self.youtube_id}: skip embed, video not indexed") + return + + ThumbManager(self.youtube_id).embed_video_art(self.json_data) + def index_new_video(youtube_id, video_type=VideoTypeEnum.VIDEOS): """combined classes to create new video in index"""