import logging from datetime import date, timedelta from pathlib import Path import arrow from django.core.serializers.json import DjangoJSONEncoder from django.db import models from icalevents import icalevents from wagtail.admin.panels import FieldPanel 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)