Add automatic plan model and functionality to track changes

aleksis/apps/chronos/apps.py

+from django.db import transaction
+
+from reversion.signals import post_revision_commit
+
 from aleksis.core.util.apps import AppConfig
 
 
@@ -18,3 +22,13 @@ class ChronosConfig(AppConfig):
         ([2019], "Tom Teichler", "tom.teichler@teckids.org"),
         ([2021], "Lloyd Meins", "meinsll@katharineum.de"),
     )
+
+    def ready(self):
+        super().ready()
+
+        from .util.change_tracker import handle_new_revision  # noqa
+
+        def _handle_post_revision_commit(sender, revision, versions, **kwargs):
+            transaction.on_commit(lambda: handle_new_revision.delay(revision.pk))
+
+        post_revision_commit.connect(_handle_post_revision_commit)

aleksis/apps/chronos/migrations/0009_automatic_plan.py

+# Generated by Django 3.2.4 on 2021-07-23 19:48
+
+import django.contrib.sites.managers
+import django.core.validators
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('reversion', '0001_squashed_0004_auto_20160611_1202'),
+        ('sites', '0002_alter_domain_unique'),
+        ('chronos', '0008_unique_constraints'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='AutomaticPlan',
+            fields=[
+                ('id', models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('extended_data', models.JSONField(default=dict, editable=False)),
+                ('slug', models.SlugField(help_text='This will be used for the name of the PDF file with the generated plan.', verbose_name='Slug')),
+                ('name', models.CharField(max_length=255, verbose_name='Name')),
+                ('number_of_days', models.PositiveIntegerField(default=1, validators=[django.core.validators.MinValueValidator(1)], verbose_name='Number of days shown in the plan')),
+                ('show_header_box', models.BooleanField(default=True, help_text='The header box shows affected teachers/groups.', verbose_name='Show header box')),
+                ('last_update', models.DateTimeField(blank=True, null=True, verbose_name='Date and time of the last update')),
+                ('last_update_triggered_manually', models.BooleanField(default=False, verbose_name='Was the last update triggered manually?')),
+                ('current_file', models.FileField(blank=True, null=True, upload_to='chronos/plan_pdfs/', verbose_name='Current file')),
+                ('last_revision', models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.SET_NULL, to='reversion.revision', verbose_name='Revision which triggered the last update')),
+                ('site', models.ForeignKey(default=1, editable=False, on_delete=django.db.models.deletion.CASCADE, to='sites.site')),
+            ],
+            options={
+                'verbose_name': 'Automatic plan',
+                'verbose_name_plural': 'Automatic plans',
+            },
+            managers=[
+                ('objects', django.contrib.sites.managers.CurrentSiteManager()),
+            ],
+        ),
+        migrations.AddConstraint(
+            model_name='automaticplan',
+            constraint=models.UniqueConstraint(fields=('site_id', 'slug'), name='site_slug'),
+        ),
+    ]

aleksis/apps/chronos/models.py

@@ -3,13 +3,18 @@
 from __future__ import annotations
 
 from datetime import date, datetime, time, timedelta
-from typing import Dict, Iterator, List, Optional, Tuple, Union
+from typing import Any, Dict, Iterable, Iterator, List, Optional, Tuple, Union
 
 from django.core.exceptions import ValidationError
+from django.core.files.base import ContentFile, File
+from django.core.files.storage import default_storage
+from django.core.validators import MinValueValidator
 from django.db import models
 from django.db.models import Max, Min, Q
 from django.db.models.functions import Coalesce
+from django.dispatch import receiver
 from django.forms import Media
+from django.template.loader import render_to_string
 from django.urls import reverse
 from django.utils import timezone
 from django.utils.formats import date_format
@@ -18,7 +23,10 @@ from django.utils.translation import gettext_lazy as _
 
 from cache_memoize import cache_memoize
 from calendarweek.django import CalendarWeek, i18n_day_abbr_choices_lazy, i18n_day_name_choices_lazy
+from celery.result import allow_join_result
+from celery.states import SUCCESS
 from colorfield.fields import ColorField
+from reversion.models import Revision, Version
 
 from aleksis.apps.chronos.managers import (
     AbsenceQuerySet,
@@ -45,6 +53,7 @@ from aleksis.apps.chronos.mixins import (
     WeekAnnotationMixin,
     WeekRelatedMixin,
 )
+from aleksis.apps.chronos.util.change_tracker import substitutions_changed
 from aleksis.apps.chronos.util.date import get_current_year
 from aleksis.apps.chronos.util.format import format_m2m
 from aleksis.core.managers import CurrentSiteManagerWithoutMigrations
@@ -53,8 +62,9 @@ from aleksis.core.mixins import (
     GlobalPermissionModel,
     SchoolTermRelatedExtensibleModel,
 )
-from aleksis.core.models import DashboardWidget, SchoolTerm
+from aleksis.core.models import DashboardWidget, PDFFile, SchoolTerm
 from aleksis.core.util.core_helpers import has_person
+from aleksis.core.util.pdf import generate_pdf_from_template
 
 
 class ValidityRange(ExtensibleModel):
@@ -1111,6 +1121,143 @@ class ExtraLesson(
         indexes = [models.Index(fields=["week", "year"], name="extra_lesson_week_year")]
 
 
+@receiver(substitutions_changed)
+def automatic_plan_signal_receiver(sender: Revision, versions: Iterable[Version], **kwargs):
+    """Check all automatic plans for updates after substitutions changed."""
+    for automatic_plan in AutomaticPlan.objects.all():
+        automatic_plan.check_update(sender, versions)
+
+
+class AutomaticPlan(ExtensibleModel):
+    slug = models.SlugField(
+        verbose_name=_("Slug"),
+        help_text=_("This will be used for the name of the PDF file with the generated plan."),
+    )
+    name = models.CharField(max_length=255, verbose_name=_("Name"))
+    number_of_days = models.PositiveIntegerField(
+        default=1,
+        validators=[MinValueValidator(1)],
+        verbose_name=_("Number of days shown in the plan"),
+    )
+    show_header_box = models.BooleanField(
+        default=True,
+        verbose_name=_("Show header box"),
+        help_text=_("The header box shows affected teachers/groups."),
+    )
+
+    last_revision = models.ForeignKey(
+        to=Revision,
+        on_delete=models.SET_NULL,
+        blank=True,
+        null=True,
+        verbose_name=_("Revision which triggered the last update"),
+    )
+    last_update = models.DateTimeField(
+        blank=True, null=True, verbose_name=_("Date and time of the last update")
+    )
+    last_update_triggered_manually = models.BooleanField(
+        default=False, verbose_name=_("Was the last update triggered manually?")
+    )
+
+    current_file = models.FileField(
+        upload_to="chronos/plan_pdfs/", null=True, blank=True, verbose_name=_("Current file")
+    )
+
+    @property
+    def current_start_day(self) -> date:
+        """Get first day which should be shown in the PDF."""
+        return TimePeriod.get_next_relevant_day(timezone.now().date(), datetime.now().time())
+
+    @property
+    def current_end_day(self) -> date:
+        """Get last day which should be shown in the PDF."""
+        return self.current_start_day + timedelta(days=self.number_of_days - 1)
+
+    def get_context_data(self) -> Dict[str, Any]:
+        """Get context data for generating the substitutions PDF."""
+        from aleksis.apps.chronos.views import get_substitutions_context_data
+
+        context = get_substitutions_context_data(
+            request=None,
+            is_print=True,
+            number_of_days=self.number_of_days,
+            show_header_box=self.show_header_box,
+        )
+
+        return context
+
+    def check_update(self, revision: Revision, versions: Iterable[Version]):
+        """Check if the PDF file has to be updated and do the update then."""
+        if not self.last_revision or (
+            self.last_revision != revision
+            and revision.date_created > self.last_revision.date_created
+        ):
+            update = False
+            for version in versions:
+                # Check if the changed object is relevant for the time period of the PDF file
+                if isinstance(version.object, Event):
+                    date_start = version.object.date_start
+                    date_end = version.object.date_end
+                else:
+                    date_start = date_end = version.object.date
+                if date_start <= self.current_end_day and date_end >= self.current_start_day:
+                    update = True
+                    break
+
+            if update:
+                self.update(triggered_manually=False)
+                self.last_revision = revision
+                self.save()
+
+    def update(self, triggered_manually: bool = True):
+        """Regenerate the PDF file with the substitutions plan."""
+        file_object, result = generate_pdf_from_template(
+            "chronos/substitutions_print.html", self.get_context_data()
+        )
+        with allow_join_result():
+            result.wait()
+            file_object.refresh_from_db()
+            if result.status == SUCCESS and file_object.file:
+                self.current_file.save("current.pdf", file_object.file.file)
+                self.last_update = timezone.now()
+                self.last_update_triggered_manually = triggered_manually
+                self.save()
+
+    def get_current_file(self) -> File:
+        """Get current PDF file."""
+        if not self.current_file:
+            self.update()
+        return self.current_file.file
+
+    @property
+    def filename(self) -> str:
+        """Get filename (without path) of the PDF file."""
+        return f"{self.slug}.pdf"
+
+    @property
+    def path(self) -> str:
+        """Get the relative path of the PDF file in the media directory."""
+        return f"chronos/plans/{self.filename}"
+
+    @property
+    def local_path(self) -> str:
+        """Get the full path under which the PDF file can accessed on the local system."""
+        return default_storage.path(self.path)
+
+    def save(self, *args, **kwargs):
+        super().save(*args, **kwargs)
+
+        if self.current_file:
+            if default_storage.exists(self.path):
+                default_storage.delete(self.path)
+            default_storage.save(self.path, self.current_file.file)
+
+    class Meta:
+        verbose_name = _("Automatic plan")
+        verbose_name_plural = _("Automatic plans")
+        constraints = [models.UniqueConstraint(fields=["site_id", "slug"], name="site_slug")]
+
+
 class ChronosGlobalPermissions(GlobalPermissionModel):
     class Meta:
         managed = False

aleksis/apps/chronos/util/change_tracker.py

+from django.contrib.contenttypes.models import ContentType
+from django.dispatch import Signal, receiver
+
+from celery import shared_task
+from reversion.models import Revision
+
+
+def _get_substitution_models():
+    from aleksis.apps.chronos.models import (
+        Event,
+        ExtraLesson,
+        LessonSubstitution,
+        SupervisionSubstitution,
+    )
+
+    return [LessonSubstitution, Event, ExtraLesson, SupervisionSubstitution]
+
+
+chronos_revision_created = Signal()
+substitutions_changed = Signal()
+
+
+@shared_task
+def handle_new_revision(revision_pk: int):
+    """Handle a new revision in background using Celery."""
+    revision = Revision.objects.get(pk=revision_pk)
+    if revision.version_set.filter(content_type__app_label="chronos").exists():
+        chronos_revision_created.send(sender=revision)
+
+
+@receiver(chronos_revision_created)
+def handle_substitution_change(sender: Revision, **kwargs):
+    """Handle the change of a substitution-like object."""
+    # Filter versions by substitution-like models
+    content_types = ContentType.objects.get_for_models(*_get_substitution_models()).values()
+    versions = sender.version_set.filter(content_type__in=content_types)
+    if versions:
+        substitutions_changed.send(sender=sender, versions=versions)