import logging from datetime import date, timedelta import arrow from django.core.serializers.json import DjangoJSONEncoder from django.db import models from icalevnt import icalevents from .parser import process_event_list logger = logging.getLogger(__name__) def _convert_arrow_to_datetime(event): event["start"] = event["start"].datetime event["end"] = event["end"].datetime return event class EventsJSONField(models.JSONField): """ JSONField for lists of events which converts `begin` and `end` to datetime on load from DB. """ def from_db_value(self, value, expression, connection): value = super().from_db_value(value, expression, connection) if value: for event in value: event["start"] = arrow.get(event.get("start")).datetime event["end"] = arrow.get(event["end"]).datetime return value class Calendar(models.Model): CURRENT_NUM = 6 url = models.URLField() event_hash = models.CharField(max_length=256, null=True) last_update = models.DateTimeField(null=True) past_events = EventsJSONField(encoder=DjangoJSONEncoder, null=True) future_events = EventsJSONField(encoder=DjangoJSONEncoder, null=True) def current_events(self): return self.future_events[: self.CURRENT_NUM] def handle_event_list(self, event_list): event_list_hash = str(hash(str(event_list))) if event_list_hash != self.event_hash: past, future = process_event_list(event_list) self.past_events = past self.future_events = future self.event_hash = event_list_hash self.last_update = arrow.utcnow().datetime self.save() def update_source(self): event_list = icalevents.events( url=self.url, start=date.today() - timedelta(days=30), end=date.today() + timedelta(days=60), ) self.handle_event_list(event_list) class CalendarMixin(models.Model): """ Mixin to be used in other models, like site settings, which adds relation to Calendar. """ calendar_url = models.URLField( "URL kalendáře ve formátu iCal", blank=True, null=True ) calendar = models.ForeignKey( Calendar, null=True, blank=True, on_delete=models.PROTECT ) class Meta: abstract = True def save(self, *args, **kwargs): # create or update related Calendar if self.calendar_url: if self.calendar: if self.calendar.url != self.calendar_url: self.calendar.url = self.calendar_url self.calendar.save() else: self.calendar = Calendar.objects.create(url=self.calendar_url) try: self.calendar.update_source() except: logger.error( "Calendar update failed for %s", self.calendar.url, exc_info=True ) # delete related Calendar when URL is cleared if not self.calendar_url and self.calendar: self.calendar = None super().save(*args, **kwargs)