authentic2-cut/src/authentic2_cut/models.py

import os.path
from datetime import timedelta

from django.conf import settings
from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.core.files.storage import default_storage
from django.db import models
from django.db.models.query import Q
from django.db.models.signals import post_delete
from django.dispatch import receiver
from django.urls import reverse
from django.utils.timezone import now
from sorl.thumbnail import delete, get_thumbnail


class ValidationRequestManager(models.Manager):
    def for_origin(self, origin):
        ct = ContentType.objects.get_for_model(origin)
        pk = origin.pk
        return self.filter(origin_ct=ct, origin_id=pk)

    def next_request(self, user, after=None):
        qs = self.select_for_update()
        qs = qs.order_by('pk')
        # la demande est nouvelle
        # la demande est en cours de traitement mais
        # aucune date de prise en main n'est posée
        # la demande est prise depuis plus d'une heure
        # la demande est prise par <user>
        qs = qs.filter(
            Q(status=ValidationRequest.STATUS_RECEIVED)
            & (
                Q(taken__isnull=True)
                | Q(taken__lt=now() - timedelta(seconds=self.model.TAKEN_DELAY))
                | Q(taken_by=user)
            )
        )
        # on veut la prochaine demande à traiter
        if after:
            qs = qs.filter(pk__gt=after.pk)
        validation_request = qs[:1].first()
        if validation_request:
            # la demande est associée à l'utilisateur en cours
            validation_request.taken_by = user
            validation_request.taken = now()
            validation_request.save()
        return validation_request

    def received(self):
        return self.filter(status=ValidationRequest.STATUS_RECEIVED)

    def cleanup(self):
        self.filter(validated__lt=now() - timedelta(days=92)).delete()


class ValidationRequest(models.Model):
    TAKEN_DELAY = 3600

    STATUS_RECEIVED = 'received'
    STATUS_ACCEPTED = 'accepted'
    STATUS_REFUSED = 'refused'

    STATUS_CHOICES = (
        (STATUS_RECEIVED, 'reçue'),
        (STATUS_ACCEPTED, 'acceptée'),
        (STATUS_REFUSED, 'refusée'),
    )

    REASON_UNREADABLE = 'unreadable'
    REASON_INVALID = 'invalid'
    REASON_UNDERAGED = 'underaged'
    REASON_CHOICES = (
        (REASON_UNREADABLE, 'Pièce(s) illisible(s)'),
        (REASON_INVALID, 'Pièce(s) invalides(s)'),
        (REASON_UNDERAGED, 'Invividu mineur'),
    )

    created = models.DateTimeField(verbose_name='Date de création', db_index=True, auto_now_add=True)
    origin_ct = models.ForeignKey(
        'contenttypes.ContentType', verbose_name='origin ct', on_delete=models.CASCADE
    )
    origin_id = models.PositiveIntegerField(verbose_name='origin id')
    origin = GenericForeignKey('origin_ct', 'origin_id')
    user = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        verbose_name='Utilisateur',
        related_name='validation_requests',
        on_delete=models.CASCADE,
    )
    status = models.CharField(
        max_length=16, choices=STATUS_CHOICES, verbose_name='Statut', default=STATUS_RECEIVED
    )
    reason = models.TextField(choices=REASON_CHOICES, blank=True, verbose_name='Raison du refus')
    validated = models.DateTimeField(null=True, blank=True, verbose_name='Date de validation')
    validated_by = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        verbose_name='Validé/refusé par',
        null=True,
        blank=True,
        related_name='validation_requests_validated',
        on_delete=models.SET_NULL,
    )
    external_id = models.TextField(blank=True, null=True, verbose_name='Identifiant externe')
    taken = models.DateTimeField(verbose_name='En cours', null=True)
    taken_by = models.ForeignKey(
        settings.AUTH_USER_MODEL,
        verbose_name='Traité par',
        null=True,
        related_name='validation_requests_taken',
        on_delete=models.SET_NULL,
    )

    objects = ValidationRequestManager()

    @property
    def is_taken(self):
        return self.taken and now() - self.taken < timedelta(seconds=self.TAKEN_DELAY)

    def get_taken(self):
        return self.taken if self.is_taken else None

    def get_taken_by(self):
        return self.taken_by if self.is_taken else None

    @property
    def human_status(self):
        if self.status == self.STATUS_REFUSED:
            return 'refusé'
        elif self.status == self.STATUS_ACCEPTED:
            return 'accepté'
        elif self.status == self.STATUS_RECEIVED:
            if self.is_taken:
                return 'en traitement'
            else:
                return 'reçu'
        else:
            raise NotImplementedError

    @property
    def human_status_who(self):
        if self.status == self.STATUS_REFUSED:
            return self.validated_by
        elif self.status == self.STATUS_ACCEPTED:
            return self.validated_by
        elif self.status == self.STATUS_RECEIVED:
            if self.is_taken:
                return self.taken_by

    def accept(self, user):
        self.status = self.STATUS_ACCEPTED
        self.validated = now()
        self.validated_by = user
        self.save()

    class Meta:
        verbose_name = 'Requête de validation'
        verbose_name_plural = 'Requêtes de validation'
        ordering = ('-created',)
        unique_together = (('origin_ct', 'origin_id', 'user', 'external_id'),)


class ValidationRequestAttachment(models.Model):
    validation_request = models.ForeignKey(
        ValidationRequest,
        verbose_name='requête de validation',
        related_name='attachments',
        on_delete=models.CASCADE,
    )

    image = models.ImageField(verbose_name='contenu')

    @property
    def extension(self):
        return self.image.name.rsplit('.', 1)[-1]

    @property
    def url(self):
        return reverse(
            'cut-manager-user-validation-attachment',
            kwargs={'pk': self.pk, 'filename': self.image.name.rsplit('/', 1)[-1]},
        )

    @property
    def thumbnail(self):
        try:
            thumbnail = get_thumbnail(self.image, '200x200')
        except Exception:
            raise
            return None
        try:
            # check file exists and is readable
            with default_storage.open(thumbnail.name):
                pass
            return thumbnail
        except OSError:
            pass
        return None

    @property
    def thumbnail_image(self):
        thumbnail = self.thumbnail
        if thumbnail:
            return {
                'src': reverse(
                    'cut-manager-user-validation-attachment-thumbnail',
                    kwargs={'pk': self.pk, 'filename': self.image.name.rsplit('/', 1)[-1]},
                ),
                'height': thumbnail.height,
                'width': thumbnail.width,
            }
        return {}

    class Meta:
        ordering = ('pk',)
        verbose_name = 'Pièce jointe'
        verbose_name_plural = 'Pièces jointes'


@receiver(post_delete, sender=ValidationRequestAttachment)
def auto_delete_image_files(sender, instance, **kwargs):
    if instance.image:
        try:
            delete(instance.image)
        except OSError:
            pass
        if os.path.isfile(instance.image.path):
            instance.image.delete(save=False)