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

[videos] Fix saving imdb_id duplicates
All checks were successful
build & deploy / test (push) Successful in 2m1s
Details
build & deploy / deploy (push) Successful in 22s
Details

Browse Source
This commit is contained in:
Colin Powell
2026-03-21 14:35:39 -04:00
parent a36abacfbd
commit 190f486c49
5 changed files with 88 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@ -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})