aircox-radiocampus/aircox/models/schedule.py

import calendar
import zoneinfo

from django.db import models
from django.utils import timezone as tz
from django.utils.functional import cached_property
from django.utils.translation import gettext_lazy as _

from aircox import utils

from .rerun import Rerun


__all__ = ("Schedule",)


def current_timezone_key():
    return tz.get_current_timezone().key


# ? BIG FIXME: self.date is still used as datetime
class Schedule(Rerun):
    """A Schedule defines time slots of programs' diffusions.

    It can be an initial run or a rerun (in such case it is linked to
    the related schedule).
    """

    # Frequency for schedules. Basically, it is a mask of bits where each bit
    # is a week. Bits > rank 5 are used for special schedules.
    # Important: the first week is always the first week where the weekday of
    # the schedule is present.
    # For ponctual programs, there is no need for a schedule, only a diffusion
    class Frequency(models.IntegerChoices):
        ponctual = 0b000000, _("ponctual")
        first = 0b000001, _("1st {day} of the month")
        second = 0b000010, _("2nd {day} of the month")
        third = 0b000100, _("3rd {day} of the month")
        fourth = 0b001000, _("4th {day} of the month")
        last = 0b010000, _("last {day} of the month")
        first_and_third = 0b000101, _("1st and 3rd {day} of the month")
        second_and_fourth = 0b001010, _("2nd and 4th {day} of the month")
        every = 0b011111, _("{day}")
        one_on_two = 0b100000, _("one {day} on two")
        # every_weekday = 0b10000000 _("from Monday to Friday")

    date = models.DateField(
        _("date"),
        help_text=_("date of the first diffusion"),
    )
    time = models.TimeField(
        _("time"),
        help_text=_("start time"),
    )
    timezone = models.CharField(
        _("timezone"),
        default=current_timezone_key,
        max_length=100,
        choices=sorted([(x, x) for x in zoneinfo.available_timezones()]),
        help_text=_("timezone used for the date"),
    )
    duration = models.TimeField(
        _("duration"),
        help_text=_("regular duration"),
    )
    frequency = models.SmallIntegerField(
        _("frequency"),
        choices=Frequency.choices,
    )

    class Meta:
        verbose_name = _("Schedule")
        verbose_name_plural = _("Schedules")

    def __init__(self, *args, **kwargs):
        self._initial = kwargs
        super().__init__(*args, **kwargs)

    def __str__(self):
        return "{} - {}, {}".format(
            self.program.title,
            self.get_frequency_display(),
            self.time.strftime("%H:%M"),
        )

    def save_rerun(self):
        super().save_rerun()
        self.duration = self.initial.duration
        self.frequency = self.initial.frequency

    @cached_property
    def tz(self):
        """Pytz timezone of the schedule."""
        return zoneinfo.ZoneInfo(self.timezone)

    @cached_property
    def start(self):
        """Datetime of the start (timezone unaware)"""
        return tz.datetime.combine(self.date, self.time)

    @cached_property
    def end(self):
        """Datetime of the end."""
        return self.start + utils.to_timedelta(self.duration)

    def get_frequency_display(self):
        """Return frequency formated for display."""
        from django.template.defaultfilters import date

        return self._get_FIELD_display(self._meta.get_field("frequency")).format(day=date(self.date, "l")).capitalize()

    def normalize(self, date):
        """Return a datetime set to schedule's time for the provided date,
        handling timezone (based on schedule's timezone)."""
        date = tz.datetime.combine(date, self.time)
        return date.replace(tzinfo=self.tz)

    def dates_of_month(self, date, frequency=None, sched_date=None):
        """Return normalized diffusion dates of provided date's month.

        :param Date date: date of the month to get dates from;
        :param Schedule.Frequency frequency: frequency (defaults to ``self.frequency``)
        :param Date sched_date: schedule start date (defaults to ``self.date``)
        :return list of diffusion dates
        """
        if frequency is None:
            frequency = self.frequency

        if sched_date is None:
            sched_date = self.date

        if frequency == Schedule.Frequency.ponctual:
            return []

        sched_wday = sched_date.weekday()
        date = date.replace(day=1)

        # last of the month
        if frequency == Schedule.Frequency.last:
            date = date.replace(day=calendar.monthrange(date.year, date.month)[1])
            date_wday = date.weekday()

            # end of month before the wanted weekday: move one week back
            if date_wday < sched_wday:
                date -= tz.timedelta(days=7)
            date += tz.timedelta(days=sched_wday - date_wday)
            return [self.normalize(date)]

        # move to the first day of the month that matches the schedule's
        # weekday. Check on SO#3284452 for the formula
        date_wday, month = date.weekday(), date.month
        date += tz.timedelta(days=(7 if date_wday > sched_wday else 0) - date_wday + sched_wday)

        if frequency == Schedule.Frequency.one_on_two:
            # - adjust date with modulo 14 (= 2 weeks in days)
            # - there are max 3 "weeks on two" per month
            if (date - sched_date).days % 14:
                date += tz.timedelta(days=7)
            dates = (date + tz.timedelta(days=14 * i) for i in range(0, 3))
        else:
            dates = (date + tz.timedelta(days=7 * week) for week in range(0, 5) if frequency & (0b1 << week))

        return [self.normalize(date) for date in dates if date.month == month]

    def diffusions_of_month(self, date, frequency=None, sched_date=None):
        """Get episodes and diffusions for month of provided date, including
        reruns.

        :param Date date: date of the month to get diffusions from;
        :param Schedule.Frequency frequency: frequency (defaults to ``self.frequency``)
        :param Date sched_date: schedule start date (defaults to ``self.date``)
        :returns: tuple([Episode], [Diffusion])
        """
        from .diffusion import Diffusion
        from .episode import Episode

        if frequency is None:
            frequency = self.frequency

        if sched_date is None:
            sched_date = self.date

        if self.initial is not None or frequency == Schedule.Frequency.ponctual:
            return [], []

        # dates for self and reruns as (date, initial)
        reruns = [(rerun, rerun.date - sched_date) for rerun in self.rerun_set.all()]

        dates = {date: None for date in self.dates_of_month(date, frequency, sched_date)}
        dates.update(
            (rerun.normalize(date.date() + delta), date) for date in list(dates.keys()) for rerun, delta in reruns
        )

        # remove dates corresponding to existing diffusions
        saved = set(
            Diffusion.objects.filter(start__in=dates.keys(), program=self.program, schedule=self).values_list(
                "start", flat=True
            )
        )

        # make diffs
        duration = utils.to_timedelta(self.duration)
        diffusions = {}
        episodes = {}

        for date, initial in dates.items():
            if date in saved:
                continue

            if initial is None:
                episode = Episode.from_page(self.program, date=date)
                episode.date = date
                episodes[date] = episode
            else:
                episode = episodes[initial]
                initial = diffusions[initial]

            diffusions[date] = Diffusion(
                episode=episode,
                schedule=self,
                type=Diffusion.TYPE_ON_AIR,
                initial=initial,
                start=date,
                end=date + duration,
            )
        return episodes.values(), diffusions.values()