diff --git a/tests/scrobbles_tests/test_scrobblers.py b/tests/scrobbles_tests/test_scrobblers.py index 2d742ed..0b67c8c 100644 --- a/tests/scrobbles_tests/test_scrobblers.py +++ b/tests/scrobbles_tests/test_scrobblers.py @@ -3,6 +3,24 @@ from unittest.mock import MagicMock, patch from scrobbles.scrobblers import jellyfin_scrobble_media, mopidy_scrobble_media +def test_jellyfin_scrobble_video_with_no_imdb_id(): + with patch("scrobbles.scrobblers.Video") as mock_video_class: + mock_video_class.find_or_create.return_value = None + + post_data = { + "ItemType": "Video", + "Name": "Test Video", + "Provider_imdb": "", + "PlaybackPosition": "00:05:00", + "NotificationType": "PlaybackProgress", + "UtcTimestamp": "2024-01-15T10:30:00Z", + } + + result = jellyfin_scrobble_media(post_data, 1) + + mock_video_class.find_or_create.assert_called_once_with(None) + + def test_jellyfin_scrobble_media_ignores_progress_with_zero_position(): post_data = { diff --git a/tests/videos_tests/test_video_find_or_create.py b/tests/videos_tests/test_video_find_or_create.py new file mode 100644 index 0000000..0361529 --- /dev/null +++ b/tests/videos_tests/test_video_find_or_create.py @@ -0,0 +1,17 @@ +import pytest +from videos.models import Video + + +@pytest.mark.django_db +class TestVideoFindOrCreate: + def test_find_or_create_with_none_returns_none(self): + result = Video.find_or_create(None) + assert result is None + + def test_find_or_create_with_empty_string_returns_none(self): + result = Video.find_or_create("") + assert result is None + + def test_find_or_create_with_invalid_id_returns_none(self): + result = Video.find_or_create("invalid-id") + assert result is None diff --git a/vrobbler/apps/scrobbles/scrobblers.py b/vrobbler/apps/scrobbles/scrobblers.py index 78921e3..22ce300 100644 --- a/vrobbler/apps/scrobbles/scrobblers.py +++ b/vrobbler/apps/scrobbles/scrobblers.py @@ -84,11 +84,7 @@ def mopidy_scrobble_media(post_data: dict, user_id: int) -> Scrobble: run_time_seconds=post_data.get("run_time", 900000), ) try: - album_id = ( - Album.objects.filter(name=post_data.get("album", "")) - .first() - .id - ) + album_id = Album.objects.filter(name=post_data.get("album", "")).first().id except Exception: pass @@ -108,17 +104,14 @@ def mopidy_scrobble_media(post_data: dict, user_id: int) -> Scrobble: user_id, source="Mopidy", playback_position_seconds=int( - post_data.get(MOPIDY_POST_KEYS.get("PLAYBACK_POSITION_TICKS"), 1) - / 1000 + post_data.get(MOPIDY_POST_KEYS.get("PLAYBACK_POSITION_TICKS"), 1) / 1000 ), status=post_data.get(MOPIDY_POST_KEYS.get("STATUS"), ""), log=log, ) -def jellyfin_scrobble_media( - post_data: dict, user_id: int -) -> Optional[Scrobble]: +def jellyfin_scrobble_media(post_data: dict, user_id: int) -> Optional[Scrobble]: media_type = Scrobble.MediaType.VIDEO if post_data.pop("ItemType", "") in JELLYFIN_AUDIO_ITEM_TYPES: media_type = Scrobble.MediaType.TRACK @@ -136,33 +129,26 @@ def jellyfin_scrobble_media( ) return - timestamp = parse( - post_data.get(JELLYFIN_POST_KEYS.get("TIMESTAMP"), "") - ).replace(tzinfo=pytz.utc) + timestamp = parse(post_data.get(JELLYFIN_POST_KEYS.get("TIMESTAMP"), "")).replace( + tzinfo=pytz.utc + ) playback_position_seconds = int( - post_data.get(JELLYFIN_POST_KEYS.get("PLAYBACK_POSITION_TICKS"), 1) - / 10000000 + post_data.get(JELLYFIN_POST_KEYS.get("PLAYBACK_POSITION_TICKS"), 1) / 10000000 ) album_id = None if media_type == Scrobble.MediaType.VIDEO: - imdb_id = post_data.get("Provider_imdb", "") + imdb_id = post_data.get("Provider_imdb") or None media_obj = Video.find_or_create(imdb_id) else: media_obj = Track.find_or_create( title=post_data.get("Name", ""), artist_name=post_data.get("Artist", ""), album_name=post_data.get("Album", ""), - run_time_seconds=convert_to_seconds( - post_data.get("RunTime", 900000) - ), + run_time_seconds=convert_to_seconds(post_data.get("RunTime", 900000)), ) try: - album_id = ( - Album.objects.filter(name=post_data.get("Album", "")) - .first() - .id - ) + album_id = Album.objects.filter(name=post_data.get("Album", "")).first().id except Exception: pass # A hack because we don't worry about updating music ... we either finish it or we don't @@ -328,9 +314,7 @@ def manual_scrobble_book( if not page: page = 1 - logger.info( - "[scrobblers] Book page included in scrobble, should update!" - ) + logger.info("[scrobblers] Book page included in scrobble, should update!") source = READCOMICSONLINE_URL.replace("https://", "") @@ -447,9 +431,7 @@ def email_scrobble_board_game( if not location.name: location.name = location_dict.get("name") update_fields.append("name") - geoloc = GeoLocation.objects.filter( - title__icontains=location.name - ).first() + geoloc = GeoLocation.objects.filter(title__icontains=location.name).first() if geoloc: location.geo_location = geoloc update_fields.append("geo_location") @@ -501,9 +483,7 @@ def email_scrobble_board_game( log_data.pop("expansion_ids") if play_dict.get("locationRefId", False): - log_data["location_id"] = locations[ - play_dict.get("locationRefId") - ].id + log_data["location_id"] = locations[play_dict.get("locationRefId")].id if play_dict.get("rounds", False): log_data["rounds"] = play_dict.get("rounds") if play_dict.get("board", False): @@ -526,9 +506,7 @@ def email_scrobble_board_game( timestamp = parse(play_dict.get("playDate")) if hour and minute: logger.info(f"Scrobble playDate has manual start time {timestamp}") - timestamp = timestamp.replace( - hour=hour, minute=minute, second=second or 0 - ) + timestamp = timestamp.replace(hour=hour, minute=minute, second=second or 0) logger.info(f"Update to {timestamp}") profile = UserProfile.objects.filter(user_id=user_id).first() @@ -623,9 +601,7 @@ def manual_scrobble_from_url( scrobbler. Otherwise, return nothing.""" if RecipeScraperService.is_recipe_url(url): - return scrobble_from_recipe_website( - url, user_id, source=source, action=action - ) + return scrobble_from_recipe_website(url, user_id, source=source, action=action) content_key = "" domain = extract_domain(url) @@ -828,9 +804,7 @@ def emacs_scrobble_update_task( if not scrobble.log.get('notes"'): scrobble.log["notes"] = [] if note.get("timestamp") not in existing_note_ts: - scrobble.log["notes"].append( - {note.get("timestamp"): note.get("content")} - ) + scrobble.log["notes"].append({note.get("timestamp"): note.get("content")}) notes_updated = True if notes_updated: @@ -856,9 +830,7 @@ def emacs_scrobble_task( user_context_list: list[str] = [], ) -> Scrobble | None: orgmode_id = task_data.get("source_id") - title = get_title_from_labels( - task_data.get("labels", []), user_context_list - ) + title = get_title_from_labels(task_data.get("labels", []), user_context_list) task = Task.find_or_create(title) @@ -981,9 +953,7 @@ def manual_scrobble_webpage( ): if RecipeScraperService.is_recipe_url(url): - return scrobble_from_recipe_website( - url, user_id, source=source, action=action - ) + return scrobble_from_recipe_website(url, user_id, source=source, action=action) webpage = WebPage.find_or_create({"url": url}) diff --git a/vrobbler/apps/videos/migrations/0025_add_imdb_id_unique.py b/vrobbler/apps/videos/migrations/0025_add_imdb_id_unique.py new file mode 100644 index 0000000..a2e2c79 --- /dev/null +++ b/vrobbler/apps/videos/migrations/0025_add_imdb_id_unique.py @@ -0,0 +1,24 @@ +# Generated by Django 4.2.29 on 2026-03-21 18:29 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ("videos", "0024_remove_video_run_time_seconds_and_more"), + ] + + operations = [ + migrations.AlterUniqueTogether( + name="video", + unique_together=set(), + ), + migrations.AlterField( + model_name="video", + name="imdb_id", + field=models.CharField( + blank=True, max_length=20, null=True, unique=True + ), + ), + ] diff --git a/vrobbler/apps/videos/models.py b/vrobbler/apps/videos/models.py index d09873b..d649334 100644 --- a/vrobbler/apps/videos/models.py +++ b/vrobbler/apps/videos/models.py @@ -178,17 +178,13 @@ class Series(TimeStampedModel): def last_scrobbled_episode(self, user_id: int) -> Optional["Video"]: episode = None - last_scrobble = self.scrobbles_for_user( - user_id, include_playing=True - ).first() + last_scrobble = self.scrobbles_for_user(user_id, include_playing=True).first() if last_scrobble: episode = last_scrobble.media_obj return episode def is_episode_playing(self, user_id: int) -> bool: - last_scrobble = self.scrobbles_for_user( - user_id, include_playing=True - ).first() + last_scrobble = self.scrobbles_for_user(user_id, include_playing=True).first() return not last_scrobble.played_to_completion def fix_metadata(self, force_update=False): @@ -265,7 +261,7 @@ class Video(ScrobblableMixin): season_number = models.IntegerField(**BNULL) episode_number = models.IntegerField(**BNULL) next_imdb_id = models.CharField(max_length=20, **BNULL) - imdb_id = models.CharField(max_length=20, **BNULL) + imdb_id = models.CharField(max_length=20, **BNULL, unique=True) imdb_rating = models.FloatField(**BNULL) tmdb_rating = models.FloatField(**BNULL) cover_image = models.ImageField(upload_to="videos/video/", **BNULL) @@ -288,9 +284,6 @@ class Video(ScrobblableMixin): plot = models.TextField(**BNULL) upload_date = models.DateField(**BNULL) - class Meta: - unique_together = [["title", "imdb_id"]] - def __str__(self): if not self.title: return self.youtube_id or self.imdb_id @@ -363,9 +356,7 @@ class Video(ScrobblableMixin): self.cover_image.save(fname, ContentFile(r.content), save=True) @classmethod - def get_from_youtube_id( - cls, youtube_id: str, overwrite: bool = False - ) -> "Video": + def get_from_youtube_id(cls, youtube_id: str, overwrite: bool = False) -> "Video": video, created = cls.objects.get_or_create(youtube_id=youtube_id) if not created and not overwrite: return video @@ -385,9 +376,7 @@ class Video(ScrobblableMixin): return video @classmethod - def get_from_imdb_id( - cls, imdb_id: str, overwrite: bool = False - ) -> "Video": + def get_from_imdb_id(cls, imdb_id: str, overwrite: bool = False) -> "Video": video, created = cls.objects.get_or_create(imdb_id=imdb_id) if not created and not overwrite: return video @@ -414,14 +403,16 @@ class Video(ScrobblableMixin): @classmethod def find_or_create( - cls, source_id: str, overwrite: bool = False - ) -> "Video": + cls, source_id: Optional[str], overwrite: bool = False + ) -> Optional["Video"]: + if not source_id: + logger.warning("No source_id provided for Video.find_or_create") + return None if "tt" in source_id: return cls.get_from_imdb_id(source_id, overwrite) if bool(YOUTUBE_ID_PATTERN.match(source_id)): return cls.get_from_youtube_id(source_id, overwrite) - # TODO scrobble but without a video obj? logger.warning("Video ID not recognized, not scrobbling") - return + return None