834 lines
26 KiB
Python
834 lines
26 KiB
Python
import hashlib
|
||
import logging
|
||
import requests
|
||
import re
|
||
from datetime import date, datetime, timedelta
|
||
from typing import TYPE_CHECKING, Optional
|
||
from urllib.parse import urlparse
|
||
from zoneinfo import ZoneInfo
|
||
|
||
import pendulum
|
||
import pytz
|
||
from django.apps import apps
|
||
from django.contrib.auth import get_user_model
|
||
from django.db import models
|
||
from django.db.models.fields.json import KeyTextTransform
|
||
from django.db.models.functions import Cast, TruncDate
|
||
from django.utils import timezone
|
||
from django.utils.dateformat import DateFormat
|
||
from profiles.models import UserProfile
|
||
from profiles.utils import now_user_timezone
|
||
from scrobbles.constants import LONG_PLAY_MEDIA
|
||
from scrobbles.notifications import (
|
||
MoodNtfyNotification,
|
||
ScrobbleNtfyNotification,
|
||
)
|
||
from scrobbles.tasks import (
|
||
process_koreader_import,
|
||
process_lastfm_import,
|
||
process_retroarch_import,
|
||
)
|
||
from webdav.client import get_webdav_client
|
||
|
||
if TYPE_CHECKING:
|
||
from scrobbles.models import Scrobble
|
||
|
||
|
||
NOTE_TIMESTAMP_FORMAT = "%Y-%m-%dT%H:%M:%S.%fZ"
|
||
|
||
|
||
def make_note_timestamp(dt: datetime | None = None) -> str:
|
||
if dt is None:
|
||
dt = timezone.now()
|
||
return dt.strftime(NOTE_TIMESTAMP_FORMAT)
|
||
|
||
|
||
logger = logging.getLogger(__name__)
|
||
User = get_user_model()
|
||
|
||
|
||
def timestamp_user_tz_to_utc(timestamp: int, user_tz: ZoneInfo) -> datetime:
|
||
return user_tz.localize(datetime.utcfromtimestamp(timestamp)).astimezone(pytz.utc)
|
||
|
||
|
||
def convert_to_seconds(run_time: str) -> int:
|
||
"""Jellyfin sends run time as 00:00:00 string. We want the run time to
|
||
actually be in seconds so we'll convert it"
|
||
|
||
This is actually deprecated, as we now convert to seconds before saving.
|
||
But for older videos, we'll leave this here.
|
||
"""
|
||
run_time_int = 0
|
||
if ":" in str(run_time):
|
||
run_time_list = run_time.split(":")
|
||
hours = int(run_time_list[0])
|
||
minutes = int(run_time_list[1])
|
||
seconds = int(run_time_list[2])
|
||
run_time_int = int((((hours * 60) + minutes) * 60) + seconds)
|
||
return run_time_int
|
||
|
||
|
||
def get_scrobbles_for_media(media_obj, user: User) -> models.QuerySet:
|
||
Scrobble = apps.get_model(app_label="scrobbles", model_name="Scrobble")
|
||
|
||
media_query = None
|
||
media_class = media_obj.__class__.__name__
|
||
if media_class == "Book":
|
||
media_query = models.Q(book=media_obj)
|
||
if media_class == "VideoGame":
|
||
media_query = models.Q(video_game=media_obj)
|
||
if media_class == "Brickset":
|
||
media_query = models.Q(brickset=media_obj)
|
||
if media_class == "Task":
|
||
media_query = models.Q(task=media_obj)
|
||
|
||
if not media_query:
|
||
logger.warn(f"Do not know about media {media_class} 🙍")
|
||
return QuerySet()
|
||
return Scrobble.objects.filter(media_query, user=user)
|
||
|
||
|
||
def get_recently_played_board_games(user: User) -> dict: ...
|
||
|
||
|
||
def get_long_plays_in_progress(user: User) -> dict:
|
||
"""Find all books where the last scrobble is not marked complete"""
|
||
media_dict = {
|
||
"active": [],
|
||
"inactive": [],
|
||
}
|
||
now = now_user_timezone(user.profile)
|
||
for app, model in LONG_PLAY_MEDIA.items():
|
||
media_obj = apps.get_model(app_label=app, model_name=model)
|
||
for media in media_obj.objects.all():
|
||
last_scrobble = media.scrobble_set.filter(user=user).last()
|
||
if last_scrobble and last_scrobble.long_play_complete == False:
|
||
days_past = (now - last_scrobble.timestamp).days
|
||
if days_past > 7:
|
||
media_dict["inactive"].append(media)
|
||
else:
|
||
media_dict["active"].append(media)
|
||
media_dict["active"].reverse()
|
||
media_dict["inactive"].reverse()
|
||
return media_dict
|
||
|
||
|
||
def get_long_plays_completed(user: User) -> list:
|
||
"""Find all books where the last scrobble is not marked complete"""
|
||
media_list = []
|
||
for app, model in LONG_PLAY_MEDIA.items():
|
||
media_obj = apps.get_model(app_label=app, model_name=model)
|
||
for media in media_obj.objects.all():
|
||
if (
|
||
media.scrobble_set.all()
|
||
and media.scrobble_set.filter(user=user)
|
||
.order_by("timestamp")
|
||
.last()
|
||
.long_play_complete
|
||
== True
|
||
):
|
||
media_list.append(media)
|
||
return media_list
|
||
|
||
|
||
def import_lastfm_for_all_users(restart=False):
|
||
"""Grab a list of all users with LastFM enabled and kickoff imports for them"""
|
||
from scrobbles.importers.lastfm import LastFM
|
||
|
||
LastFmImport = apps.get_model("scrobbles", "LastFMImport")
|
||
lastfm_enabled_user_ids = UserProfile.objects.filter(
|
||
lastfm_username__isnull=False,
|
||
lastfm_password__isnull=False,
|
||
lastfm_auto_import=True,
|
||
).values_list("user_id", flat=True)
|
||
|
||
lastfm_import_count = 0
|
||
|
||
for user_id in lastfm_enabled_user_ids:
|
||
|
||
lfm_import = LastFmImport.objects.filter(
|
||
user_id=user_id, processed_finished__isnull=False
|
||
).last()
|
||
if lfm_import:
|
||
last_processed = lfm_import.processed_finished
|
||
else:
|
||
logger.info(
|
||
f"Not resuming failed LastFM import {lfm_import.id} for user {user_id}, use restart=True to restart"
|
||
"No existing LastFM import, we should start a monthly parsing of lastFm for this user going back to 2002"
|
||
)
|
||
continue
|
||
|
||
lfm_client = LastFM(user=get_user_model().objects.filter(id=user_id).first())
|
||
|
||
has_scrobbles = lfm_client.get_last_scrobbles(
|
||
time_from=last_processed, check=True
|
||
)
|
||
|
||
if not has_scrobbles:
|
||
logger.info("No new scrobbles to import from LastFM")
|
||
continue
|
||
|
||
lfm_import, created = LastFmImport.objects.get_or_create(
|
||
user_id=user_id, processed_finished__isnull=True
|
||
)
|
||
if not created and not restart:
|
||
logger.info(
|
||
f"Not resuming failed LastFM import {lfm_import.id} for user {user_id}, use restart=True to restart"
|
||
)
|
||
continue
|
||
process_lastfm_import.delay(lfm_import.id)
|
||
lastfm_import_count += 1
|
||
return lastfm_import_count
|
||
|
||
|
||
def import_retroarch_for_all_users(restart=False):
|
||
"""Grab a list of all users with Retroarch enabled and kickoff imports for them"""
|
||
RetroarchImport = apps.get_model("scrobbles", "RetroarchImport")
|
||
retroarch_enabled_user_ids = UserProfile.objects.filter(
|
||
retroarch_path__isnull=False,
|
||
retroarch_auto_import=True,
|
||
).values_list("user_id", flat=True)
|
||
|
||
retroarch_import_count = 0
|
||
|
||
for user_id in retroarch_enabled_user_ids:
|
||
retroarch_import, created = RetroarchImport.objects.get_or_create(
|
||
user_id=user_id, processed_finished__isnull=True
|
||
)
|
||
if not created and not restart:
|
||
logger.info(
|
||
f"Not resuming failed LastFM import {retroarch_import.id} for user {user_id}, use restart=True to restart"
|
||
)
|
||
continue
|
||
process_retroarch_import.delay(retroarch_import.id)
|
||
retroarch_import_count += 1
|
||
return retroarch_import_count
|
||
|
||
|
||
def delete_zombie_scrobbles(dry_run=True):
|
||
"""Look for any scrobble over a day old that is not paused and still in progress and delete it"""
|
||
Scrobble = apps.get_model("scrobbles", "Scrobble")
|
||
now = timezone.now()
|
||
three_days_ago = now - timedelta(days=3)
|
||
|
||
# TODO This should be part of a custom manager
|
||
zombie_scrobbles = Scrobble.objects.filter(
|
||
timestamp__lte=three_days_ago,
|
||
is_paused=False,
|
||
played_to_completion=False,
|
||
)
|
||
|
||
zombies_found = zombie_scrobbles.count()
|
||
|
||
if not dry_run:
|
||
logger.info(f"Deleted {zombies_found} zombie scrobbles")
|
||
zombie_scrobbles.delete()
|
||
return zombies_found
|
||
|
||
logger.info(
|
||
f"Found {zombies_found} zombie scrobbles to delete, use dry_run=False to proceed"
|
||
)
|
||
return zombies_found
|
||
|
||
|
||
def import_from_webdav_for_all_users(restart=False):
|
||
"""Grab a list of all users with WebDAV enabled and kickoff imports for them"""
|
||
from books.koreader import fetch_file_from_webdav
|
||
from scrobbles.models import KoReaderImport
|
||
|
||
# LastFmImport = apps.get_model("scrobbles", "LastFMImport")
|
||
webdav_enabled_user_ids = UserProfile.objects.filter(
|
||
webdav_url__isnull=False,
|
||
webdav_user__isnull=False,
|
||
webdav_pass__isnull=False,
|
||
webdav_auto_import=True,
|
||
).values_list("user_id", flat=True)
|
||
logger.info(f"start import of {webdav_enabled_user_ids.count()} webdav accounts")
|
||
|
||
koreader_import_count = 0
|
||
|
||
for user_id in webdav_enabled_user_ids:
|
||
webdav_client = get_webdav_client(user_id)
|
||
|
||
try:
|
||
webdav_client.info("var/koreader/statistics.sqlite3")
|
||
koreader_found = True
|
||
except:
|
||
koreader_found = False
|
||
logger.info(
|
||
"no koreader stats file found on webdav",
|
||
extra={"user_id": user_id},
|
||
)
|
||
|
||
if koreader_found:
|
||
last_import = (
|
||
KoReaderImport.objects.filter(
|
||
user_id=user_id, processed_finished__isnull=False
|
||
)
|
||
.order_by("processed_finished")
|
||
.last()
|
||
)
|
||
|
||
koreader_file_path = fetch_file_from_webdav(1)
|
||
new_hash = get_file_md5_hash(koreader_file_path)
|
||
old_hash = None
|
||
if last_import:
|
||
old_hash = last_import.file_md5_hash()
|
||
|
||
if old_hash and new_hash == old_hash:
|
||
logger.info(
|
||
"koreader stats file has not changed",
|
||
extra={
|
||
"user_id": user_id,
|
||
"new_hash": new_hash,
|
||
"old_hash": old_hash,
|
||
"last_import_id": last_import.id,
|
||
},
|
||
)
|
||
continue
|
||
|
||
koreader_import, created = KoReaderImport.objects.get_or_create(
|
||
user_id=user_id, processed_finished__isnull=True
|
||
)
|
||
|
||
if not created and not restart:
|
||
logger.info(
|
||
f"Not resuming failed KoReader import {koreader_import.id} for user {user_id}, use restart=True to restart"
|
||
)
|
||
continue
|
||
|
||
koreader_import.save_sqlite_file_to_self(koreader_file_path)
|
||
|
||
process_koreader_import.delay(koreader_import.id)
|
||
koreader_import_count += 1
|
||
return koreader_import_count
|
||
|
||
|
||
def media_class_to_foreign_key(media_class: str) -> str:
|
||
return re.sub(r"(?<!^)(?=[A-Z])", "_", media_class).lower()
|
||
|
||
|
||
def get_file_md5_hash(file_path: str) -> str:
|
||
with open(file_path, "rb") as f:
|
||
file_hash = hashlib.md5()
|
||
while chunk := f.read(8192):
|
||
file_hash.update(chunk)
|
||
return file_hash.hexdigest()
|
||
|
||
|
||
def send_stop_notifications_for_in_progress_scrobbles() -> int:
|
||
"""Get all inprogress scrobbles and check if they're passed their media obj length.
|
||
|
||
If so, send out a notification to offer to stop the scrobble."""
|
||
from scrobbles.models import Scrobble
|
||
|
||
scrobbles_in_progress_qs = Scrobble.objects.filter(
|
||
played_to_completion=False, in_progress=True
|
||
).exclude(media_type=Scrobble.MediaType.GEO_LOCATION)
|
||
|
||
notifications_sent = 0
|
||
for scrobble in scrobbles_in_progress_qs:
|
||
finished = scrobble.auto_finish()
|
||
if scrobble.is_over_time and not finished:
|
||
ScrobbleNtfyNotification(scrobble, end=True).send()
|
||
notifications_sent += 1
|
||
|
||
return notifications_sent
|
||
|
||
|
||
def send_mood_checkin_reminders() -> int:
|
||
"""Send mood check-in to every user with the setting enabled."""
|
||
from profiles.models import UserProfile
|
||
|
||
notifications_sent = 0
|
||
for profile in UserProfile.objects.filter(mood_checkin_enabled=True):
|
||
MoodNtfyNotification(profile).send()
|
||
notifications_sent += 1
|
||
|
||
return notifications_sent
|
||
|
||
|
||
def extract_domain(url):
|
||
parsed_url = urlparse(url)
|
||
domain = parsed_url.netloc.split(".")[-2] + "." + parsed_url.netloc.split(".")[-1]
|
||
return domain
|
||
|
||
|
||
def fix_playback_position_seconds(
|
||
*scrobbles: "Scrobble", commit=True
|
||
) -> list["Scrobble"]:
|
||
updated_scrobbles = list()
|
||
for scrobble in scrobbles:
|
||
if not scrobble.media_obj:
|
||
logger.info(
|
||
f"No media object found for scrobble {scrobble.id}, cannot update elapsed time"
|
||
)
|
||
continue
|
||
|
||
if scrobble.media_type == "Track" and scrobble.media_obj.run_time_seconds:
|
||
too_long = (
|
||
scrobble.playback_position_seconds > scrobble.media_obj.run_time_seconds
|
||
)
|
||
zero = scrobble.playback_position_seconds == 0
|
||
null = not scrobble.playback_position_seconds
|
||
if too_long or zero or null:
|
||
scrobble.playback_position_seconds = scrobble.media_obj.run_time_seconds
|
||
updated_scrobbles.append(scrobble)
|
||
if commit:
|
||
scrobble.save(update_fields=["playback_position_seconds"])
|
||
|
||
return updated_scrobbles
|
||
|
||
|
||
def base_scrobble_qs(user_id: int) -> models.QuerySet:
|
||
"""Base queryset with calories extracted as integer and day annotated."""
|
||
from scrobbles.models import Scrobble
|
||
|
||
return (
|
||
Scrobble.objects.annotate(day=TruncDate("timestamp"))
|
||
.annotate(
|
||
calories_int=Cast(
|
||
KeyTextTransform("calories", "log"), models.IntegerField()
|
||
)
|
||
)
|
||
.filter(calories_int__isnull=False, user_id=user_id)
|
||
)
|
||
|
||
|
||
def get_daily_calories_for_user_by_day(user_id: int, date: date | str) -> int:
|
||
"""Return total calories for a user on a specific day."""
|
||
|
||
if isinstance(date, str):
|
||
date = pendulum.parse(date)
|
||
|
||
try:
|
||
qs = base_scrobble_qs(user_id).filter(day=date)
|
||
agg = qs.aggregate(total_calories=models.Sum("calories_int"))
|
||
except AttributeError as e:
|
||
logger.warning(f"Can't generate calorie total: {e}")
|
||
agg = {}
|
||
|
||
return agg.get("total_calories") or 0
|
||
|
||
|
||
def get_daily_calorie_dict_for_user(user_id: int) -> dict[date, int]:
|
||
"""Return {day: total_calories} for all days with scrobbles, in one query."""
|
||
qs = (
|
||
base_scrobble_qs(user_id)
|
||
.values("day")
|
||
.annotate(total_calories=models.Sum("calories_int"))
|
||
.order_by("day")
|
||
)
|
||
|
||
return {entry["day"]: entry["total_calories"] for entry in qs}
|
||
|
||
|
||
def _mopidy_rpc(profile, method, params=None):
|
||
rpc_url = profile.mopidy_api_url.rstrip("/") + "/rpc"
|
||
payload = {
|
||
"jsonrpc": "2.0",
|
||
"id": 1,
|
||
"method": method,
|
||
}
|
||
if params:
|
||
payload["params"] = params
|
||
resp = requests.post(rpc_url, json=payload, timeout=10)
|
||
resp.raise_for_status()
|
||
result = resp.json()
|
||
if result.get("error"):
|
||
raise RuntimeError(f'Mopidy error: {result["error"]}')
|
||
return result.get("result")
|
||
|
||
|
||
def _ensure_mopidy_playlist(profile):
|
||
playlist_name = profile.favorites_mopidy_playlist
|
||
# Strip any m3u: prefix and .m3u8 suffix the user may have included
|
||
playlist_name = playlist_name.removeprefix("m3u:").removesuffix(".m3u8")
|
||
|
||
try:
|
||
playlists = _mopidy_rpc(profile, "core.playlists.as_list") or []
|
||
for pl in playlists:
|
||
if pl.get("name") == playlist_name:
|
||
existing = _mopidy_rpc(
|
||
profile, "core.playlists.lookup", {"uri": pl["uri"]}
|
||
)
|
||
if existing:
|
||
return existing
|
||
except (requests.RequestException, RuntimeError):
|
||
pass
|
||
|
||
result = _mopidy_rpc(
|
||
profile, "core.playlists.create",
|
||
{"name": playlist_name, "uri_scheme": "m3u"},
|
||
)
|
||
logger.info(
|
||
"Created Mopidy favorites playlist",
|
||
extra={"uri": result.get("uri") if result else playlist_name},
|
||
)
|
||
return result
|
||
|
||
|
||
def _scrobble_with_mopidy_uri(track, user):
|
||
"""Find a scrobble for this track+user that has a mopidy_uri in its log."""
|
||
from scrobbles.models import Scrobble
|
||
|
||
for scrobble in (
|
||
Scrobble.objects.filter(track=track, user=user)
|
||
.order_by("-timestamp")
|
||
.iterator()
|
||
):
|
||
raw_data = scrobble.log.get("raw_data") or {}
|
||
if raw_data.get("mopidy_uri"):
|
||
return scrobble
|
||
return None
|
||
|
||
|
||
def add_track_to_mopidy_favorites_playlist(favorite):
|
||
if favorite.media_type != "Track" or not favorite.track:
|
||
return
|
||
|
||
profile = favorite.user.profile
|
||
if not profile.favorites_mopidy_playlist or not profile.mopidy_api_url:
|
||
return
|
||
|
||
track = favorite.track
|
||
scrobble = _scrobble_with_mopidy_uri(track, favorite.user)
|
||
|
||
if not scrobble:
|
||
logger.warning(
|
||
"No Mopidy URI found for track",
|
||
extra={"track_id": track.id, "user_id": favorite.user_id},
|
||
)
|
||
return
|
||
|
||
mopidy_uri = (scrobble.log or {}).get("raw_data", {}).get("mopidy_uri")
|
||
|
||
try:
|
||
playlist = _ensure_mopidy_playlist(profile)
|
||
if playlist and playlist.get("uri"):
|
||
existing_tracks = playlist.get("tracks") or []
|
||
track_uris = [t["uri"] for t in existing_tracks if isinstance(t, dict)]
|
||
if mopidy_uri in track_uris:
|
||
logger.info(
|
||
"Track already in Mopidy favorites playlist",
|
||
extra={"track_id": track.id, "mopidy_uri": mopidy_uri},
|
||
)
|
||
favorite.sent_to_mopidy = True
|
||
favorite.save(update_fields=["sent_to_mopidy"])
|
||
return
|
||
|
||
new_track = {"__model__": "Track", "uri": mopidy_uri}
|
||
existing_tracks.append(new_track)
|
||
_mopidy_rpc(
|
||
profile,
|
||
"core.playlists.save",
|
||
{
|
||
"playlist": {
|
||
"__model__": "Playlist",
|
||
"uri": playlist["uri"],
|
||
"name": playlist.get("name", "Favorites"),
|
||
"tracks": existing_tracks,
|
||
"last_modified": playlist.get("last_modified"),
|
||
},
|
||
},
|
||
)
|
||
else:
|
||
_mopidy_rpc(profile, "core.tracklist.add", {"uris": [mopidy_uri]})
|
||
|
||
favorite.sent_to_mopidy = True
|
||
favorite.save(update_fields=["sent_to_mopidy"])
|
||
logger.info(
|
||
"Added track to Mopidy favorites playlist",
|
||
extra={"track_id": track.id, "user_id": favorite.user_id},
|
||
)
|
||
except (requests.RequestException, RuntimeError) as e:
|
||
logger.debug(e)
|
||
logger.error(
|
||
"Failed to add track to Mopidy favorites playlist",
|
||
extra={"track_id": track.id, "user_id": favorite.user_id, "error": str(e)},
|
||
)
|
||
|
||
|
||
def resubmit_favorites_to_mopidy(user):
|
||
from scrobbles.models import FavoriteMedia
|
||
|
||
favorites = FavoriteMedia.objects.filter(
|
||
user=user,
|
||
media_type="Track",
|
||
track__isnull=False,
|
||
)
|
||
for favorite in favorites:
|
||
add_track_to_mopidy_favorites_playlist(favorite)
|
||
|
||
|
||
def remove_track_from_mopidy_favorites_playlist(favorite):
|
||
if favorite.media_type != "Track" or not favorite.track:
|
||
return
|
||
|
||
profile = favorite.user.profile
|
||
if not profile.favorites_mopidy_playlist or not profile.mopidy_api_url:
|
||
return
|
||
|
||
track = favorite.track
|
||
scrobble = _scrobble_with_mopidy_uri(track, favorite.user)
|
||
|
||
if not scrobble:
|
||
logger.warning(
|
||
"No Mopidy URI found for track",
|
||
extra={"track_id": track.id, "user_id": favorite.user_id},
|
||
)
|
||
return
|
||
|
||
mopidy_uri = (scrobble.log or {}).get("raw_data", {}).get("mopidy_uri")
|
||
|
||
try:
|
||
playlist = _ensure_mopidy_playlist(profile)
|
||
if playlist and playlist.get("uri"):
|
||
existing_tracks = playlist.get("tracks") or []
|
||
filtered = [
|
||
t for t in existing_tracks
|
||
if not (isinstance(t, dict) and t.get("uri") == mopidy_uri)
|
||
]
|
||
if len(filtered) == len(existing_tracks):
|
||
logger.info(
|
||
"Track not found in Mopidy favorites playlist",
|
||
extra={"track_id": track.id, "mopidy_uri": mopidy_uri},
|
||
)
|
||
return
|
||
|
||
_mopidy_rpc(
|
||
profile,
|
||
"core.playlists.save",
|
||
{
|
||
"playlist": {
|
||
"__model__": "Playlist",
|
||
"uri": playlist["uri"],
|
||
"name": playlist.get("name", "Favorites"),
|
||
"tracks": filtered,
|
||
"last_modified": playlist.get("last_modified"),
|
||
},
|
||
},
|
||
)
|
||
logger.info(
|
||
"Removed track from Mopidy favorites playlist",
|
||
extra={"track_id": track.id, "user_id": favorite.user_id},
|
||
)
|
||
except (requests.RequestException, RuntimeError) as e:
|
||
logger.debug(e)
|
||
logger.error(
|
||
"Failed to remove track from Mopidy favorites playlist",
|
||
extra={"track_id": track.id, "user_id": favorite.user_id, "error": str(e)},
|
||
)
|
||
|
||
|
||
def _ensure_mopidy_playlist_by_name(profile, playlist_name):
|
||
"""Find or create a Mopidy playlist by name (without m3u: prefix handling)."""
|
||
playlist_name = playlist_name.removeprefix("m3u:").removesuffix(".m3u8")
|
||
try:
|
||
playlists = _mopidy_rpc(profile, "core.playlists.as_list") or []
|
||
for pl in playlists:
|
||
if pl.get("name") == playlist_name:
|
||
existing = _mopidy_rpc(
|
||
profile, "core.playlists.lookup", {"uri": pl["uri"]}
|
||
)
|
||
if existing:
|
||
return existing
|
||
except (requests.RequestException, RuntimeError):
|
||
pass
|
||
|
||
result = _mopidy_rpc(
|
||
profile, "core.playlists.create",
|
||
{"name": playlist_name, "uri_scheme": "m3u"},
|
||
)
|
||
return result
|
||
|
||
|
||
def add_track_to_mopidy_monthly_playlist(scrobble):
|
||
"""Add a scrobbled track to a monthly Mopidy playlist based on the user's pattern."""
|
||
profile = scrobble.user.profile
|
||
pattern = profile.monthly_mopidy_playlist_pattern
|
||
if not pattern or not profile.mopidy_api_url:
|
||
return
|
||
|
||
mopidy_uri = (scrobble.log or {}).get("raw_data", {}).get("mopidy_uri")
|
||
if not mopidy_uri:
|
||
return
|
||
|
||
now = now_user_timezone(profile)
|
||
playlist_name = DateFormat(now).format(pattern)
|
||
if not playlist_name:
|
||
return
|
||
|
||
try:
|
||
playlist = _ensure_mopidy_playlist_by_name(profile, playlist_name)
|
||
if playlist and playlist.get("uri"):
|
||
existing_tracks = playlist.get("tracks") or []
|
||
track_uris = [t["uri"] for t in existing_tracks if isinstance(t, dict)]
|
||
if mopidy_uri in track_uris:
|
||
logger.info(
|
||
"Track already in monthly Mopidy playlist",
|
||
extra={"playlist": playlist_name, "mopidy_uri": mopidy_uri},
|
||
)
|
||
return
|
||
|
||
new_track = {"__model__": "Track", "uri": mopidy_uri}
|
||
existing_tracks.append(new_track)
|
||
_mopidy_rpc(
|
||
profile,
|
||
"core.playlists.save",
|
||
{
|
||
"playlist": {
|
||
"__model__": "Playlist",
|
||
"uri": playlist["uri"],
|
||
"name": playlist.get("name", playlist_name),
|
||
"tracks": existing_tracks,
|
||
"last_modified": playlist.get("last_modified"),
|
||
},
|
||
},
|
||
)
|
||
else:
|
||
_mopidy_rpc(profile, "core.tracklist.add", {"uris": [mopidy_uri]})
|
||
|
||
logger.info(
|
||
"Added track to monthly Mopidy playlist",
|
||
extra={
|
||
"playlist": playlist_name,
|
||
"track_id": scrobble.media_obj.id,
|
||
"user_id": scrobble.user_id,
|
||
},
|
||
)
|
||
except (requests.RequestException, RuntimeError) as e:
|
||
logger.debug(e)
|
||
logger.error(
|
||
"Failed to add track to monthly Mopidy playlist",
|
||
extra={"playlist": playlist_name, "error": str(e)},
|
||
)
|
||
|
||
|
||
def remove_last_part(url: str) -> str:
|
||
url = url.rstrip("/")
|
||
if "/" not in url:
|
||
return url
|
||
return url.rsplit("/", 1)[0]
|
||
|
||
|
||
def next_url_if_exists(url: str) -> str:
|
||
# Normalize (remove trailing slash)
|
||
url = url.rstrip("/")
|
||
|
||
# Find last number in the URL path
|
||
match = re.search(r"(\d+)(?:/?$)", url)
|
||
if not match:
|
||
logger.info("No numeric segment found in the URL", extra={"url": url})
|
||
return ""
|
||
|
||
number = int(match.group(1))
|
||
new_number = number + 1
|
||
|
||
# Replace only the last occurrence of that number
|
||
new_url = re.sub(rf"{number}(?:/?$)", f"{new_number}/", url + "/", 1)
|
||
|
||
# Check if the new URL exists
|
||
try:
|
||
resp = requests.head(new_url, allow_redirects=True, timeout=5)
|
||
if resp.status_code == 200:
|
||
return new_url
|
||
else:
|
||
# Fallback: some sites may not support HEAD well — try GET
|
||
resp = requests.get(new_url, timeout=5)
|
||
if resp.status_code == 200:
|
||
return new_url
|
||
except requests.RequestException:
|
||
pass
|
||
|
||
# If it doesn’t exist
|
||
return ""
|
||
|
||
|
||
STOPWORDS = {
|
||
"a",
|
||
"an",
|
||
"the",
|
||
"and",
|
||
"or",
|
||
"but",
|
||
"in",
|
||
"on",
|
||
"at",
|
||
"to",
|
||
"for",
|
||
"of",
|
||
"with",
|
||
"by",
|
||
"from",
|
||
"is",
|
||
"as",
|
||
"it",
|
||
"be",
|
||
"are",
|
||
"was",
|
||
"were",
|
||
"been",
|
||
"have",
|
||
"has",
|
||
"had",
|
||
"do",
|
||
"does",
|
||
"did",
|
||
"will",
|
||
"would",
|
||
"could",
|
||
"should",
|
||
"might",
|
||
"must",
|
||
"this",
|
||
"that",
|
||
"these",
|
||
"those",
|
||
"your",
|
||
"their",
|
||
}
|
||
|
||
|
||
def tokenize_title_to_tags(title: str) -> list[str]:
|
||
"""Tokenize a title into tags, filtering common words and short tokens."""
|
||
if not title:
|
||
return []
|
||
|
||
cleaned = re.sub(r"[\(\)\[\]\{\}]", "", title)
|
||
cleaned = re.sub(r"[^\w\s]", "", cleaned)
|
||
|
||
words = [
|
||
w.lower() for w in cleaned.split() if w.lower() not in STOPWORDS and len(w) > 2
|
||
]
|
||
return words
|
||
|
||
|
||
def analyze_scrobble_sentiment(scrobble, overwrite=False) -> bool:
|
||
"""Run VADER sentiment analysis on a scrobble's notes.
|
||
|
||
Stores result in log["sentiment"] as a dict with keys:
|
||
neg, neu, pos, compound.
|
||
|
||
Returns True if analyzed, False if skipped (no notes or already done).
|
||
"""
|
||
from vaderSentiment.vaderSentiment import SentimentIntensityAnalyzer
|
||
|
||
log = scrobble.log or {}
|
||
if not overwrite and log.get("sentiment") is not None:
|
||
return False
|
||
|
||
notes_str = ""
|
||
if scrobble.logdata:
|
||
notes_str = scrobble.logdata.notes_as_str()
|
||
if not notes_str:
|
||
return False
|
||
|
||
analyzer = SentimentIntensityAnalyzer()
|
||
scores = analyzer.polarity_scores(notes_str)
|
||
|
||
log["sentiment"] = scores
|
||
scrobble.log = log
|
||
scrobble.save(update_fields=["log"])
|
||
return True
|