secstate/vrobbler
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[templates] Try using defensive checks for missing media
All checks were successful
build & deploy / test (push) Successful in 1m46s
Details
build & deploy / deploy (push) Successful in 21s
Details

Browse Source
This commit is contained in:
Colin Powell
2026-03-12 14:12:01 -04:00
parent f5675e5319
commit 7cf817025b
7 changed files with 97 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
vrobbler/apps/scrobbles/scrobblers.py
View File

@ -84,7 +84,11 @@ 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
@ -104,14 +108,17 @@ 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
@ -129,12 +136,13 @@ def jellyfin_scrobble_media(post_data: dict, user_id: int) -> Optional[Scrobble]
)
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:
@ -145,10 +153,16 @@ def jellyfin_scrobble_media(post_data: dict, user_id: int) -> Optional[Scrobble]
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
@ -314,7 +328,9 @@ 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://", "")
@ -431,7 +447,9 @@ 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")
@ -483,7 +501,9 @@ 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):
@ -506,7 +526,9 @@ 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()
@ -601,7 +623,9 @@ 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)
@ -804,7 +828,9 @@ 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:
@ -830,7 +856,9 @@ 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)
@ -953,7 +981,9 @@ 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})