extensions-website/extensions/models.py

from typing import List
from statistics import median
import datetime
import logging

from django.contrib.auth import get_user_model
from django.core.exceptions import ObjectDoesNotExist, BadRequest, ValidationError
from django.db import models, transaction
from django.db.models import F, Q, Count
from django.urls import reverse

from common.fields import FilterableManyToManyField
from common.model_mixins import CreatedModifiedMixin, TrackChangesMixin
from constants.base import (
    AUTHOR_ROLE_CHOICES,
    AUTHOR_ROLE_DEV,
    EXTENSION_STATUS_CHOICES,
    EXTENSION_TYPE_CHOICES,
    EXTENSION_TYPE_SLUGS,
    FILE_STATUS_CHOICES,
)
from constants.licenses import ALL_LICENSES
from constants.version_permissions import ALL_VERSION_PERMISSIONS
import common.help_texts
import extensions.fields

import utils

User = get_user_model()
log = logging.getLogger(__name__)


class RatingMixin:
    @property
    def text_ratings_count(self) -> int:
        return self.ratings.listed_texts.count()

    @property
    def total_ratings_count(self) -> int:
        return self.ratings.listed.count()

    @property
    def ratings_by_score(self):
        total_count = self.ratings.listed.count()
        counts_q = self.ratings.listed.values('score').annotate(count=Count('score'))
        counts = {
            **{score: 0 for score in self.ratings.model.SCORES.values},
            **{_['score']: _['count'] for _ in counts_q},
        }
        return sorted(
            [
                {
                    'score': score,
                    'count': count,
                    'percent': 100 / (total_count / count) if count else 0,
                }
                for score, count in counts.items()
            ],
            key=lambda _: _['score'],
            reverse=True,
        )

    def recalculate_average_score(self) -> int:
        old_average_score = self.average_score
        rating_values = self.ratings.listed.values_list('score', flat=True)
        if rating_values:
            self.average_score = median(rating_values)
        elif self.average_score:
            self.average_score = 0
        self.save(update_fields={'average_score'})
        log.info(
            'Average score of %s pk=%s changed from %s to %s',
            self.__class__.__name__,
            self.pk,
            old_average_score,
            self.average_score,
        )
        return self.average_score


class License(CreatedModifiedMixin, models.Model):
    name = models.CharField(max_length=128, null=False, blank=False, unique=True)
    slug = models.SlugField(
        blank=False,
        null=False,
        help_text='Should be taken from https://spdx.org/licenses/',
    )
    url = models.URLField(blank=False, null=False)

    def __str__(self) -> str:
        return f'{self.name}'

    @classmethod
    def generate(cls):
        """Generate License records from constants."""
        licenses = [cls(id=li.id, name=li.name, slug=li.slug, url=li.url) for li in ALL_LICENSES]
        cls.objects.bulk_create(licenses)

    @classmethod
    def get_by_name(cls, name: str):
        return cls.objects.filter(name__startswith=name).first()

    @classmethod
    def get_by_slug(cls, slug: str):
        return cls.objects.filter(slug__startswith=slug).first()


class ExtensionManager(models.Manager):
    @property
    def listed(self):
        return self.filter(
            status=self.model.STATUSES.APPROVED,
            is_listed=True,
        )

    @property
    def unlisted(self):
        return self.exclude(status=self.model.STATUSES.APPROVED)

    def authored_by(self, user_id: int):
        return self.filter(maintainer__user_id=user_id)

    def listed_or_authored_by(self, user_id: int):
        return self.filter(
            Q(status=self.model.STATUSES.APPROVED) | Q(maintainer__user_id=user_id)
        ).distinct()


class Extension(CreatedModifiedMixin, RatingMixin, TrackChangesMixin, models.Model):
    track_changes_to_fields = {
        'status',
        'name',
        'description',
        'support',
        'website',
    }
    TYPES = EXTENSION_TYPE_CHOICES
    STATUSES = EXTENSION_STATUS_CHOICES

    date_approved = models.DateTimeField(null=True, blank=True, editable=False)
    date_status_changed = models.DateTimeField(null=True, blank=True, editable=False)

    type = models.PositiveSmallIntegerField(choices=TYPES, default=TYPES.BPY, editable=False)
    slug = models.SlugField(unique=True, null=False, blank=False, editable=False)
    extension_id = models.CharField(max_length=255, unique=True, null=False, blank=False)
    name = models.CharField(max_length=255, unique=True, null=False, blank=False)
    description = models.TextField(help_text=common.help_texts.markdown)
    is_listed = models.BooleanField(
        help_text='Whether the extension should be listed. It is kept in sync via signals.',
        default=False,
    )
    previews = FilterableManyToManyField(
        'files.File',
        through='Preview',
        related_name='extensions',
        # TODO: filter only images and videos.
        # q_filter=Q(type=FILE_TYPE_CHOICES.IMAGE),
    )
    status = models.PositiveSmallIntegerField(choices=STATUSES, default=STATUSES.INCOMPLETE)
    support = models.URLField(
        help_text='URL for reporting issues or contact details for support.', null=True, blank=True
    )
    website = models.URLField(help_text='External URL for the extension.', null=True, blank=True)

    authors = FilterableManyToManyField(
        User,
        through='Maintainer',
        related_name='extensions',
    )
    team = models.ForeignKey('teams.Team', null=True, blank=True, on_delete=models.SET_NULL)

    average_score = models.FloatField(max_length=255, default=0, null=False)
    download_count = models.PositiveIntegerField(default=0)
    view_count = models.PositiveIntegerField(default=0)

    objects = ExtensionManager()

    class Meta:
        ordering = ['-average_score', '-date_created', 'name']

    def __str__(self):
        return f'{self.get_type_display()} "{self.name}"'

    @property
    def type_slug(self) -> str:
        return EXTENSION_TYPE_SLUGS[self.type]

    @property
    def status_slug(self) -> str:
        return utils.slugify(EXTENSION_STATUS_CHOICES[self.status - 1][1])

    def clean(self) -> None:
        if not self.slug:
            self.slug = utils.slugify(self.name)
        # Require at least one approved version with a file for approved extensions
        if self.status == self.STATUSES.APPROVED:
            if not self.latest_version:
                raise ValidationError(
                    {
                        'status': (
                            'Extension cannot have an approved status without an approved version'
                        )
                    }
                )
        super().clean()

    def save(self, *args, **kwargs):
        self.clean()
        return super().save(*args, **kwargs)

    @transaction.atomic
    def approve(self, reviewer=None):
        """TODO: Approve an extension which is currently in review."""
        # TODO: require notes and reviewer, do extra checks before cascading to version/file.

        # TODO: check that latest version is currently in review (whatever that means in data?)
        latest_version = self.latest_version
        file = latest_version.file
        file.status = FILE_STATUS_CHOICES.APPROVED
        file.save()

        # TODO: check that this extension is currently in review (whatever that means in data?)
        self.previews.update(status=FILE_STATUS_CHOICES.APPROVED)
        self.status = self.STATUSES.APPROVED
        self.save()

    @property
    def cannot_be_deleted_reasons(self) -> List[str]:
        """Return a list of reasons why this extension cannot be deleted."""
        reasons = []
        if self.is_listed:
            reasons.append('is_listed')
        if self.ratings.count() > 0:
            reasons.append('has_ratings')
        if self.abusereport_set.count() > 0:
            reasons.append('has_abuse_reports')
        for v in self.versions.all():
            reasons.extend(v.cannot_be_deleted_reasons)
        return reasons

    def get_absolute_url(self):
        return reverse('extensions:detail', args=[self.type_slug, self.slug])

    def get_draft_url(self):
        return reverse('extensions:draft', args=[self.type_slug, self.slug])

    def get_manage_url(self):
        return reverse('extensions:manage', args=[self.type_slug, self.slug])

    def get_manage_versions_url(self):
        return reverse('extensions:manage-versions', args=[self.type_slug, self.slug])

    def get_delete_url(self):
        return reverse('extensions:delete', args=[self.type_slug, self.slug])

    def get_new_version_url(self):
        return reverse('extensions:new-version', args=[self.type_slug, self.slug])

    def get_versions_url(self):
        return reverse('extensions:versions', args=[self.type_slug, self.slug])

    def get_ratings_url(self):
        return reverse('ratings:for-extension', args=[self.type_slug, self.slug])

    def get_rate_url(self):
        return reverse('ratings:new', args=[self.type_slug, self.slug])

    def get_report_url(self):
        return reverse('abuse:report-extension', args=[self.type_slug, self.slug])

    def get_review_url(self):
        return reverse('reviewers:approval-detail', args=[self.slug])

    def get_previews(self):
        """Get preview files, sorted by Preview.position.

        TODO: Might be better to query Previews directly instead of going
        for the reverse relationship.
        """
        return self.previews.listed.order_by('extension_preview__position')

    @property
    def valid_file_statuses(self) -> List[int]:
        if self.status == self.STATUSES.APPROVED:
            return [FILE_STATUS_CHOICES.APPROVED]
        return [FILE_STATUS_CHOICES.AWAITING_REVIEW, FILE_STATUS_CHOICES.APPROVED]

    @property
    def latest_version(self):
        """Retrieve the latest version."""
        return (
            self.versions.filter(
                file__status__in=self.valid_file_statuses,
                file__isnull=False,
            )
            .order_by('date_created')
            .last()
        )

    @property
    def current_version(self):
        """Return the latest public listed version of an extension.

        If the add-on is not public, it can return a listed version awaiting
        review (since non-public add-ons should not have public versions).

        If the add-on has not been created yet or is deleted, it returns None.
        """
        if not self.id:
            return None
        try:
            return self.version
        except ObjectDoesNotExist:
            pass
        return None

    def can_request_review(self):
        """Return whether an add-on can request a review or not."""
        if self.is_disabled or self.status in (
            self.STATUSES.APPROVED,
            self.STATUSES.AWAITING_REVIEW,
        ):
            return False

        latest_version = self.latest_version

        return latest_version is not None and not latest_version.file.reviewed

    @property
    def is_approved(self) -> bool:
        return self.status == self.STATUSES.APPROVED

    @property
    def is_disabled(self):
        """True if this Extension is disabled.

        It could be disabled by an admin or disabled by the developer
        """
        return self.status in (self.STATUSES.DISABLED_BY_AUTHOR, self.STATUSES.DISABLED)

    def should_redirect_to_submit_flow(self):
        return (
            self.status == self.STATUSES.INCOMPLETE
            and not self.has_complete_metadata()
            and self.latest_version is not None
        )

    def has_maintainer(self, user) -> bool:
        """Return True if given user is listed as a maintainer."""
        if user is None or user.is_anonymous:
            return False
        return self.authors.filter(maintainer__user_id=user.pk).exists()

    def can_rate(self, user) -> bool:
        """Return True if given user can rate this extension.

        Allow leaving a rating once and don't allow maintainers rate their own extensions.
        """
        return (
            not self.has_maintainer(user)
            and not self.ratings.filter(
                reply_to=None,
                user_id=user.pk,
            ).exists()
        )

    @classmethod
    def get_lookup_field(cls, identifier):
        lookup_field = 'pk'
        if identifier and not str(identifier).isdigit():
            lookup_field = 'slug'
        return lookup_field


class VersionPermission(CreatedModifiedMixin, models.Model):
    name = models.CharField(max_length=128, null=False, blank=False, unique=True)
    slug = models.SlugField(
        blank=False,
        null=False,
        help_text='Permissions add-ons are expected to need.',
    )
    help = models.CharField(max_length=128, null=False, blank=False, unique=True)

    def __str__(self) -> str:
        return f'{self.name}'

    @classmethod
    def generate(cls):
        """Generate Permission records from constants."""
        permissions = [
            cls(id=li.id, name=li.name, slug=li.slug, help=li.help)
            for li in ALL_VERSION_PERMISSIONS
        ]
        cls.objects.bulk_create(permissions)

    @classmethod
    def get_by_name(cls, name: str):
        return cls.objects.filter(name__startswith=name).first()

    @classmethod
    def get_by_slug(cls, slug: str):
        return cls.objects.filter(slug__startswith=slug).first()


class Tag(CreatedModifiedMixin, models.Model):
    TYPES = EXTENSION_TYPE_CHOICES

    name = models.CharField(max_length=128, null=False, blank=False)
    slug = models.SlugField(blank=False, null=False)
    type = models.PositiveSmallIntegerField(choices=TYPES, editable=True, null=False, blank=False)

    class Meta:
        unique_together = ('name', 'type')

    def __str__(self) -> str:
        return f'{self.name}'


class VersionManager(models.Manager):
    @property
    def listed(self):
        return self.filter(file__status=FILE_STATUS_CHOICES.APPROVED)

    @property
    def unlisted(self):
        return self.exclude(file__status=FILE_STATUS_CHOICES.APPROVED)

    def update_or_create(self, *args, **kwargs):
        # Stash the ManyToMany to be created after the Version has a valid ID already
        permissions = kwargs.pop('permissions', [])
        licenses = kwargs.pop('licenses', [])
        tags = kwargs.pop('tags', [])

        version, result = super().update_or_create(*args, **kwargs)

        # Add the ManyToMany to the already initialized Version
        version.set_initial_licenses(licenses)
        version.set_initial_permissions(permissions)
        version.set_initial_tags(tags)
        return version, result


class Version(CreatedModifiedMixin, RatingMixin, TrackChangesMixin, models.Model):
    track_changes_to_fields = {
        'blender_version_min',
        'blender_version_max',
        'permissions',
        'version',
        'licenses',
        'file',
        'tagline',
    }

    extension = models.ForeignKey(
        'extensions.Extension', related_name='versions', on_delete=models.CASCADE
    )
    file = models.OneToOneField(
        'files.File', related_name='version', on_delete=models.CASCADE, null=False, blank=False
    )

    version = extensions.fields.VersionStringField(
        max_length=255,
        default='0.1.0',
        null=False,
        blank=False,
        help_text=(
            'Current (latest) version of extension. '
            'The value is taken from <code>"version"</code> in the manifest.'
        ),
    )
    blender_version_min = extensions.fields.VersionStringField(
        max_length=64,
        default='2.93.0',
        null=False,
        blank=False,
        help_text=(
            'Minimum version of Blender this extension is compatible with. '
            'The value is taken from <code>"blender_version_min"</code> in the manifest file.'
        ),
    )
    blender_version_max = extensions.fields.VersionStringField(
        max_length=64,
        null=True,
        blank=True,
        help_text=(
            'Maximum version of Blender this extension was tested and is compatible with. '
            'The value is taken from <code>"blender_version_max"</code> in the manifest file.'
        ),
    )
    licenses = models.ManyToManyField(License, related_name='versions', blank=False)

    tags = models.ManyToManyField(Tag, related_name='versions', blank=True)

    schema_version = extensions.fields.VersionStringField(
        max_length=64,
        null=False,
        blank=False,
        default='1.0.0',
        help_text=('Specification version the manifest file is following.'),
    )

    tagline = models.CharField(
        max_length=64, null=False, blank=False, help_text='A very short description.'
    )

    permissions = models.ManyToManyField(VersionPermission, related_name='versions', blank=True)

    release_notes = models.TextField(help_text=common.help_texts.markdown, blank=True)

    average_score = models.FloatField(max_length=255, default=0, null=False)
    download_count = models.PositiveIntegerField(default=0)

    objects = VersionManager()

    class Meta:
        ordering = ['-date_created']
        constraints = [
            models.UniqueConstraint(
                fields=('extension', 'version'),
                name='version_extension_version_key',
            ),
        ]

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

    def set_initial_permissions(self, _permissions):
        if not _permissions:
            return

        for permission_name in _permissions:
            permission = VersionPermission.get_by_name(permission_name)

            # Just ignore versions that are incompatible.
            if not permission:
                continue

            self.permissions.add(permission)

    def set_initial_licenses(self, _licenses):
        if not _licenses:
            return

        for license_slug in _licenses:
            try:
                license = License.get_by_slug(license_slug)
            except models.ObjectDoesNotExist:
                error_message = f'Unsupported license in manifest file: {license_slug}'
                log.error(error_message)
                raise BadRequest(error_message)
            self.licenses.add(license)

    def set_initial_tags(self, _tags):
        if not _tags:
            return

        for tag_name in _tags:
            try:
                tag = Tag.objects.filter(name=tag_name).first()
            except ObjectDoesNotExist:
                error_message = f'Unsupported tag in manifest file: {tag_name}'
                log.error(error_message)
                raise BadRequest(error_message)
            self.tags.add(tag)

    def __str__(self) -> str:
        return f'{self.extension} v{self.version}'

    @property
    def is_listed(self):
        # To be public, version file must have a public status.
        return self.file is not None and self.file.status == self.file.STATUSES.APPROVED

    @property
    def cannot_be_deleted_reasons(self) -> List[str]:
        """Return a list of reasons why this version cannot be deleted."""
        reasons = []
        if self.is_listed:
            reasons.append('version_is_listed')
        if self.ratings.count() > 0:
            reasons.append('version_has_ratings')
        return reasons

    @property
    def pending_rejection(self):
        try:
            return self.reviewerflags.pending_rejection
        except VersionReviewerFlags.DoesNotExist:
            return None

    @property
    def pending_rejection_by(self):
        try:
            return self.reviewerflags.pending_rejection_by
        except VersionReviewerFlags.DoesNotExist:
            return None

    @property
    def download_name(self) -> str:
        """Return a file name for downloads."""
        replace_char = f'{self}'.replace('.', '-')
        return f'{utils.slugify(replace_char)}{self.file.suffix}'

    @property
    def downloadable_signed_url(self) -> str:
        # TODO: actual signed URLs?
        return self.file.source.url

    @property
    def download_url(self) -> str:
        return reverse(
            'extensions:version-download',
            kwargs={
                'type_slug': self.extension.type_slug,
                'slug': self.extension.slug,
                'version': self.version,
            },
        )

    def get_delete_url(self) -> str:
        return reverse(
            'extensions:version-delete',
            kwargs={
                'type_slug': self.extension.type_slug,
                'slug': self.extension.slug,
                'pk': self.pk,
            },
        )

    @property
    def update_url(self) -> str:
        return reverse(
            'extensions:version-update',
            kwargs={
                'type_slug': self.extension.type_slug,
                'slug': self.extension.slug,
                'pk': self.pk,
            },
        )


class Maintainer(CreatedModifiedMixin, models.Model):
    extension = models.ForeignKey(Extension, on_delete=models.CASCADE)
    user = models.ForeignKey(User, on_delete=models.CASCADE)
    role = models.SmallIntegerField(default=AUTHOR_ROLE_DEV, choices=AUTHOR_ROLE_CHOICES)
    position = models.IntegerField(default=0)

    class Meta:
        constraints = [
            models.UniqueConstraint(fields=('extension', 'user'), name='maintainer_extension_user'),
        ]


class Preview(CreatedModifiedMixin, models.Model):
    extension = models.ForeignKey(Extension, on_delete=models.CASCADE)
    file = models.ForeignKey(
        'files.File', related_name='extension_preview', on_delete=models.CASCADE
    )
    caption = models.CharField(max_length=255, default='', null=False, blank=True)
    position = models.IntegerField(default=0)

    class Meta:
        ordering = ('position', 'date_created')
        unique_together = [['extension', 'file']]

    @property
    def cannot_be_deleted_reasons(self) -> List[str]:
        """Return a list of reasons why this preview cannot be deleted."""
        return []


class ExtensionReviewerFlags(models.Model):
    extension = models.OneToOneField(
        Extension, primary_key=True, on_delete=models.CASCADE, related_name='reviewerflags'
    )
    needs_admin_code_review = models.BooleanField(default=False)
    auto_approval_disabled = models.BooleanField(default=False)
    auto_approval_disabled_unlisted = models.BooleanField(default=None, null=True)
    auto_approval_disabled_until_next_approval = models.BooleanField(default=None, null=True)
    auto_approval_disabled_until_next_approval_unlisted = models.BooleanField(
        default=None, null=True
    )
    auto_approval_delayed_until = models.DateTimeField(default=None, null=True)
    notified_about_auto_approval_delay = models.BooleanField(default=None, null=True)
    notified_about_expiring_delayed_rejections = models.BooleanField(default=None, null=True)


class VersionReviewerFlags(models.Model):
    version = models.OneToOneField(
        Version,
        primary_key=True,
        on_delete=models.CASCADE,
        related_name='reviewerflags',
    )
    pending_rejection = models.DateTimeField(default=None, null=True, blank=True, db_index=True)
    pending_rejection_by = models.ForeignKey(User, null=True, on_delete=models.CASCADE)

    class Meta:
        constraints = [
            models.CheckConstraint(
                name='pending_rejection_all_none',
                check=(
                    models.Q(
                        pending_rejection__isnull=True,
                        pending_rejection_by__isnull=True,
                    )
                    | models.Q(
                        pending_rejection__isnull=False,
                    )
                ),
            ),
        ]


class ExtensionApprovalsCounter(models.Model):
    """Count the number of times a listed version of an add-on has been approved by a human.

    Reset everytime a listed version is auto-approved for this add-on.

    Holds 2 additional date fields:
    - last_human_review, the date of the last time a human fully reviewed the
      add-on
    """

    extension = models.OneToOneField(Extension, primary_key=True, on_delete=models.CASCADE)
    counter = models.PositiveIntegerField(default=0)
    last_human_review = models.DateTimeField(null=True)

    def __str__(self):
        return '%s: %d' % (str(self.pk), self.counter) if self.pk else ''

    @classmethod
    def increment_for_extension(cls, extension):
        """Increment approval counter for the specified extension.

        Set the last human review date and last content review date to now.
        If an ExtensionApprovalsCounter already exists, it updates it, otherwise it
        creates and saves a new instance.
        """
        now = datetime.now()
        data = {
            'counter': 1,
            'last_human_review': now,
        }
        obj, created = cls.objects.get_or_create(extension=extension, defaults=data)
        if not created:
            data['counter'] = F('counter') + 1
            obj.update(**data)
        return obj

    @classmethod
    def reset_for_extension(cls, extension):
        """Reset the approval counter (but not the dates) for the specified extension."""
        obj, created = cls.objects.update_or_create(extension=extension, defaults={'counter': 0})
        return obj