from zoneinfo import ZoneInfo import pendulum from django.utils import timezone import logging from django.conf import settings from django.contrib.auth import get_user_model from django.db import models from django.utils.functional import cached_property from django_extensions.db.models import TimeStampedModel from encrypted_field import EncryptedField from profiles.constants import PRETTY_TIMEZONE_CHOICES VISIBILITY_CHOICES = ( ("public", "Public"), ("shared", "Shared"), ("private", "Private"), ) User = get_user_model() BNULL = {"blank": True, "null": True} logger = logging.getLogger(__name__) class WeighUnit(models.TextChoices): METRIC = "metric", "Metric (kg, cm)" IMPERIAL = "imperial", "Imperial (lbs, in)" class UserProfile(TimeStampedModel): user = models.OneToOneField(User, on_delete=models.CASCADE, related_name="profile") timezone = models.CharField( max_length=255, choices=PRETTY_TIMEZONE_CHOICES, default="UTC" ) timezone_change_log = models.TextField(**BNULL) lastfm_username = models.CharField(max_length=255, **BNULL) lastfm_password = EncryptedField(**BNULL) lastfm_auto_import = models.BooleanField(default=False) retroarch_path = models.CharField(max_length=255, **BNULL) retroarch_auto_import = models.BooleanField(default=False) archivebox_username = models.CharField(max_length=255, **BNULL) archivebox_password = EncryptedField(**BNULL) archivebox_url = models.CharField(max_length=255, **BNULL) task_context_tags_str = models.CharField(max_length=255, **BNULL) bgstats_id = models.CharField(max_length=255, **BNULL) bgg_username = models.CharField(max_length=255, **BNULL) lichess_username = models.CharField(max_length=255, **BNULL) todoist_auth_key = EncryptedField(**BNULL) todoist_state = EncryptedField(**BNULL) todoist_user_id = models.CharField(max_length=100, **BNULL) webdav_url = models.CharField(max_length=255, **BNULL) webdav_user = models.CharField(max_length=255, **BNULL) webdav_pass = EncryptedField(**BNULL) webdav_auto_import = models.BooleanField(default=False) imap_url = models.CharField(max_length=255, **BNULL) imap_user = models.CharField(max_length=255, **BNULL) imap_pass = EncryptedField(**BNULL) imap_auto_import = models.BooleanField(default=False) mood_checkin_enabled = models.BooleanField(default=False) mood_checkin_frequency = models.CharField(max_length=20, default="hourly") ntfy_url = models.CharField(max_length=255, **BNULL) ntfy_enabled = models.BooleanField(default=False) mopidy_api_url = models.CharField(max_length=255, **BNULL) favorites_mopidy_playlist = models.CharField( max_length=255, **BNULL, help_text="Playlist name (e.g. 'Favorites'). Will map to m3u:Favorites.m3u8", ) monthly_mopidy_playlist_pattern = models.CharField( max_length=255, **BNULL, help_text="Django date format pattern for monthly playlists (e.g. 'Y F')", ) redirect_to_webpage = models.BooleanField(default=True) enable_public_widgets = models.BooleanField(default=False) widget_custom_css = models.TextField(**BNULL) default_scrobble_visibility = models.CharField( max_length=10, choices=VISIBILITY_CHOICES, default="private", ) home_scrobble_limit = models.IntegerField(default=20) weigh_in_units = models.CharField( max_length=16, choices=WeighUnit.choices, default=WeighUnit.METRIC, ) def __str__(self): return f"User profile for {self.user}" @property def tzinfo(self): return ZoneInfo(self.timezone) def save(self, *args, **kwargs): if not self._state.adding: old_instance = UserProfile.objects.get(pk=self.pk) is_timezone_change = self.timezone != old_instance.timezone if is_timezone_change: logger.info( "Updating timezone changelog for user", extra={"profile_id": self.id}, ) previous_changes = old_instance.timezone_change_log now = timezone.now().replace(microsecond=0) new_log = f"{self.timezone} - {now}" if previous_changes: new_log = previous_changes + f"\n{new_log}" self.timezone_change_log = new_log super(UserProfile, self).save(*args, **kwargs) @property def historic_timezone_changes(self) -> list: """Return a list of datetimes with timezones for the specific changed time""" history = [pendulum.datetime(1900, 1, 1, 0, 0, 0, tz=self.tzinfo.key)] if self.timezone_change_log: for change in self.timezone_change_log.split("\n"): if " - " in change: tz, date = change.split(" - ") history.append(pendulum.parse(date).in_timezone(tz)) return history def get_timestamp_with_tz(self, timestamp): timezone = self.tzinfo if self.timezone_change_log: change_list = self.historic_timezone_changes for idx, start in enumerate(change_list): try: end = change_list[idx + 1] except IndexError: end = None if end: if start <= timestamp.replace(tzinfo=end.timezone) <= end: timezone = start.timezone else: if start <= timestamp.replace(tzinfo=start.timezone): timezone = start.timezone return timestamp.replace(tzinfo=timezone) def adjust_timezone_of_scrobbles(self, commit=False): current_dt = None scrobbles_to_change_qs_list = [] for boundry_dt in self.historic_timezone_changes: if current_dt and boundry_dt: logger.info( f"Checking for scrobbles between {current_dt} and {boundry_dt} to update to {current_dt.tzinfo.name}" ) scrobbles = self.user.scrobble_set.filter( timestamp__gte=current_dt, timestamp__lt=boundry_dt, ).exclude(timezone=current_dt.tzinfo.name) scrobbles_to_change_qs_list.append(scrobbles) logger.info( f"Updating {scrobbles.count()} scrobble timezones to {current_dt.tzinfo.name}" ) if commit: scrobbles.update(timezone=current_dt.tzinfo.name) current_dt = boundry_dt return scrobbles_to_change_qs_list @cached_property def task_context_tags(self) -> list[str]: tag_list = settings.DEFAULT_TASK_CONTEXT_TAGS tags = "" if self.task_context_tags_str: tags = self.task_context_tags_str tag_list = [t.strip().capitalize() for t in tags.split(",")] return tag_list