525 lines
16 KiB
Python
525 lines
16 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 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 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
|