Repository URL to install this package:
|
Version:
0.1.4 ▾
|
import logging
import sys
from datetime import datetime, timedelta
from enum import Enum
from ics import Event
from ics.alarm import DisplayAlarm
import arrow
from django.contrib.postgres.fields import ArrayField, JSONField
from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models
from django.db.models import Q
from django.db.models.signals import (m2m_changed, post_delete, post_save,
pre_save)
from django.dispatch import receiver
from django_fsm import FSMField, transition
from django_fsm_log.decorators import fsm_log_by, fsm_log_description
from dynamic_preferences.registries import global_preferences_registry
from slugify import slugify
class DisplayAlarm2(DisplayAlarm):
def __hash__(self):
return hash(repr(self))
class ActivityType(models.Model):
title = models.CharField(max_length=255)
class Meta:
verbose_name = 'Activity Type'
verbose_name_plural = 'Activity Types'
def __str__(self):
return self.title
class ActivityStatus(Enum):
UNPLANNED = 'UNPLANNED'
PLANNED = 'PLANNED'
FINISHED = 'FINISHED'
REVIEWED = 'REVIEWED'
CANCELLED = 'CANCELLED'
status_choices = (
(ActivityStatus.UNPLANNED.value, 'Unplanned'),
(ActivityStatus.PLANNED.value, 'Planned'),
(ActivityStatus.FINISHED.value, 'Finished'),
(ActivityStatus.REVIEWED.value, 'Reviewed'),
(ActivityStatus.CANCELLED.value, 'Cancelled'),
)
class Activity(models.Model):
class Meta:
verbose_name = 'Activity'
verbose_name_plural = 'Activities'
notification_email_template = 'crm/notification_emails/activity.html'
status = FSMField(default='unplanned')
@fsm_log_by
@fsm_log_description
@transition(field=status, source='unplanned', target='planned', permission=lambda instance, user: user.has_perm('crm.change_activity'))
def plan(self, by=None, description=None):
pass
def can_plan(self):
errors = dict()
if not self.start:
errors['start'] = ['With this status, this has to be provided.']
if not self.end:
errors['end'] = ['With this status, this has to be provided.']
if self.start and self.end and self.start > self.end:
errors['non_field_errors'] = ['The End date has to be after the Start date']
return errors
@fsm_log_by
@fsm_log_description
@transition(field=status, source=['unplanned', 'planned'], target='finished', permission=lambda instance, user: user.has_perm('crm.change_activity'))
def finish(self, by=None, description=None):
pass
def can_finish(self):
return self.can_plan()
@fsm_log_by
@fsm_log_description
@transition(field=status, source=['unplanned', 'planned', 'finished'], target='reviewed', permission=lambda instance, user: user.has_perm('crm.change_activity'))
def review(self, by=None, description=None):
pass
def can_review(self):
errors = self.can_finish()
if not self.result or self.result == '' or self.result == '<p></p>':
errors['result'] = ['When reviewing an activity, a result has to be provided!']
return errors
@fsm_log_by
@fsm_log_description
@transition(field=status, source=['unplanned', 'planned', 'finished'], target='cancelled', permission=lambda instance, user: user.has_perm('crm.change_activity'))
def cancel(self, by=None, description=None):
pass
title = models.CharField(max_length=255)
description = models.TextField(blank=True, null=True)
result = models.TextField(blank=True, null=True)
type = models.ForeignKey('crm.ActivityType', related_name='activities', null=True, blank=True, on_delete=models.SET_NULL)
start = models.DateTimeField(blank=True, null=True)
end = models.DateTimeField(blank=True, null=True)
due = models.DateTimeField(blank=True, null=True)
reminder = models.DateTimeField(blank=True, null=True)
duration = models.IntegerField(blank=True, null=True) # In seconds
all_day = models.BooleanField(default=False)
# Inbox
unplanned_year = models.IntegerField(validators=[MinValueValidator(1000), MaxValueValidator(9999)], null=True, blank=True)
unplanned_quarter = models.IntegerField(validators=[MinValueValidator(1), MaxValueValidator(4)], null=True, blank=True)
unplanned_month = models.IntegerField(validators=[MinValueValidator(1), MaxValueValidator(12)], null=True, blank=True)
unplanned_week = models.IntegerField(validators=[MinValueValidator(1), MaxValueValidator(52)], null=True, blank=True)
unplanned_day = models.IntegerField(validators=[MinValueValidator(1), MaxValueValidator(366)], null=True, blank=True)
unplanned_week_day = models.IntegerField(validators=[MinValueValidator(1), MaxValueValidator(7)], null=True, blank=True)
location = models.CharField(max_length=255, null=True, blank=True)
location_longitude = models.DecimalField(max_digits=9, decimal_places=6, null=True, blank=True)
location_latitude = models.DecimalField(max_digits=9, decimal_places=6, null=True, blank=True)
created = models.DateTimeField(auto_now_add=True)
creator = models.ForeignKey('crm.Person', related_name='activities_owned', null=True, blank=True, on_delete=models.SET_NULL)
edited = models.DateTimeField(auto_now=True)
assigned_to = models.ForeignKey('crm.Person', related_name='activities', null=True, blank=True, on_delete=models.SET_NULL)
participants = models.ManyToManyField('crm.Entry', related_name='participates', blank=True)
preceded_by = models.ForeignKey('self', related_name='followed_by', blank=True, null=True, on_delete=models.SET_NULL)
reoccurance_rule = models.ForeignKey('crm.ActivityReoccuranceRule', null=True, blank=True, related_name='activities', on_delete=models.SET_NULL)
def __str__(self):
return '%s' % (self.title,)
def to_ics(self):
"""
Convert activity model to ICS event
"""
e = Event()
if self.start < self.end:
e.name = self.title
e.begin = self.start
e.end = self.end
e.created = arrow.get(self.created)
e.uid = f'workbench-{self.id}'
if self.location:
e.location = self.location
if self.description:
e.description = self.description
if self.reminder:
a = DisplayAlarm2(trigger=self.reminder)
e.alarms = set([a])
return e
else:
return None
@receiver(post_save, sender=Activity)
def post_save_activity(sender, instance, created, **kwargs):
if instance.status in ['finished', 'reviewed', 'cancelled'] and instance.reoccurance_rule:
instance.reoccurance_rule.create_activities_from_date(instance.start)
class ActivityReoccuranceRule(models.Model):
DAILY = 'DAILY'
WEEK_DAILY = 'WEEK_DAILY'
WEEKLY = 'WEEKLY'
MONTHLY = 'MONTHLY'
BIMONTHLY = 'BIMONTHLY'
QUARTERLY = 'QUARTERLY'
YEARLY = 'YEARLY'
def get_offset(self):
offset_dict = {
self.DAILY: timedelta(days=1),
self.WEEK_DAILY: timedelta(days=1),
self.WEEKLY: timedelta(days=7),
self.MONTHLY: timedelta(days=30),
self.BIMONTHLY: timedelta(days=60),
self.QUARTERLY: timedelta(days=90),
self.YEARLY: timedelta(days=365),
}
return offset_dict[self.reoccurance]
reoccurance_choices = (
(DAILY, 'Daily'),
(WEEK_DAILY, 'Week Daily'),
(WEEKLY, 'Weekly'),
(MONTHLY, 'Monthly'),
(BIMONTHLY, 'Bi Monthly'),
(QUARTERLY, 'Quarterly'),
(YEARLY, 'Yearly'),
)
activity_title = models.CharField(max_length=255)
activity_description = models.TextField(blank=True, null=True)
activity_assigned_to = models.ForeignKey('crm.Person', related_name='activity_reoccurance_rules', null=True, blank=True, on_delete=models.SET_NULL)
activity_participants = models.ManyToManyField('crm.Entry', related_name='activity_reoccurance_rules_participates', blank=True)
activity_type = models.ForeignKey('crm.ActivityType', related_name='activity_reoccurance_rules', null=True, blank=True, on_delete=models.SET_NULL)
activity_start = models.TimeField()
activity_end = models.TimeField()
activity_reminder = models.TimeField(null=True, blank=True)
creator = models.ForeignKey('crm.Person', null=True, blank=True, on_delete=models.SET_NULL)
start = models.DateField()
end = models.DateField(null=True, blank=True)
reoccurance = models.CharField(max_length=10, choices=reoccurance_choices, default=WEEKLY)
class Meta:
verbose_name = 'Activity Reoccurance Rule'
verbose_name_plural = 'Activity Reoccurance Rules'
def __str__(self):
return f'{self.activity_start} - {self.activity_end} {self.reoccurance}'
def create_activities_from_date(self, date):
NUMBER_OF_NEXT_ACTIVITES = 10
existing_activities = self.activities.filter(start__gte=date)
activities_to_create = NUMBER_OF_NEXT_ACTIVITES - existing_activities.count()
for i in range(existing_activities.count(), NUMBER_OF_NEXT_ACTIVITES, 1):
dt = self.get_offset() * i
new_date = (date + dt)
if type(new_date) is datetime:
new_date = new_date.date()
if self.end and new_date > self.end:
return
activity = Activity.objects.create(
creator=self.creator,
status='planned',
reoccurance_rule=self,
start=datetime.combine((date + dt), self.activity_start),
end=datetime.combine((date + dt), self.activity_end),
reminder=datetime.combine((date + dt), self.activity_reminder) if self.activity_reminder else None,
title=self.activity_title,
description=self.activity_description,
type=self.activity_type if self.activity_type else None,
assigned_to=self.activity_assigned_to if self.activity_assigned_to else None,
)
if self.activity_participants.all().exists():
activity.participants.set(self.activity_participants.all())
@receiver(post_save, sender=ActivityReoccuranceRule)
def post_save_activity_reoccurance_rule(sender, instance, created, **kwargs):
if not instance.activities.all().exists():
instance.create_activities_from_date(instance.start)
else:
for activity in instance.activities.filter(status='planned'):
activity.start = datetime.combine(activity.start.date(), instance.activity_start)
activity.end = datetime.combine(activity.end.date(), instance.activity_end)
if instance.activity_reminder:
activity.reminder = datetime.combine(activity.reminder.date() if activity.reminder else instance.start, instance.activity_reminder)
else:
activity.reminder = None
if instance.activity_type:
activity.type = instance.activity_type
if instance.activity_participants.all().exists():
activity.participants.clear()
activity.participants.set(instance.activity_participants.all())
if instance.activity_assigned_to:
activity.assigned_to = instance.activity_assigned_to
activity.save()
if instance.end:
instance.activities.filter(end__date__gt=instance.end).delete()