1531 lines
50 KiB
Python
1531 lines
50 KiB
Python
import calendar
|
|
import datetime
|
|
import json
|
|
import logging
|
|
from collections import defaultdict
|
|
from typing import Optional
|
|
from uuid import uuid4
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pendulum
|
|
import pytz
|
|
from beers.models import Beer
|
|
from boardgames.models import BoardGame
|
|
from books.koreader import process_koreader_sqlite_file
|
|
from books.models import Book, Paper, BookPageLogData, BookLogData
|
|
from bricksets.models import BrickSet
|
|
from dataclass_wizard.errors import ParseError
|
|
from django.conf import settings
|
|
from django.contrib.auth import get_user_model
|
|
from django.core.files import File
|
|
from django.db import models
|
|
from django.urls import reverse
|
|
from django.utils import timezone
|
|
from django_extensions.db.models import TimeStampedModel
|
|
from foods.models import Food
|
|
from imagekit.models import ImageSpecField
|
|
from imagekit.processors import ResizeToFit
|
|
from lifeevents.models import LifeEvent
|
|
from locations.models import GeoLocation
|
|
from moods.models import Mood
|
|
from music.models import Artist, Track
|
|
from podcasts.models import PodcastEpisode
|
|
from profiles.utils import (
|
|
end_of_day,
|
|
end_of_month,
|
|
end_of_week,
|
|
fix_profile_historic_timezones,
|
|
start_of_day,
|
|
start_of_month,
|
|
start_of_week,
|
|
)
|
|
from puzzles.models import Puzzle
|
|
from scrobbles import dataclasses as logdata
|
|
from scrobbles.constants import LONG_PLAY_MEDIA, MEDIA_END_PADDING_SECONDS
|
|
from scrobbles.importers.lastfm import LastFM
|
|
from scrobbles.notifications import ScrobbleNtfyNotification
|
|
from scrobbles.stats import build_charts
|
|
from scrobbles.utils import get_file_md5_hash, media_class_to_foreign_key
|
|
from sports.models import SportEvent
|
|
from tasks.models import Task
|
|
from trails.models import Trail
|
|
from videogames import retroarch
|
|
from videogames.models import VideoGame
|
|
from videos.models import Series, Video
|
|
from webpages.models import WebPage
|
|
|
|
logger = logging.getLogger(__name__)
|
|
User = get_user_model()
|
|
BNULL = {"blank": True, "null": True}
|
|
|
|
POINTS_FOR_MOVEMENT_HISTORY = int(
|
|
getattr(settings, "POINTS_FOR_MOVEMENT_HISTORY", 3)
|
|
)
|
|
|
|
|
|
class BaseFileImportMixin(TimeStampedModel):
|
|
user = models.ForeignKey(User, on_delete=models.DO_NOTHING, **BNULL)
|
|
uuid = models.UUIDField(editable=False, default=uuid4)
|
|
processing_started = models.DateTimeField(**BNULL)
|
|
processed_finished = models.DateTimeField(**BNULL)
|
|
process_log = models.TextField(**BNULL)
|
|
process_count = models.IntegerField(**BNULL)
|
|
|
|
class Meta:
|
|
abstract = True
|
|
|
|
def __str__(self):
|
|
return f"{self.import_type} import on {self.human_start}"
|
|
|
|
@property
|
|
def human_start(self):
|
|
start = "Unknown"
|
|
if self.processing_started:
|
|
start = self.processing_started.strftime("%B %d, %Y at %H:%M")
|
|
return start
|
|
|
|
@property
|
|
def import_type(self) -> str:
|
|
return "Unknown Import Source"
|
|
|
|
def process(self, force=False):
|
|
logger.warning("Process not implemented")
|
|
|
|
def undo(self, dryrun=False):
|
|
"""Accepts the log from a scrobble import and removes the scrobbles"""
|
|
from scrobbles.models import Scrobble
|
|
|
|
if not self.process_log:
|
|
|
|
logger.warning("No lines in process log found to undo")
|
|
return
|
|
|
|
for line in self.process_log.split("\n"):
|
|
scrobble_id = line.split("\t")[0]
|
|
scrobble = Scrobble.objects.filter(id=scrobble_id).first()
|
|
if not scrobble:
|
|
logger.warning(
|
|
f"Could not find scrobble {scrobble_id} to undo"
|
|
)
|
|
continue
|
|
logger.info(f"Removing scrobble {scrobble_id}")
|
|
if not dryrun:
|
|
scrobble.delete()
|
|
self.processed_finished = None
|
|
self.processing_started = None
|
|
self.process_count = None
|
|
self.process_log = ""
|
|
self.save(
|
|
update_fields=[
|
|
"processed_finished",
|
|
"processing_started",
|
|
"process_log",
|
|
"process_count",
|
|
]
|
|
)
|
|
|
|
def scrobbles(self) -> models.QuerySet:
|
|
scrobble_ids = []
|
|
if self.process_log:
|
|
for line in self.process_log.split("\n"):
|
|
sid = line.split("\t")[0]
|
|
if sid:
|
|
scrobble_ids.append(sid)
|
|
return Scrobble.objects.filter(id__in=scrobble_ids)
|
|
|
|
def mark_started(self):
|
|
self.processing_started = timezone.now()
|
|
self.save(update_fields=["processing_started"])
|
|
|
|
def mark_finished(self):
|
|
self.processed_finished = timezone.now()
|
|
self.save(update_fields=["processed_finished"])
|
|
|
|
def record_log(self, scrobbles):
|
|
self.process_log = ""
|
|
if not scrobbles:
|
|
self.process_count = 0
|
|
self.save(update_fields=["process_log", "process_count"])
|
|
return
|
|
|
|
for count, scrobble in enumerate(scrobbles):
|
|
scrobble_str = (
|
|
f"{scrobble.id}\t{scrobble.timestamp}\t{scrobble.media_obj}"
|
|
)
|
|
log_line = f"{scrobble_str}"
|
|
if count > 0:
|
|
log_line = "\n" + log_line
|
|
self.process_log += log_line
|
|
self.process_count = len(scrobbles)
|
|
self.save(update_fields=["process_log", "process_count"])
|
|
|
|
@property
|
|
def upload_file_path(self):
|
|
raise NotImplementedError
|
|
|
|
|
|
class KoReaderImport(BaseFileImportMixin):
|
|
class Meta:
|
|
verbose_name = "KOReader Import"
|
|
|
|
@property
|
|
def import_type(self) -> str:
|
|
return "KOReader"
|
|
|
|
def get_absolute_url(self):
|
|
return reverse(
|
|
"scrobbles:koreader-import-detail", kwargs={"slug": self.uuid}
|
|
)
|
|
|
|
def get_path(instance, filename):
|
|
extension = filename.split(".")[-1]
|
|
uuid = instance.uuid
|
|
return f"koreader-uploads/{uuid}.{extension}"
|
|
|
|
@property
|
|
def upload_file_path(self) -> str:
|
|
if getattr(settings, "USE_S3_STORAGE"):
|
|
path = self.sqlite_file.url
|
|
else:
|
|
path = self.sqlite_file.path
|
|
return path
|
|
|
|
sqlite_file = models.FileField(upload_to=get_path, **BNULL)
|
|
|
|
def save_sqlite_file_to_self(self, file_path):
|
|
with open(file_path, "rb") as f:
|
|
self.sqlite_file.save(
|
|
f"{self.user_id}-koreader-statistics.sqlite",
|
|
File(f),
|
|
save=True,
|
|
)
|
|
|
|
def file_md5_hash(self) -> str:
|
|
if self.sqlite_file:
|
|
return get_file_md5_hash(self.sqlite_file.path)
|
|
return ""
|
|
|
|
def process(self, force=False):
|
|
|
|
if self.user.id == 1:
|
|
fix_profile_historic_timezones(self.user.profile)
|
|
|
|
if self.processed_finished and not force:
|
|
logger.info(
|
|
f"{self} already processed on {self.processed_finished}"
|
|
)
|
|
return
|
|
|
|
self.mark_started()
|
|
scrobbles = process_koreader_sqlite_file(
|
|
self.upload_file_path, self.user.id
|
|
)
|
|
self.record_log(scrobbles)
|
|
self.mark_finished()
|
|
|
|
|
|
class AudioScrobblerTSVImport(BaseFileImportMixin):
|
|
class Meta:
|
|
verbose_name = "AudioScrobbler TSV Import"
|
|
|
|
@property
|
|
def import_type(self) -> str:
|
|
return "AudiosScrobbler"
|
|
|
|
def get_absolute_url(self):
|
|
return reverse(
|
|
"scrobbles:tsv-import-detail", kwargs={"slug": self.uuid}
|
|
)
|
|
|
|
def get_path(instance, filename):
|
|
extension = filename.split(".")[-1]
|
|
uuid = instance.uuid
|
|
return f"audioscrobbler-uploads/{uuid}.{extension}"
|
|
|
|
@property
|
|
def upload_file_path(self):
|
|
if getattr(settings, "USE_S3_STORAGE"):
|
|
path = self.tsv_file.url
|
|
else:
|
|
path = self.tsv_file.path
|
|
return path
|
|
|
|
tsv_file = models.FileField(upload_to=get_path, **BNULL)
|
|
|
|
def process(self, force=False):
|
|
from scrobbles.importers.tsv import import_audioscrobbler_tsv_file
|
|
|
|
if self.user.id == 1:
|
|
fix_profile_historic_timezones(self.user.profile)
|
|
|
|
if self.processed_finished and not force:
|
|
logger.info(
|
|
f"{self} already processed on {self.processed_finished}"
|
|
)
|
|
return
|
|
|
|
self.mark_started()
|
|
|
|
scrobbles = import_audioscrobbler_tsv_file(
|
|
self.upload_file_path, self.user.id
|
|
)
|
|
self.record_log(scrobbles)
|
|
self.mark_finished()
|
|
|
|
|
|
class LastFmImport(BaseFileImportMixin):
|
|
class Meta:
|
|
verbose_name = "Last.FM Import"
|
|
|
|
@property
|
|
def import_type(self) -> str:
|
|
return "LastFM"
|
|
|
|
def get_absolute_url(self):
|
|
return reverse(
|
|
"scrobbles:lastfm-import-detail", kwargs={"slug": self.uuid}
|
|
)
|
|
|
|
def process(self, import_all=False):
|
|
"""Import scrobbles found on LastFM"""
|
|
|
|
if self.user.id == 1:
|
|
fix_profile_historic_timezones(self.user.profile)
|
|
|
|
if self.processed_finished:
|
|
logger.info(
|
|
f"{self} already processed on {self.processed_finished}"
|
|
)
|
|
return
|
|
|
|
last_import = None
|
|
if not import_all:
|
|
try:
|
|
last_import = LastFmImport.objects.exclude(id=self.id).last()
|
|
except:
|
|
pass
|
|
|
|
if not import_all and not last_import:
|
|
logger.warn(
|
|
"No previous import, to import all Last.fm scrobbles, pass import_all=True"
|
|
)
|
|
return
|
|
|
|
lastfm = LastFM(self.user)
|
|
last_processed = None
|
|
if last_import:
|
|
last_processed = last_import.processed_finished
|
|
|
|
self.mark_started()
|
|
|
|
scrobbles = lastfm.import_from_lastfm(last_processed)
|
|
|
|
self.record_log(scrobbles)
|
|
self.mark_finished()
|
|
|
|
|
|
class RetroarchImport(BaseFileImportMixin):
|
|
class Meta:
|
|
verbose_name = "Retroarch Import"
|
|
|
|
@property
|
|
def import_type(self) -> str:
|
|
return "Retroarch"
|
|
|
|
def get_absolute_url(self):
|
|
return reverse(
|
|
"scrobbles:retroarch-import-detail", kwargs={"slug": self.uuid}
|
|
)
|
|
|
|
def process(self, import_all=False, force=False):
|
|
"""Import scrobbles found on Retroarch"""
|
|
if self.user.id == 1:
|
|
fix_profile_historic_timezones(self.user.profile)
|
|
|
|
if self.processed_finished and not force:
|
|
logger.info(
|
|
f"{self} already processed on {self.processed_finished}"
|
|
)
|
|
return
|
|
|
|
if force:
|
|
logger.info(f"You told me to force import from Retroarch")
|
|
|
|
if not self.user.profile.retroarch_path:
|
|
logger.info(
|
|
"Tying to import Retroarch logs, but user has no retroarch_path configured"
|
|
)
|
|
|
|
self.mark_started()
|
|
|
|
scrobbles = retroarch.import_retroarch_lrtl_files(
|
|
self.user.profile.retroarch_path,
|
|
self.user.id,
|
|
)
|
|
|
|
self.record_log(scrobbles)
|
|
self.mark_finished()
|
|
|
|
|
|
class ChartRecord(TimeStampedModel):
|
|
"""Sort of like a materialized view for what we could dynamically generate,
|
|
but would kill the DB as it gets larger. Collects time-based records
|
|
generated by a cron-like archival job
|
|
|
|
1972 by Josh Rouse - #3 in 2023, January
|
|
|
|
"""
|
|
|
|
user = models.ForeignKey(User, on_delete=models.DO_NOTHING, **BNULL)
|
|
rank = models.IntegerField(db_index=True)
|
|
count = models.IntegerField(default=0)
|
|
year = models.IntegerField(**BNULL)
|
|
month = models.IntegerField(**BNULL)
|
|
week = models.IntegerField(**BNULL)
|
|
day = models.IntegerField(**BNULL)
|
|
video = models.ForeignKey(Video, on_delete=models.DO_NOTHING, **BNULL)
|
|
series = models.ForeignKey(Series, on_delete=models.DO_NOTHING, **BNULL)
|
|
artist = models.ForeignKey(Artist, on_delete=models.DO_NOTHING, **BNULL)
|
|
track = models.ForeignKey(Track, on_delete=models.DO_NOTHING, **BNULL)
|
|
period_start = models.DateTimeField(**BNULL)
|
|
period_end = models.DateTimeField(**BNULL)
|
|
|
|
def save(self, *args, **kwargs):
|
|
profile = self.user.profile
|
|
|
|
if self.week:
|
|
# set start and end to start and end of week
|
|
period = datetime.date.fromisocalendar(self.year, self.week, 1)
|
|
self.period_start = start_of_week(period, profile)
|
|
self.period_start = end_of_week(period, profile)
|
|
if self.day:
|
|
period = datetime.datetime(self.year, self.month, self.day)
|
|
self.period_start = start_of_day(period, profile)
|
|
self.period_end = end_of_day(period, profile)
|
|
if self.month and not self.day:
|
|
period = datetime.datetime(self.year, self.month, 1)
|
|
self.period_start = start_of_month(period, profile)
|
|
self.period_end = end_of_month(period, profile)
|
|
super(ChartRecord, self).save(*args, **kwargs)
|
|
|
|
@property
|
|
def media_obj(self):
|
|
media_obj = None
|
|
if self.video:
|
|
media_obj = self.video
|
|
if self.track:
|
|
media_obj = self.track
|
|
if self.artist:
|
|
media_obj = self.artist
|
|
return media_obj
|
|
|
|
@property
|
|
def month_str(self) -> str:
|
|
month_str = ""
|
|
if self.month:
|
|
month_str = calendar.month_name[self.month]
|
|
return month_str
|
|
|
|
@property
|
|
def day_str(self) -> str:
|
|
day_str = ""
|
|
if self.day:
|
|
day_str = str(self.day)
|
|
return day_str
|
|
|
|
@property
|
|
def week_str(self) -> str:
|
|
week_str = ""
|
|
if self.week:
|
|
week_str = str(self.week)
|
|
return "Week " + week_str
|
|
|
|
@property
|
|
def period(self) -> str:
|
|
period = str(self.year)
|
|
if self.month:
|
|
period = " ".join([self.month_str, period])
|
|
if self.week:
|
|
period = " ".join([self.week_str, period])
|
|
if self.day:
|
|
period = " ".join([self.day_str, period])
|
|
return period
|
|
|
|
@property
|
|
def period_type(self) -> str:
|
|
period = "year"
|
|
if self.month:
|
|
period = "month"
|
|
if self.week:
|
|
period = "week"
|
|
if self.day:
|
|
period = "day"
|
|
return period
|
|
|
|
def __str__(self):
|
|
title = f"#{self.rank} in {self.period}"
|
|
if self.day or self.week:
|
|
title = f"#{self.rank} on {self.period}"
|
|
return title
|
|
|
|
def link(self):
|
|
get_params = f"?date={self.year}"
|
|
if self.week:
|
|
get_params = get_params = get_params + f"-W{self.week}"
|
|
if self.month:
|
|
get_params = get_params = get_params + f"-{self.month}"
|
|
if self.day:
|
|
get_params = get_params = get_params + f"-{self.day}"
|
|
if self.artist:
|
|
get_params = get_params + "&media=Artist"
|
|
return reverse("scrobbles:charts-home") + get_params
|
|
|
|
@classmethod
|
|
def build(cls, user, **kwargs):
|
|
build_charts(user=user, **kwargs)
|
|
|
|
@classmethod
|
|
def for_year(cls, user, year):
|
|
return cls.objects.filter(year=year, user=user)
|
|
|
|
@classmethod
|
|
def for_month(cls, user, year, month):
|
|
return cls.objects.filter(year=year, month=month, user=user)
|
|
|
|
@classmethod
|
|
def for_day(cls, user, year, day, month):
|
|
return cls.objects.filter(year=year, month=month, day=day, user=user)
|
|
|
|
@classmethod
|
|
def for_week(cls, user, year, week):
|
|
return cls.objects.filter(year=year, week=week, user=user)
|
|
|
|
|
|
class Scrobble(TimeStampedModel):
|
|
"""A scrobble tracks played media items by a user."""
|
|
|
|
class MediaType(models.TextChoices):
|
|
"""Enum mapping a media model type to a string"""
|
|
|
|
VIDEO = "Video", "Video"
|
|
TRACK = "Track", "Track"
|
|
PODCAST_EPISODE = "PodcastEpisode", "Podcast episode"
|
|
SPORT_EVENT = "SportEvent", "Sport event"
|
|
BOOK = "Book", "Book"
|
|
PAPER = "Paper", "Paper"
|
|
VIDEO_GAME = "VideoGame", "Video game"
|
|
BOARD_GAME = "BoardGame", "Board game"
|
|
GEO_LOCATION = "GeoLocation", "GeoLocation"
|
|
TRAIL = "Trail", "Trail"
|
|
BEER = "Beer", "Beer"
|
|
PUZZLE = "Puzzle", "Puzzle"
|
|
FOOD = "Food", "Food"
|
|
TASK = "Task", "Task"
|
|
WEBPAGE = "WebPage", "Web Page"
|
|
LIFE_EVENT = "LifeEvent", "Life event"
|
|
MOOD = "Mood", "Mood"
|
|
BRICKSET = "BrickSet", "Brick set"
|
|
|
|
@classmethod
|
|
def list(cls):
|
|
return list(map(lambda c: c.value, cls))
|
|
|
|
uuid = models.UUIDField(editable=False, **BNULL)
|
|
video = models.ForeignKey(Video, on_delete=models.DO_NOTHING, **BNULL)
|
|
track = models.ForeignKey(Track, on_delete=models.DO_NOTHING, **BNULL)
|
|
podcast_episode = models.ForeignKey(
|
|
PodcastEpisode, on_delete=models.DO_NOTHING, **BNULL
|
|
)
|
|
sport_event = models.ForeignKey(
|
|
SportEvent, on_delete=models.DO_NOTHING, **BNULL
|
|
)
|
|
book = models.ForeignKey(Book, on_delete=models.DO_NOTHING, **BNULL)
|
|
paper = models.ForeignKey(Paper, on_delete=models.DO_NOTHING, **BNULL)
|
|
video_game = models.ForeignKey(
|
|
VideoGame, on_delete=models.DO_NOTHING, **BNULL
|
|
)
|
|
board_game = models.ForeignKey(
|
|
BoardGame, on_delete=models.DO_NOTHING, **BNULL
|
|
)
|
|
geo_location = models.ForeignKey(
|
|
GeoLocation, on_delete=models.DO_NOTHING, **BNULL
|
|
)
|
|
beer = models.ForeignKey(Beer, on_delete=models.DO_NOTHING, **BNULL)
|
|
puzzle = models.ForeignKey(Puzzle, on_delete=models.DO_NOTHING, **BNULL)
|
|
food = models.ForeignKey(Food, on_delete=models.DO_NOTHING, **BNULL)
|
|
trail = models.ForeignKey(Trail, on_delete=models.DO_NOTHING, **BNULL)
|
|
task = models.ForeignKey(Task, on_delete=models.DO_NOTHING, **BNULL)
|
|
web_page = models.ForeignKey(WebPage, on_delete=models.DO_NOTHING, **BNULL)
|
|
life_event = models.ForeignKey(
|
|
LifeEvent, on_delete=models.DO_NOTHING, **BNULL
|
|
)
|
|
mood = models.ForeignKey(Mood, on_delete=models.DO_NOTHING, **BNULL)
|
|
brick_set = models.ForeignKey(
|
|
BrickSet, on_delete=models.DO_NOTHING, **BNULL
|
|
)
|
|
media_type = models.CharField(
|
|
max_length=14, choices=MediaType.choices, default=MediaType.VIDEO
|
|
)
|
|
user = models.ForeignKey(
|
|
User, blank=True, null=True, on_delete=models.DO_NOTHING
|
|
)
|
|
|
|
# Time keeping
|
|
timestamp = models.DateTimeField(**BNULL)
|
|
stop_timestamp = models.DateTimeField(**BNULL)
|
|
playback_position_seconds = models.IntegerField(**BNULL)
|
|
|
|
# Status indicators
|
|
is_paused = models.BooleanField(default=False)
|
|
played_to_completion = models.BooleanField(default=False)
|
|
in_progress = models.BooleanField(default=True)
|
|
|
|
# Metadata
|
|
source = models.CharField(max_length=255, **BNULL)
|
|
log = models.JSONField(
|
|
**BNULL,
|
|
default=dict,
|
|
encoder=logdata.ScrobbleLogDataEncoder,
|
|
decoder=logdata.ScrobbleLogDataDecoder,
|
|
)
|
|
timezone = models.CharField(max_length=50, **BNULL)
|
|
|
|
# Fields for keeping track of video game data
|
|
videogame_save_data = models.FileField(
|
|
upload_to="scrobbles/videogame_save_data/", **BNULL
|
|
)
|
|
gpx_file = models.FileField(upload_to="scrobbles/gpx_file/", **BNULL)
|
|
screenshot = models.ImageField(
|
|
upload_to="scrobbles/videogame_screenshot/", **BNULL
|
|
)
|
|
screenshot_small = ImageSpecField(
|
|
source="screenshot",
|
|
processors=[ResizeToFit(100, 100)],
|
|
format="JPEG",
|
|
options={"quality": 60},
|
|
)
|
|
screenshot_medium = ImageSpecField(
|
|
source="screenshot",
|
|
processors=[ResizeToFit(300, 300)],
|
|
format="JPEG",
|
|
options={"quality": 75},
|
|
)
|
|
long_play_seconds = models.BigIntegerField(**BNULL)
|
|
long_play_complete = models.BooleanField(**BNULL)
|
|
|
|
@classmethod
|
|
def for_year(cls, user, year):
|
|
return cls.objects.filter(timestamp__year=year, user=user).order_by(
|
|
"-timestamp"
|
|
)
|
|
|
|
@classmethod
|
|
def for_month(cls, user, year, month):
|
|
return cls.objects.filter(
|
|
timestamp__year=year, timestamp__month=month, user=user
|
|
).order_by("-timestamp")
|
|
|
|
@classmethod
|
|
def for_day(cls, user, year, month, day):
|
|
return cls.objects.filter(
|
|
timestamp__year=year,
|
|
timestamp__month=month,
|
|
timestamp__day=day,
|
|
user=user,
|
|
).order_by("-timestamp")
|
|
|
|
@classmethod
|
|
def for_week(cls, user, year, week):
|
|
return cls.objects.filter(
|
|
timestamp__year=year, timestamp__week=week, user=user
|
|
).order_by("-timestamp")
|
|
|
|
@classmethod
|
|
def as_dict_by_type(cls, scrobble_qs: models.QuerySet) -> dict:
|
|
scrobbles_by_type = defaultdict(list)
|
|
|
|
for scrobble in scrobble_qs:
|
|
scrobbles_by_type[scrobble.media_type].append(scrobble)
|
|
if not scrobbles_by_type.get(scrobble.media_type + "_count"):
|
|
scrobbles_by_type[scrobble.media_type + "_count"] = 0
|
|
scrobbles_by_type[scrobble.media_type + "_count"] += 1
|
|
if not scrobbles_by_type.get(scrobble.media_type + "_time"):
|
|
scrobbles_by_type[scrobble.media_type + "_time"] = 0
|
|
scrobbles_by_type[scrobble.media_type + "_time"] += int(
|
|
(scrobble.elapsed_time)
|
|
)
|
|
|
|
# Remove any locations without titles
|
|
if "GeoLocation" in scrobbles_by_type.keys():
|
|
for loc_scrobble in scrobbles_by_type["GeoLocation"]:
|
|
if not loc_scrobble.media_obj.title:
|
|
scrobbles_by_type["GeoLocation"].remove(loc_scrobble)
|
|
scrobbles_by_type["GeoLocation_count"] -= 1
|
|
|
|
return scrobbles_by_type
|
|
|
|
@classmethod
|
|
def in_progress_for_user(cls, user_id: int) -> models.QuerySet:
|
|
return cls.objects.filter(
|
|
user=user_id,
|
|
in_progress=True,
|
|
played_to_completion=False,
|
|
is_paused=False,
|
|
)
|
|
|
|
@property
|
|
def last_serial_scrobble(self) -> Optional["Scrobble"]:
|
|
from scrobbles.models import Scrobble
|
|
|
|
if self.logdata and self.logdata.serial_scrobble_id:
|
|
return Scrobble.objects.filter(
|
|
id=self.logdata.serial_scrobble_id
|
|
).first()
|
|
|
|
@property
|
|
def finish_url(self) -> str:
|
|
return reverse("scrobbles:finish", kwargs={"uuid": self.uuid})
|
|
|
|
def save(self, *args, **kwargs):
|
|
if not self.uuid:
|
|
self.uuid = uuid4()
|
|
|
|
if not self.timezone:
|
|
timezone = settings.TIME_ZONE
|
|
if self.user and self.user.profile:
|
|
timezone = self.user.profile.timezone
|
|
self.timezone = timezone
|
|
|
|
# Microseconds mess up Django's filtering, and we don't need be that specific
|
|
if self.timestamp:
|
|
self.timestamp = self.timestamp.replace(microsecond=0)
|
|
if self.media_obj:
|
|
self.media_type = self.MediaType(self.media_obj.__class__.__name__)
|
|
|
|
return super(Scrobble, self).save(*args, **kwargs)
|
|
|
|
def get_absolute_url(self):
|
|
if not self.uuid:
|
|
self.uuid = uuid4()
|
|
self.save()
|
|
return reverse("scrobbles:detail", kwargs={"uuid": self.uuid})
|
|
|
|
def push_to_archivebox(self):
|
|
pushable_media = hasattr(
|
|
self.media_obj, "push_to_archivebox"
|
|
) and callable(self.media_obj.push_to_archivebox)
|
|
|
|
if pushable_media and self.user.profile.archivebox_url:
|
|
try:
|
|
self.media_obj.push_to_archivebox(
|
|
url=self.user.profile.archivebox_url,
|
|
username=self.user.profile.archivebox_username,
|
|
password=self.user.profile.archivebox_password,
|
|
)
|
|
except Exception:
|
|
logger.info(
|
|
"Failed to push URL to archivebox",
|
|
extra={
|
|
"archivebox_url": self.user.profile.archivebox_url,
|
|
"archivebox_username": self.user.profile.archivebox_username,
|
|
},
|
|
)
|
|
|
|
@property
|
|
def logdata(self) -> Optional[logdata.BaseLogData]:
|
|
if self.media_obj:
|
|
logdata_cls = self.media_obj.logdata_cls
|
|
else:
|
|
logdata_cls = logdata.BaseLogData
|
|
|
|
log_dict = self.log
|
|
if isinstance(self.log, str):
|
|
# There's nothing stopping django from saving a string in a JSONField :(
|
|
logger.warning(
|
|
"[scrobbles] Received string in JSON data in log",
|
|
extra={"log": self.log},
|
|
)
|
|
log_dict = json.loads(self.log)
|
|
|
|
if not log_dict:
|
|
log_dict = {}
|
|
|
|
try:
|
|
return logdata_cls(**log_dict)
|
|
except ParseError as e:
|
|
logger.warning(
|
|
"Could not parse log data",
|
|
extra={
|
|
"log_dict": log_dict,
|
|
"scrobble_id": self.id,
|
|
"error": e,
|
|
},
|
|
)
|
|
return logdata_cls()
|
|
except TypeError as e:
|
|
return logdata_cls()
|
|
|
|
def redirect_url(self, user_id) -> str:
|
|
user = User.objects.filter(id=user_id).first()
|
|
redirect_url = self.media_obj.get_absolute_url()
|
|
|
|
if (
|
|
self.media_type == self.MediaType.WEBPAGE
|
|
and user
|
|
and user.profile.redirect_to_webpage
|
|
):
|
|
logger.info(f"Redirecting to {self.media_obj.url}")
|
|
redirect_url = self.media_obj.url
|
|
|
|
if (
|
|
self.media_type == self.MediaType.VIDEO
|
|
and self.media_obj.youtube_id
|
|
):
|
|
redirect_url = self.media_obj.youtube_link
|
|
|
|
return redirect_url
|
|
|
|
@property
|
|
def tzinfo(self):
|
|
return ZoneInfo(self.timezone)
|
|
|
|
@property
|
|
def local_timestamp(self):
|
|
return timezone.localtime(self.timestamp, timezone=self.tzinfo)
|
|
|
|
@property
|
|
def local_stop_timestamp(self):
|
|
if self.stop_timestamp:
|
|
return timezone.localtime(
|
|
self.stop_timestamp, timezone=self.tzinfo
|
|
)
|
|
|
|
@property
|
|
def scrobble_media_key(self) -> str:
|
|
return media_class_to_foreign_key(self.media_type) + "_id"
|
|
|
|
@property
|
|
def status(self) -> str:
|
|
if self.is_paused:
|
|
return "paused"
|
|
if self.played_to_completion:
|
|
return "finished"
|
|
if self.in_progress:
|
|
return "in-progress"
|
|
return "zombie"
|
|
|
|
@property
|
|
def is_stale(self) -> bool:
|
|
"""Mark scrobble as stale if it's been more than an hour since it was updated
|
|
|
|
Effectively, this allows 'resuming' a video scrobble within an hour of starting it.
|
|
"""
|
|
is_stale = False
|
|
now = timezone.now()
|
|
seconds_since_last_update = (now - self.modified).seconds
|
|
if seconds_since_last_update >= self.media_obj.SECONDS_TO_STALE:
|
|
is_stale = True
|
|
return is_stale
|
|
|
|
@property
|
|
def previous(self) -> "Scrobble":
|
|
return (
|
|
self.media_obj.scrobble_set.order_by("-timestamp")
|
|
.filter(timestamp__lt=self.timestamp)
|
|
.first()
|
|
)
|
|
|
|
@property
|
|
def next(self) -> "Scrobble":
|
|
return (
|
|
self.media_obj.scrobble_set.order_by("timestamp")
|
|
.filter(timestamp__gt=self.timestamp)
|
|
.first()
|
|
)
|
|
|
|
@property
|
|
def previous_by_media(self) -> "Scrobble":
|
|
return (
|
|
Scrobble.objects.filter(
|
|
media_type=self.media_type,
|
|
user=self.user,
|
|
timestamp__lt=self.timestamp,
|
|
)
|
|
.order_by("-timestamp")
|
|
.first()
|
|
)
|
|
|
|
@property
|
|
def next_by_media(self) -> "Scrobble":
|
|
return (
|
|
Scrobble.objects.filter(
|
|
media_type=self.media_type,
|
|
user=self.user,
|
|
timestamp__gt=self.timestamp,
|
|
)
|
|
.order_by("-timestamp")
|
|
.first()
|
|
)
|
|
|
|
@property
|
|
def previous_by_user(self) -> "Scrobble":
|
|
return (
|
|
Scrobble.objects.order_by("-timestamp")
|
|
.filter(timestamp__lt=self.timestamp)
|
|
.first()
|
|
)
|
|
|
|
@property
|
|
def next_by_user(self) -> "Scrobble":
|
|
return (
|
|
Scrobble.objects.order_by("-timestamp")
|
|
.filter(timestamp__gt=self.timestamp)
|
|
.first()
|
|
)
|
|
|
|
@property
|
|
def session_pages_read(self) -> Optional[int]:
|
|
pages_read = 0
|
|
if self.log and isinstance(self.log, dict):
|
|
pages_read = self.log.get("pages_read", 0)
|
|
return pages_read
|
|
|
|
@property
|
|
def is_long_play(self) -> bool:
|
|
return self.media_obj.__class__.__name__ in LONG_PLAY_MEDIA.values()
|
|
|
|
@property
|
|
def elapsed_time(self) -> int | None:
|
|
if self.played_to_completion:
|
|
if self.playback_position_seconds:
|
|
return self.playback_position_seconds
|
|
if self.media_obj and self.media_obj.run_time_seconds:
|
|
return self.media_obj.run_time_seconds
|
|
|
|
return (timezone.now() - self.timestamp).seconds
|
|
|
|
@property
|
|
def percent_played(self) -> int:
|
|
if not self.media_obj:
|
|
return 0
|
|
|
|
if self.media_obj and not self.media_obj.run_time_seconds:
|
|
return 100
|
|
|
|
if not self.playback_position_seconds and self.played_to_completion:
|
|
return 100
|
|
|
|
playback_seconds = self.playback_position_seconds
|
|
if not playback_seconds:
|
|
playback_seconds = self.elapsed_time
|
|
|
|
run_time_secs = self.media_obj.run_time_seconds
|
|
percent = int((playback_seconds / run_time_secs) * 100)
|
|
|
|
if self.is_long_play:
|
|
long_play_secs = 0
|
|
if self.previous and not self.previous.long_play_complete:
|
|
long_play_secs = self.previous.long_play_seconds or 0
|
|
percent = int(
|
|
((playback_seconds + long_play_secs) / run_time_secs) * 100
|
|
)
|
|
|
|
return percent
|
|
|
|
@property
|
|
def probably_still_in_progress(self) -> bool:
|
|
"""Add our start time to our media run time to get when we expect to
|
|
|
|
Audio tracks should be given a second or two of grace, videos should
|
|
be given closer to 30 minutes, because the odds of watching it back to
|
|
back are very slim.
|
|
|
|
"""
|
|
is_in_progress = False
|
|
padding_seconds = MEDIA_END_PADDING_SECONDS.get(self.media_type)
|
|
if not padding_seconds:
|
|
return is_in_progress
|
|
|
|
if not self.media_obj:
|
|
logger.info(
|
|
"[scrobbling] scrobble has no media obj",
|
|
extra={
|
|
"media_id": self.media_obj,
|
|
"scrobble_id": self.id,
|
|
"media_type": self.media_type,
|
|
"probably_still_in_progress": is_in_progress,
|
|
},
|
|
)
|
|
return is_in_progress
|
|
|
|
if not self.media_obj.run_time_seconds:
|
|
logger.info(
|
|
"[scrobbling] media has no run time seconds, cannot calculate end",
|
|
extra={
|
|
"media_id": self.media_obj.id,
|
|
"scrobble_id": self.id,
|
|
"media_type": self.media_type,
|
|
"probably_still_in_progress": is_in_progress,
|
|
},
|
|
)
|
|
return is_in_progress
|
|
|
|
expected_end = self.timestamp + datetime.timedelta(
|
|
seconds=self.media_obj.run_time_seconds
|
|
)
|
|
expected_end_padded = expected_end + datetime.timedelta(
|
|
seconds=padding_seconds
|
|
)
|
|
# Take our start time, add our media length and an extra 30 min (1800s) is it still in the future? keep going
|
|
is_in_progress = expected_end_padded > pendulum.now()
|
|
logger.info(
|
|
"[scrobbling] checking if we're probably still playing",
|
|
extra={
|
|
"media_id": self.media_obj.id,
|
|
"scrobble_id": self.id,
|
|
"media_type": self.media_type,
|
|
"probably_still_in_progress": is_in_progress,
|
|
},
|
|
)
|
|
return is_in_progress
|
|
|
|
@property
|
|
def can_be_updated(self) -> bool:
|
|
if self.media_obj.__class__.__name__ in LONG_PLAY_MEDIA.values() and self.source != "readcomicsonline.ru":
|
|
logger.info(
|
|
"[scrobbling] cannot be updated, long play media",
|
|
extra={
|
|
"media_id": self.media_obj.id,
|
|
"scrobble_id": self.id,
|
|
"media_type": self.media_type,
|
|
},
|
|
)
|
|
return False
|
|
|
|
if self.percent_played >= 100 and not self.probably_still_in_progress:
|
|
logger.info(
|
|
"[scrobbling] cannot be updated, existing scrobble is 100% played",
|
|
extra={
|
|
"media_id": self.media_obj.id,
|
|
"scrobble_id": self.id,
|
|
"media_type": self.media_type,
|
|
},
|
|
)
|
|
return False
|
|
|
|
if self.is_stale:
|
|
logger.info(
|
|
"[scrobbling] cannot be udpated, stale",
|
|
extra={
|
|
"media_id": self.media_obj.id,
|
|
"scrobble_id": self.id,
|
|
"media_type": self.media_type,
|
|
},
|
|
)
|
|
return False
|
|
logger.info(
|
|
"[scrobbling] can be updated",
|
|
extra={
|
|
"media_id": self.media_obj.id,
|
|
"scrobble_id": self.id,
|
|
"media_type": self.media_type,
|
|
},
|
|
)
|
|
return True
|
|
|
|
@classmethod
|
|
def by_date(cls, media_type: str = "Track"):
|
|
cls.objects.filter(media_type=media_type).values(
|
|
"timestamp__date"
|
|
).annotate(count=models.Count("id")).values(
|
|
"timestamp__date", "count"
|
|
).order_by(
|
|
"-count",
|
|
)
|
|
|
|
@property
|
|
def media_obj(self):
|
|
media_obj = None
|
|
if self.video:
|
|
media_obj = self.video
|
|
if self.track:
|
|
media_obj = self.track
|
|
if self.podcast_episode:
|
|
media_obj = self.podcast_episode
|
|
if self.sport_event:
|
|
media_obj = self.sport_event
|
|
if self.book:
|
|
media_obj = self.book
|
|
if self.video_game:
|
|
media_obj = self.video_game
|
|
if self.board_game:
|
|
media_obj = self.board_game
|
|
if self.geo_location:
|
|
media_obj = self.geo_location
|
|
if self.web_page:
|
|
media_obj = self.web_page
|
|
if self.life_event:
|
|
media_obj = self.life_event
|
|
if self.mood:
|
|
media_obj = self.mood
|
|
if self.brick_set:
|
|
media_obj = self.brick_set
|
|
if self.trail:
|
|
media_obj = self.trail
|
|
if self.beer:
|
|
media_obj = self.beer
|
|
if self.puzzle:
|
|
media_obj = self.puzzle
|
|
if self.task:
|
|
media_obj = self.task
|
|
if self.food:
|
|
media_obj = self.food
|
|
return media_obj
|
|
|
|
def __str__(self):
|
|
return f"Scrobble of {self.media_obj} ({self.timestamp})"
|
|
|
|
def calc_reading_duration(self) -> int:
|
|
duration = 0
|
|
if self.logdata.page_data:
|
|
for k, v in self.logdata.page_data.items():
|
|
duration += v.get("duration")
|
|
return duration
|
|
|
|
def calc_pages_read(self) -> int:
|
|
pages_read = 0
|
|
if self.logdata.page_data:
|
|
pages = [int(k) for k in self.logdata.page_data.keys()]
|
|
pages.sort()
|
|
if len(pages) == 1:
|
|
pages_read = 1
|
|
elif len(pages) >= 2:
|
|
pages_read += pages[-1] - pages[0]
|
|
else:
|
|
pages_read = pages[-1] - pages[0]
|
|
return pages_read
|
|
|
|
@property
|
|
def last_page_read(self) -> int:
|
|
last_page = 0
|
|
if self.logdata.page_data:
|
|
pages = [int(k) for k in self.logdata.page_data.keys()]
|
|
pages.sort()
|
|
last_page = pages[-1]
|
|
return last_page
|
|
|
|
@property
|
|
def get_media_source_url(self) -> str:
|
|
url = ""
|
|
if self.media_type == "Website":
|
|
url = self.media_obj.url
|
|
if self.media_type == "Task":
|
|
url = self.media_obj.source_url_for_user(self.user)
|
|
return url
|
|
|
|
@classmethod
|
|
def create_or_update(
|
|
cls, media, user_id: int, scrobble_data: dict, **kwargs
|
|
) -> "Scrobble":
|
|
key = media_class_to_foreign_key(media.__class__.__name__)
|
|
media_query = models.Q(**{key: media})
|
|
scrobble_data[key + "_id"] = media.id
|
|
skip_in_progress_check = kwargs.get("skip_in_progress_check", False)
|
|
read_log_page = kwargs.get("read_log_page", None)
|
|
|
|
|
|
# Find our last scrobble of this media item (track, video, etc)
|
|
scrobble = (
|
|
cls.objects.filter(
|
|
media_query,
|
|
user_id=user_id,
|
|
)
|
|
.order_by("-timestamp")
|
|
.first()
|
|
)
|
|
source = scrobble_data.get("source", "Vrobbler")
|
|
mtype = media.__class__.__name__
|
|
|
|
# GeoLocations are a special case scrobble
|
|
if mtype == cls.MediaType.GEO_LOCATION:
|
|
logger.warning(
|
|
f"[create_or_update] geoloc requires create_or_update_location"
|
|
)
|
|
scrobble = cls.create_or_update_location(
|
|
media, scrobble_data, user_id
|
|
)
|
|
return scrobble
|
|
|
|
if not skip_in_progress_check or read_log_page:
|
|
logger.info(
|
|
f"[create_or_update] check for existing scrobble to update ",
|
|
extra={
|
|
"scrobble_id": scrobble.id if scrobble else None,
|
|
"media_type": mtype,
|
|
"media_id": media.id,
|
|
"scrobble_data": scrobble_data,
|
|
},
|
|
)
|
|
scrobble_data["playback_status"] = scrobble_data.pop(
|
|
"status", None
|
|
)
|
|
# If it's marked as stopped, send it through our update mechanism, which will complete it
|
|
if scrobble and (
|
|
scrobble.can_be_updated
|
|
or (read_log_page and scrobble.can_be_updated)
|
|
or scrobble_data["playback_status"] == "stopped"
|
|
):
|
|
if read_log_page:
|
|
page_list = scrobble.log.get("page_data", [])
|
|
if page_list:
|
|
for page in page_list:
|
|
if not page.get("end_ts", None):
|
|
page["end_ts"] = int(timezone.now().timestamp())
|
|
page["duration"] = page["end_ts"] - page.get("start_ts")
|
|
|
|
page_list.append(
|
|
BookPageLogData(
|
|
page_number=read_log_page,
|
|
start_ts=int(timezone.now().timestamp())
|
|
)
|
|
)
|
|
scrobble.log["page_data"] = page_list
|
|
scrobble.save(update_fields=["log"])
|
|
elif "log" in scrobble_data.keys() and scrobble.log:
|
|
scrobble_data["log"] = scrobble.log | scrobble_data["log"]
|
|
return scrobble.update(scrobble_data)
|
|
|
|
# Discard status before creating
|
|
scrobble_data.pop("playback_status")
|
|
|
|
if read_log_page:
|
|
scrobble_data["log"] = BookLogData(page_data=[BookPageLogData(page_number=read_log_page, start_ts=int(timezone.now().timestamp()))])
|
|
|
|
logger.info(
|
|
f"[scrobbling] creating new scrobble",
|
|
extra={
|
|
"scrobble_id": scrobble.id if scrobble else None,
|
|
"media_type": mtype,
|
|
"media_id": media.id,
|
|
"source": source,
|
|
},
|
|
)
|
|
|
|
scrobble = cls.create(scrobble_data)
|
|
return scrobble
|
|
|
|
@classmethod
|
|
def create_or_update_location(
|
|
cls, location: GeoLocation, scrobble_data: dict, user_id: int
|
|
) -> "Scrobble":
|
|
"""Location is special type, where the current scrobble for a user is always the
|
|
current active scrobble, and we only finish it a move on if we get a new location
|
|
that is far enough (and far enough over the last three past scrobbles) to have
|
|
actually moved.
|
|
"""
|
|
key = media_class_to_foreign_key(location.__class__.__name__)
|
|
scrobble_data[key + "_id"] = location.id
|
|
|
|
scrobble = (
|
|
cls.objects.filter(
|
|
media_type=cls.MediaType.GEO_LOCATION,
|
|
user_id=user_id,
|
|
timestamp__lte=scrobble_data.get("timestamp"),
|
|
)
|
|
.order_by("-timestamp")
|
|
.first()
|
|
)
|
|
|
|
logger.info(
|
|
f"[scrobbling] fetching last location scrobble",
|
|
extra={
|
|
"scrobble_id": scrobble.id if scrobble else None,
|
|
"media_type": cls.MediaType.GEO_LOCATION,
|
|
"media_id": location.id,
|
|
"scrobble_data": scrobble_data,
|
|
},
|
|
)
|
|
|
|
if not scrobble:
|
|
logger.info(
|
|
f"[scrobbling] finished - no existing location scrobbles found",
|
|
extra={
|
|
"media_id": location.id,
|
|
"media_type": cls.MediaType.GEO_LOCATION,
|
|
},
|
|
)
|
|
return cls.create(scrobble_data)
|
|
|
|
if scrobble.media_obj == location:
|
|
logger.info(
|
|
f"[scrobbling] finished - same location - not moved",
|
|
extra={
|
|
"media_type": cls.MediaType.GEO_LOCATION,
|
|
"media_id": location.id,
|
|
"scrobble_id": scrobble.id,
|
|
"scrobble_media_id": scrobble.media_obj.id,
|
|
},
|
|
)
|
|
return scrobble
|
|
|
|
has_moved = location.has_moved(scrobble.media_obj)
|
|
logger.info(
|
|
f"[scrobbling] checking - has last location has moved?",
|
|
extra={
|
|
"scrobble_id": scrobble.id,
|
|
"scrobble_media_id": scrobble.media_obj.id,
|
|
"media_type": cls.MediaType.GEO_LOCATION,
|
|
"media_id": location.id,
|
|
"has_moved": has_moved,
|
|
},
|
|
)
|
|
if not has_moved:
|
|
logger.info(
|
|
f"[scrobbling] finished - not from old location - not moved",
|
|
extra={
|
|
"scrobble_id": scrobble.id,
|
|
"media_id": location.id,
|
|
"media_type": cls.MediaType.GEO_LOCATION,
|
|
"old_media__id": scrobble.media_obj.id,
|
|
},
|
|
)
|
|
return scrobble
|
|
|
|
if existing_locations := location.in_proximity(named=True):
|
|
existing_location = existing_locations.first()
|
|
ts = int(pendulum.now().timestamp())
|
|
scrobble.log[
|
|
ts
|
|
] = f"Location {location.id} too close to this scrobble"
|
|
scrobble.save(update_fields=["log"])
|
|
logger.info(
|
|
f"[scrobbling] finished - found existing named location",
|
|
extra={
|
|
"media_id": location.id,
|
|
"media_type": cls.MediaType.GEO_LOCATION,
|
|
"old_media_id": existing_location.id,
|
|
},
|
|
)
|
|
return scrobble
|
|
|
|
scrobble.stop(force_finish=True)
|
|
scrobble = cls.create(scrobble_data)
|
|
logger.info(
|
|
f"[scrobbling] finished - created for location",
|
|
extra={
|
|
"scrobble_id": scrobble.id,
|
|
"media_id": location.id,
|
|
"scrobble_data": scrobble_data,
|
|
"media_type": cls.MediaType.GEO_LOCATION,
|
|
"source": scrobble_data.get("source"),
|
|
},
|
|
)
|
|
return scrobble
|
|
|
|
def update(self, scrobble_data: dict) -> "Scrobble":
|
|
# Status is a field we get from Mopidy, which refuses to poll us
|
|
playback_status = scrobble_data.pop("playback_status", None)
|
|
|
|
logger.info(
|
|
"[update] called",
|
|
extra={
|
|
"scrobble_id": self.id,
|
|
"scrobble_data": scrobble_data,
|
|
"media_type": self.media_type,
|
|
"playback_status": playback_status,
|
|
},
|
|
)
|
|
if self.beyond_completion_percent:
|
|
playback_status = "stopped"
|
|
|
|
if playback_status == "stopped":
|
|
self.stop()
|
|
if playback_status == "paused":
|
|
self.pause()
|
|
if playback_status == "resumed":
|
|
self.resume()
|
|
|
|
if playback_status != "resumed":
|
|
scrobble_data["stop_timestamp"] = (
|
|
scrobble_data.pop("timestamp", None) or timezone.now()
|
|
)
|
|
|
|
# timestamp should be more-or-less immutable
|
|
scrobble_data.pop("timestamp", None)
|
|
|
|
update_fields = []
|
|
for key, value in scrobble_data.items():
|
|
setattr(self, key, value)
|
|
update_fields.append(key)
|
|
self.save(update_fields=update_fields)
|
|
|
|
return self
|
|
|
|
@classmethod
|
|
def create(
|
|
cls,
|
|
scrobble_data: dict,
|
|
) -> "Scrobble":
|
|
scrobble = cls.objects.create(**scrobble_data)
|
|
ScrobbleNtfyNotification(scrobble).send()
|
|
return scrobble
|
|
|
|
def stop(self, timestamp=None, force_finish=False) -> None:
|
|
self.stop_timestamp = timestamp or timezone.now()
|
|
self.played_to_completion = True
|
|
self.in_progress = False
|
|
|
|
if not self.playback_position_seconds:
|
|
self.playback_position_seconds = int(
|
|
(self.stop_timestamp - self.timestamp).total_seconds()
|
|
)
|
|
|
|
self.save(
|
|
update_fields=[
|
|
"in_progress",
|
|
"played_to_completion",
|
|
"stop_timestamp",
|
|
"playback_position_seconds",
|
|
]
|
|
)
|
|
|
|
class_name = self.media_obj.__class__.__name__
|
|
if class_name in LONG_PLAY_MEDIA.values():
|
|
self.finish_long_play()
|
|
|
|
if class_name == "Book":
|
|
self.calculate_reading_stats()
|
|
|
|
logger.info(
|
|
f"[scrobbling] stopped",
|
|
extra={
|
|
"scrobble_id": self.id,
|
|
"media_id": self.media_obj.id,
|
|
"media_type": self.media_type,
|
|
"source": self.source,
|
|
},
|
|
)
|
|
|
|
def pause(self) -> None:
|
|
if self.is_paused:
|
|
logger.warning(f"{self.id} - already paused - {self.source}")
|
|
return
|
|
self.is_paused = True
|
|
self.save(update_fields=["is_paused"])
|
|
logger.info(
|
|
f"[scrobbling] paused",
|
|
extra={
|
|
"scrobble_id": self.id,
|
|
"media_type": self.media_type,
|
|
"source": self.source,
|
|
},
|
|
)
|
|
|
|
def resume(self) -> None:
|
|
if self.is_paused or not self.in_progress:
|
|
self.is_paused = False
|
|
self.in_progress = True
|
|
self.save(update_fields=["is_paused", "in_progress"])
|
|
logger.info(
|
|
f"[scrobbling] resumed",
|
|
extra={
|
|
"scrobble_id": self.id,
|
|
"media_type": self.media_type,
|
|
"source": self.source,
|
|
},
|
|
)
|
|
|
|
def cancel(self) -> None:
|
|
self.delete()
|
|
|
|
def update_ticks(self, data) -> None:
|
|
self.playback_position_seconds = data.get("playback_position_seconds")
|
|
self.save(update_fields=["playback_position_seconds"])
|
|
|
|
def finish_long_play(self):
|
|
seconds_elapsed = (timezone.now() - self.timestamp).seconds
|
|
past_seconds = 0
|
|
|
|
# Set our playback seconds, and calc long play seconds
|
|
self.playback_position_seconds = seconds_elapsed
|
|
if self.previous:
|
|
past_seconds = self.previous.long_play_seconds or 0
|
|
|
|
self.long_play_seconds = past_seconds + seconds_elapsed
|
|
|
|
# Long play scrobbles are always finished when we say they are
|
|
self.played_to_completion = True
|
|
|
|
self.save(
|
|
update_fields=[
|
|
"playback_position_seconds",
|
|
"played_to_completion",
|
|
"long_play_seconds",
|
|
]
|
|
)
|
|
logger.info(
|
|
f"[scrobbling] finishing long play",
|
|
extra={
|
|
"scrobble_id": self.id,
|
|
},
|
|
)
|
|
|
|
@property
|
|
def beyond_completion_percent(self) -> bool:
|
|
"""Returns true if our media is beyond our completion percent, unless
|
|
our type is geolocation in which case we always return false
|
|
"""
|
|
beyond_completion = (
|
|
self.percent_played >= self.media_obj.COMPLETION_PERCENT
|
|
)
|
|
|
|
if self.media_type == "GeoLocation":
|
|
logger.info(
|
|
f"[scrobbling] locations are ONLY completed when new one is created",
|
|
extra={
|
|
"scrobble_id": self.id,
|
|
"media_type": self.media_type,
|
|
"beyond_completion": beyond_completion,
|
|
},
|
|
)
|
|
beyond_completion = False
|
|
|
|
return beyond_completion
|
|
|
|
def calculate_reading_stats(self, commit=True):
|
|
# --- Sort safely by numeric page_number ---
|
|
def safe_page_number(entry):
|
|
try:
|
|
return int(getattr("page_number", entry), 0)
|
|
except (ValueError, TypeError):
|
|
return float("inf") # push invalid entries to the end
|
|
|
|
page_data = self.log.get("page_data")
|
|
|
|
if not page_data:
|
|
logger.warning("No page data found to calculate")
|
|
return
|
|
|
|
if isinstance(page_data, dict):
|
|
logger.warning("Page data is dict, migrate koreader data")
|
|
return
|
|
|
|
page_data.sort(key=safe_page_number)
|
|
|
|
# --- Extract valid numeric page numbers ---
|
|
valid_pages = []
|
|
for page in page_data:
|
|
try:
|
|
valid_pages.append(int(page["page_number"]))
|
|
except (ValueError, TypeError):
|
|
continue
|
|
|
|
# --- Compute stats ---
|
|
if valid_pages:
|
|
self.log["page_start"] = min(valid_pages)
|
|
self.log["page_end"] = max(valid_pages)
|
|
self.log["pages_read"] = len(set(valid_pages))
|
|
|
|
if commit:
|
|
self.save(update_fields=["log"])
|