apps.py

from typing import Any, Optional

import django.apps
from django.db import transaction

from reversion.signals import post_revision_commit

from aleksis.core.util.apps import AppConfig


class ChronosConfig(AppConfig):
    name = "aleksis.apps.chronos"
    verbose_name = "AlekSIS — Chronos (Timetables)"

    urls = {
        "Repository": "https://edugit.org/AlekSIS/official/AlekSIS-App-Chronos/",
    }
    licence = "EUPL-1.2+"
    copyright_info = (
        ([2018, 2019, 2020, 2021, 2022], "Jonathan Weth", "wethjo@katharineum.de"),
        ([2018, 2019], "Frank Poetzsch-Heffter", "p-h@katharineum.de"),
        ([2019, 2020, 2022], "Dominik George", "dominik.george@teckids.org"),
        ([2019, 2021], "Hangzhi Yu", "yuha@katharineum.de"),
        ([2019], "Julian Leucker", "leuckeju@katharineum.de"),
        ([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):
            """Handle a new post revision commit signal in background."""
            transaction.on_commit(lambda: handle_new_revision.delay(revision.pk))

        post_revision_commit.connect(_handle_post_revision_commit, weak=False)

    def _ensure_notification_task(self):
        """Update or create the task for sending notifications."""
        from django.conf import settings  # noqa

        from celery import schedules
        from django_celery_beat.models import CrontabSchedule, PeriodicTask

        from aleksis.core.util.core_helpers import get_site_preferences

        time_for_sending = get_site_preferences()["chronos__time_for_sending_notifications"]
        active = get_site_preferences()["chronos__send_notifications_site"]

        if active:
            schedule = schedules.crontab(
                minute=str(time_for_sending.minute), hour=str(time_for_sending.hour)
            )
            schedule = CrontabSchedule.from_schedule(schedule)
            schedule.timezone = settings.TIME_ZONE
            schedule.save()

        possible_periodic_tasks = PeriodicTask.objects.filter(
            task="chronos_send_notifications_for_next_day"
        )

        if not active:
            possible_periodic_tasks.delete()

        elif possible_periodic_tasks.exists():
            task = possible_periodic_tasks[0]
            for d_task in possible_periodic_tasks:
                if d_task != task:
                    d_task.delete()

            if task.crontab != schedule:
                task.interval, task.solar, task.clocked = None, None, None
                task.crontab = schedule
                task.save()

        elif active:
            PeriodicTask.objects.get_or_create(
                task="chronos_send_notifications_for_next_day",
                crontab=schedule,
                defaults=dict(name="Send notifications for next day (automatic schedule)"),
            )

    def preference_updated(
        self,
        sender: Any,
        section: Optional[str] = None,
        name: Optional[str] = None,
        old_value: Optional[Any] = None,
        new_value: Optional[Any] = None,
        **kwargs,
    ) -> None:
        if section == "chronos" and name in (
            "send_notifications_site",
            "time_for_sending_notifications",
        ):
            self._ensure_notification_task()

    def post_migrate(
        self,
        app_config: django.apps.AppConfig,
        verbosity: int,
        interactive: bool,
        using: str,
        **kwargs,
    ) -> None:
        super().post_migrate(app_config, verbosity, interactive, using, **kwargs)
        # Ensure that the notification task is created after setting up AlekSIS
        self._ensure_notification_task()