Source code for shuup.tasks.models

# -*- coding: utf-8 -*-
# This file is part of Shuup.
#
# Copyright (c) 2012-2021, Shuup Commerce Inc. All rights reserved.
#
# This source code is licensed under the OSL-3.0 license found in the
# LICENSE file in the root directory of this source tree.
from __future__ import unicode_literals

from django.db import models
from django.db.models import Q
from django.utils.encoding import python_2_unicode_compatible
from django.utils.safestring import mark_safe
from django.utils.timezone import now
from django.utils.translation import ugettext_lazy as _
from enumfields import Enum, EnumIntegerField
from parler.models import TranslatableModel, TranslatedFields

from shuup.core.fields import InternalIdentifierField
from shuup.utils.analog import LogEntryKind, define_log_model
from shuup.utils.django_compat import force_text


[docs]class TaskStatus(Enum):
    NEW = 1
    IN_PROGRESS = 2
    COMPLETED = 3
    DELETED = 4

    class Labels:
        NEW = _("New")
        IN_PROGRESS = _("In progress")
        COMPLETED = _("Completed")
        DELETED = _("Deleted")


[docs]class TaskCommentVisibility(Enum):
    PUBLIC = 1
    STAFF_ONLY = 2
    ADMINS_ONLY = 3

    class Labels:
        PUBLIC = _("Public")
        STAFF_ONLY = _("Staff Only")
        ADMINS_ONLY = _("Admins Only")


@python_2_unicode_compatible
[docs]class TaskType(TranslatableModel):
    identifier = InternalIdentifierField(unique=False, blank=True, null=True, editable=True)
    shop = models.ForeignKey(
        on_delete=models.CASCADE, to="shuup.Shop", verbose_name=_("shop"), related_name="task_types"
    )
    translations = TranslatedFields(name=models.TextField(verbose_name=_("name")))

    class Meta:
        unique_together = ("shop", "identifier")
        verbose_name = _("task type")
        verbose_name_plural = _("task types")

    def __str__(self):
        return self.name


[docs]class TaskQuerySet(models.QuerySet):
[docs]    def completed(self):
        return self.filter(status=TaskStatus.COMPLETED)

[docs]    def in_progress(self):
        return self.filter(status=TaskStatus.IN_PROGRESS)

[docs]    def is_new(self):
        return self.filter(tatus=TaskStatus.NEW)

[docs]    def for_shop(self, shop):
        return self.filter(shop=shop)

[docs]    def delete(self):
        self.update(status=TaskStatus.DELETED)

[docs]    def assigned_to(self, contact):
        return self.filter(assigned_to=contact)


@python_2_unicode_compatible
[docs]class Task(models.Model):
    shop = models.ForeignKey(on_delete=models.CASCADE, to="shuup.Shop", verbose_name=_("shop"), related_name="tasks")
    name = models.CharField(verbose_name=_("name"), max_length=60)
    type = models.ForeignKey(on_delete=models.CASCADE, to=TaskType, verbose_name=_("task type"), related_name="tasks")
    status = EnumIntegerField(TaskStatus, default=TaskStatus.NEW, verbose_name=_("status"))
    priority = models.PositiveIntegerField(default=0, verbose_name=_("priority"), db_index=True)
    creator = models.ForeignKey(
        on_delete=models.CASCADE,
        to="shuup.Contact",
        blank=True,
        null=True,
        related_name="creted_tasks",
        verbose_name=_("creator"),
    )
    assigned_to = models.ForeignKey(
        on_delete=models.CASCADE,
        to="shuup.Contact",
        blank=True,
        null=True,
        related_name="assigned_tasks",
        verbose_name=_("assigned to"),
    )
    completed_by = models.ForeignKey(
        on_delete=models.CASCADE,
        to="shuup.Contact",
        blank=True,
        null=True,
        related_name="completed_tasks",
        verbose_name=_("completed by"),
    )
    completed_on = models.DateTimeField(verbose_name=_("completed on"), null=True, blank=True)
    created_on = models.DateTimeField(auto_now_add=True, editable=False, db_index=True, verbose_name=_("created on"))
    modified_on = models.DateTimeField(auto_now=True, editable=False, db_index=True, verbose_name=_("modified on"))

    objects = TaskQuerySet.as_manager()

    def __str__(self):
        return self.name

[docs]    def assign(self, user):
        self.assigned_to = user
        self.status = TaskStatus.IN_PROGRESS
        self.save()

[docs]    def delete(self):
        self.status = TaskStatus.DELETED
        self.save(update_fields=["status"])
        self.add_log_entry("Success! Deleted (soft).", kind=LogEntryKind.DELETION)

[docs]    def comment(self, contact, comment, visibility=TaskCommentVisibility.PUBLIC):
        comment = TaskComment(task=self, author=contact, body=comment, visibility=visibility)
        comment.full_clean()
        comment.save()
        return comment

[docs]    def set_in_progress(self):
        self.status = TaskStatus.IN_PROGRESS
        self.add_log_entry("Info! In progress.", kind=LogEntryKind.EDIT)
        self.save()

[docs]    def set_completed(self, contact):
        self.completed_by = contact
        self.completed_on = now()
        self.status = TaskStatus.COMPLETED
        self.add_log_entry("Success! Completed.", kind=LogEntryKind.EDIT)
        self.save()

[docs]    def get_completion_time(self):
        if self.completed_on:
            return self.completed_on - self.created_on


[docs]class TaskCommentQuerySet(models.QuerySet):
[docs]    def for_contact(self, contact):
        visibility_filters = Q(visibility=TaskCommentVisibility.PUBLIC)

        if hasattr(contact, "user") and contact.user:
            # see everything
            if contact.user.is_superuser:
                visibility_filters = Q()

            elif contact.user.is_staff:
                visibility_filters |= Q(
                    visibility=TaskCommentVisibility.STAFF_ONLY, task__shop__staff_members=contact.user
                )

        return self.filter(visibility_filters).distinct()


[docs]class TaskComment(models.Model):
    task = models.ForeignKey(on_delete=models.CASCADE, to=Task, verbose_name=_("task"), related_name="comments")
    author = models.ForeignKey(
        on_delete=models.CASCADE,
        to="shuup.Contact",
        blank=True,
        null=True,
        related_name="task_comments",
        verbose_name=_("author"),
    )
    visibility = EnumIntegerField(
        TaskCommentVisibility, default=TaskCommentVisibility.PUBLIC, db_index=True, verbose_name=_("visibility")
    )
    body = models.TextField(verbose_name=_("body"))
    created_on = models.DateTimeField(auto_now_add=True, editable=False, db_index=True, verbose_name=_("created on"))
    modified_on = models.DateTimeField(auto_now=True, editable=False, db_index=True, verbose_name=_("modified on"))

    objects = TaskCommentQuerySet.as_manager()

[docs]    def reply(self, contact, body):
        comment = TaskComment(task=self.task, comment_author=contact, body=body)
        comment.full_clean()
        comment.save()
        return comment

[docs]    def as_html(self):
        return mark_safe(force_text(self.body))

[docs]    def can_see(self, user):
        is_admin = user.is_superuser
        is_staff = bool(user.is_staff and user in self.task.shop.staff_members.all())

        if not (is_admin or is_staff):
            return self.visibility == TaskCommentVisibility.PUBLIC
        elif not is_admin:
            return (
                self.visibility == TaskCommentVisibility.PUBLIC or self.visibility == TaskCommentVisibility.STAFF_ONLY
            )

        return True


TaskLogEntry = define_log_model(Task)