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, RecordDeletionMixin, TrackChangesMixin
from constants.base import (
    AUTHOR_ROLE_CHOICES,
    AUTHOR_ROLE_DEV,
    EXTENSION_STATUS_CHOICES,
    EXTENSION_TYPE_CHOICES,
    EXTENSION_TYPE_SLUGS,
    FILE_STATUS_CHOICES,
)
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 len(
            [
                r
                for r in self.ratings.all()
                if r.text is not None and r.is_listed and r.reply_to is None
            ]
        )

    @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/',
        unique=True,
    )
    url = models.URLField(blank=False, null=False)

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

    @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,
    )

    featured_image = models.OneToOneField(
        'files.File',
        related_name='featured_image_of',
        null=True,
        blank=False,
        on_delete=models.SET_NULL,
        help_text=(
            "Shown by social networks when this extension is shared"
            " (used as `og:image` metadata field)."
            "Should have resolution of at least 1920 x 1080 and aspect ratio of 16:9."
        ),
    )
    icon = models.OneToOneField(
        'files.File',
        related_name='icon_of',
        null=True,
        blank=False,
        on_delete=models.SET_NULL,
        help_text="A 256 x 256 icon representing this extension.",
    )
    previews = FilterableManyToManyField('files.File', through='Preview', related_name='extensions')

    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()

        if self.featured_image:
            self.featured_image.status = FILE_STATUS_CHOICES.APPROVED
            self.save()
        if self.icon:
            self.icon.status = FILE_STATUS_CHOICES.APPROVED
            self.icon.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) -> List['Preview']:
        """Get all preview files, sorted by Preview.position.

        Avoid triggering additional querysets, rely on prefetch_related in the view.
        """
        return [p for p in self.preview_set.all()]

    def get_previews_listed(self) -> List['Preview']:
        """Get publicly listed preview files, sorted by Preview.position."""
        return [p for p in self.get_previews() if p.file.is_listed]

    @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."""
        versions = [
            v for v in self.versions.all() if v.file and v.file.status in self.valid_file_statuses
        ]
        if not versions:
            return None
        versions = sorted(versions, key=lambda v: v.date_created, reverse=True)
        return versions[0]

    @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 user in self.authors.all()

    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()
        )

    def suspicious_files(self):
        versions = self.versions.all()
        return [v.file for v in versions if not v.file.validation.is_ok]

    @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.',
        unique=True,
    )
    help = models.CharField(max_length=128, null=False, blank=False, unique=True)

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

    @classmethod
    def get_by_slug(cls, slug: str):
        return cls.objects.get(slug=slug)


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_slug(permission_name)
            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, RecordDeletionMixin, models.Model):
    extension = models.ForeignKey(Extension, on_delete=models.CASCADE)
    file = models.OneToOneField('files.File', 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