From 7addd5057782dfd71c0f0bd0a83949db3204106d Mon Sep 17 00:00:00 2001 From: Colin Powell Date: Mon, 17 Nov 2025 17:56:15 -0500 Subject: [PATCH] [videos] Refactor lookup to use new library --- vrobbler/apps/scrobbles/scrobblers.py | 13 ++-- vrobbler/apps/videos/metadata.py | 5 +- vrobbler/apps/videos/models.py | 69 +++++++++++++++---- vrobbler/apps/videos/sources/imdb.py | 96 +++++++-------------------- vrobbler/apps/videos/utils.py | 2 +- 5 files changed, 89 insertions(+), 96 deletions(-) diff --git a/vrobbler/apps/scrobbles/scrobblers.py b/vrobbler/apps/scrobbles/scrobblers.py index 41fa26c..7bc7a73 100644 --- a/vrobbler/apps/scrobbles/scrobblers.py +++ b/vrobbler/apps/scrobbles/scrobblers.py @@ -123,9 +123,8 @@ def jellyfin_scrobble_media( / 10000000 ) if media_type == Scrobble.MediaType.VIDEO: - media_obj = Video.get_from_imdb_id( - post_data.get("Provider_imdb", "").replace("tt", "") - ) + imdb_id = post_data.get("Provider_imdb", "") + media_obj = Video.find_or_create(imdb_id) else: media_obj = Track.find_or_create( title=post_data.get("Name", ""), @@ -162,18 +161,14 @@ def jellyfin_scrobble_media( def web_scrobbler_scrobble_media( youtube_id: str, user_id: int, status: str = "started" ) -> Optional[Scrobble]: - video = Video.get_from_youtube_id(youtube_id) + video = Video.find_or_create(youtube_id) return video.scrobble_for_user(user_id, status, source="Web Scrobbler") def manual_scrobble_video( video_id: str, user_id: int, source: str = "IMDb", action: Optional[str] = None ): - if "tt" in video_id: - video = Video.get_from_imdb_id(video_id) - - else: - video = Video.get_from_youtube_id(video_id) + video = Video.find_or_create(video_id) # When manually scrobbling, try finding a source from the series if video.tv_series: diff --git a/vrobbler/apps/videos/metadata.py b/vrobbler/apps/videos/metadata.py index 97b5f7c..91992af 100644 --- a/vrobbler/apps/videos/metadata.py +++ b/vrobbler/apps/videos/metadata.py @@ -59,8 +59,11 @@ class VideoMetadata: def as_dict_with_cover_and_genres(self) -> tuple: video_dict = vars(self) + series_id = "" cover = None if "cover_url" in video_dict.keys(): cover = video_dict.pop("cover_url", "") genres = video_dict.pop("genres", []) - return video_dict, cover, genres + if "tv_series_imdb_id" in video_dict.keys(): + series_id = video_dict.pop("tv_series_imdb_id") + return video_dict, series_id, cover, genres diff --git a/vrobbler/apps/videos/models.py b/vrobbler/apps/videos/models.py index 69d1156..4aadd69 100644 --- a/vrobbler/apps/videos/models.py +++ b/vrobbler/apps/videos/models.py @@ -1,4 +1,5 @@ from dataclasses import dataclass +import re import logging from typing import Optional from uuid import uuid4 @@ -27,6 +28,8 @@ from vrobbler.apps.scrobbles.dataclasses import BaseLogData, WithPeopleLogData YOUTUBE_VIDEO_URL = "https://www.youtube.com/watch?v=" YOUTUBE_CHANNEL_URL = "https://www.youtube.com/channel/" +YOUTUBE_ID_PATTERN = re.compile(r'^[A-Za-z0-9_-]{11}$') + IMDB_VIDEO_URL = "https://www.imdb.com/title/tt" logger = logging.getLogger(__name__) @@ -136,6 +139,13 @@ class Series(TimeStampedModel): url = self.cover_image_medium.url return url + def save_image_from_url(self, url: str, force_update: bool = False): + if not self.cover_image or (force_update and url): + r = requests.get(url) + if r.status_code == 200: + fname = f"{self.title}_{self.uuid}.jpg" + self.cover_image.save(fname, ContentFile(r.content), save=True) + def scrobbles_for_user(self, user_id: int, include_playing=False): from scrobbles.models import Scrobble @@ -184,6 +194,31 @@ class Series(TimeStampedModel): if genres := imdb_dict.get("genres"): self.genre.add(*genres) + @classmethod + def find_or_create(cls, imdb_id: str, overwrite: bool = True): + series, created = cls.objects.get_or_create(imdb_id=imdb_id) + + if not (created or overwrite): + return series + + vdict, _, cover, genres = lookup_video_from_imdb( + imdb_id + ).as_dict_with_cover_and_genres() + vdict.pop("video_type") + + vdict["name"] = vdict.pop("title") + for k, v in vdict.items(): + setattr(series, k, v) + series.save() + + if cover: + series.save_image_from_url(cover) + if genres: + series.genre.add(*genres) + + return series + + class Video(ScrobblableMixin): COMPLETION_PERCENT = getattr(settings, "VIDEO_COMPLETION_PERCENT", 90) @@ -304,7 +339,7 @@ class Video(ScrobblableMixin): if not created and not overwrite: return video - vdict, cover, genres = lookup_video_from_youtube( + vdict, _, cover, genres = lookup_video_from_youtube( youtube_id ).as_dict_with_cover_and_genres() if created or overwrite: @@ -320,28 +355,38 @@ class Video(ScrobblableMixin): def get_from_imdb_id( cls, imdb_id: str, overwrite: bool = False ) -> "Video": - if "tt" in imdb_id: - imdb_id = imdb_id[2:] video, created = cls.objects.get_or_create(imdb_id=imdb_id) if not created and not overwrite: return video - vdict, cover, genres = lookup_video_from_tmdb( + vdict, series_id, cover, genres = lookup_video_from_imdb( imdb_id ).as_dict_with_cover_and_genres() + if created or overwrite: for k, v in vdict.items(): setattr(video, k, v) + + if series_id: + video.tv_series = Series.find_or_create(imdb_id=series_id) + video.save() - video.save_image_from_url(cover) - video.genre.add(*genres) + if cover: + video.save_image_from_url(cover) + if genres: + video.genre.add(*genres) + return video @classmethod - def find_or_create( - cls, data_dict: dict, post_keys: dict = JELLYFIN_POST_KEYS - ) -> Optional["Video"]: - """Thes smallest of wrappers around our actual get or create utility.""" - imdb_key = post_keys.get("IMDB_ID", "").replace("tt", "") - return cls.get_from_imdb_id(data_dict.get(imdb_key)) + def find_or_create(cls, source_id: str, overwrite: bool = True) -> "Video": + if "tt" in source_id: + return cls.get_from_imdb_id(source_id) + if bool(YOUTUBE_ID_PATTERN.match(source_id)): + return cls.get_from_youtube_id(source_id) + + #TODO scrobble but without a video obj? + logger.warning("Video ID not recognized, not scrobbling") + + return diff --git a/vrobbler/apps/videos/sources/imdb.py b/vrobbler/apps/videos/sources/imdb.py index aaeb25a..cca5d2c 100644 --- a/vrobbler/apps/videos/sources/imdb.py +++ b/vrobbler/apps/videos/sources/imdb.py @@ -6,55 +6,17 @@ from videos.metadata import VideoMetadata, VideoType logger = logging.getLogger(__name__) -def lookup_video_from_imdb( - name_or_id: str, kind: str = "movie" -) -> VideoMetadata: - from videos.models import Series - - # Very few video titles start with tt, but IMDB IDs often come in with it - if name_or_id.startswith("tt"): - name_or_id = name_or_id[2:] - - imdb_id = None - - try: - imdb_id = int(name_or_id) - except ValueError: - pass +def lookup_video_from_imdb(imdb_id: str) -> VideoMetadata: + if not imdb_id.startswith("tt"): + logger.warning("This method requires an IMDB ID starting with 'tt'") + return VideoMetadata() video_metadata = VideoMetadata(imdb_id=imdb_id) imdb_result: dict = {} - imdb_result = imdb.get_title(name_or_id) + imdb_result = imdb.get_title(imdb_id) logger.debug(f"Found result from IMDB: {imdb_result.title}") - if not imdb_result: - imdb_result = {} - imdb_results: list = imdb.search_movie(name_or_id) - if len(imdb_results) > 1: - for result in imdb_results: - if result["kind"] == kind: - imdb_client.update( - result, - info=[ - "plot", - "synopsis", - "taglines", - "next_episode", - "genres", - ], - ) - imdb_result = result - break - - if len(imdb_results) == 1: - imdb_result = imdb_results[0] - - imdb_client.update( - imdb_result, - info=["plot", "synopsis", "taglines", "next_episode", "genres"], - ) - if not imdb_result: logger.info( f"[lookup_video_from_imdb] no video found on imdb", @@ -62,39 +24,27 @@ def lookup_video_from_imdb( ) return None - video_metadata.cover_url = imdb_result.get("cover url") - if video_metadata.cover_url: - video_metadata.cover_url = helpers.resizeImage( - video_metadata.cover_url, width=800 - ) + video_metadata.imdb_id = imdb_id + video_metadata.title = imdb_result.title + video_metadata.base_run_time_seconds = ( + imdb_result.runtime * 60 + ) + video_metadata.year = imdb_result.year + video_metadata.plot = imdb_result.plot.get("en-US", "") + video_metadata.imdb_rating = imdb_result.rating + video_metadata.genres = imdb_result.genres + video_metadata.cover_url = imdb_result.primary_image video_metadata.video_type = VideoType.MOVIE.value - series_name = None - if imdb_result.get("kind") == "episode": - try: - series_name = imdb_result.get("episode of", None).data.get( - "title", None - ) - except IndexError: - series_name = None - if series_name: - series, _ = Series.objects.get_or_create(name=series_name) - video_metadata.video_type = VideoType.TV_EPISODE.value - video_metadata.tv_series_id = series.id + if imdb_result.type_id == "tvEpisode": + video_metadata.video_type = VideoType.TV_EPISODE.value - if imdb_result.get("runtimes"): - video_metadata.base_run_time_seconds = ( - int(imdb_result.get("runtimes")[0]) * 60 - ) + series = imdb_result.series + video_metadata.tv_series_imdb_id = series.imdb_id + video_metadata.tv_series_title = series.title + video_metadata.episode_number = imdb_result.episode + video_metadata.season_number = imdb_result.season + video_metadata.next_imdb_id = imdb_result.next_episode_id - video_metadata.imdb_id = name_or_id - video_metadata.title = imdb_result.get("title", "") - video_metadata.episode_number = imdb_result.get("episode", None) - video_metadata.season_number = imdb_result.get("season", None) - video_metadata.next_imdb_id = imdb_result.get("next episode", None) - video_metadata.year = imdb_result.get("year", None) - video_metadata.plot = imdb_result.get("plot outline", "") - video_metadata.imdb_rating = imdb_result.get("rating", None) - video_metadata.genres = imdb_result.get("genres", []) return video_metadata diff --git a/vrobbler/apps/videos/utils.py b/vrobbler/apps/videos/utils.py index 9624dc4..d39a0a2 100644 --- a/vrobbler/apps/videos/utils.py +++ b/vrobbler/apps/videos/utils.py @@ -29,7 +29,7 @@ def get_or_create_video(data_dict: dict, post_keys: dict, force_update=False): ) return - video = Video.get_from_imdb_id(video_dict.get("imdb_id") + video = Video.find_or_create(video_dict.get("imdb_id")) if not "overview" in video_dict.keys(): video_dict["overview"] = data_dict.get(