extensions-website/extensions/models.py

from typing import List
from statistics import median
import logging

from django.contrib.auth import get_user_model
from django.core.exceptions import ObjectDoesNotExist, BadRequest
from django.db import models, transaction
from django.db.models import 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,
    EXTENSION_TYPE_SLUGS_SINGULAR,
    FILE_STATUS_CHOICES,
)
from files.models import File
from reviewers.models import ApprovalActivity
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 Platform(CreatedModifiedMixin, models.Model):
    name = models.CharField(max_length=128, null=False, blank=False, unique=True)
    slug = models.SlugField(
        blank=False,
        null=False,
        help_text='A platform tag, see https://packaging.python.org/en/latest/specifications/platform-compatibility-tags/',  # noqa
        unique=True,
    )

    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_filter(self, user):
        filter = Q(maintainer__user_id=user.pk)
        user_teams = user.teams.all()
        if user_teams:
            filter = filter | Q(team__in=[t.pk for t in user_teams])
        return filter

    def authored_by(self, user):
        return self.filter(self._authored_by_filter(user)).distinct()

    def listed_or_authored_by(self, user):
        return self.filter(
            Q(status=self.model.STATUSES.APPROVED) | self._authored_by_filter(user)
        ).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,
    )
    latest_version = models.ForeignKey(
        'Version',
        null=True,
        blank=True,
        on_delete=models.SET_NULL,
        related_name='latest_version_of',
    )

    featured_image = models.ForeignKey(
        'files.File',
        related_name='featured_image_of',
        null=True,
        blank=False,
        on_delete=models.PROTECT,
        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.ForeignKey(
        'files.File',
        related_name='icon_of',
        null=True,
        blank=False,
        on_delete=models.PROTECT,
        help_text="A 256 x 256 PNG icon representing this extension.",
    )
    previews = FilterableManyToManyField('files.File', through='Preview', related_name='extensions')

    status = models.PositiveSmallIntegerField(choices=STATUSES, default=STATUSES.DRAFT)
    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:
        indexes = [
            models.Index(fields=['is_listed', 'type', 'average_score']),
            models.Index(fields=['is_listed', 'type', 'date_approved']),
            models.Index(fields=['is_listed', 'type', 'download_count']),
            models.Index(fields=['is_listed', 'type', 'name']),
        ]
        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 type_slug_singular(self) -> str:
        return EXTENSION_TYPE_SLUGS_SINGULAR[self.type]

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

    def update_metadata_from_version(self, version):
        update_fields = set()
        metadata = version.file.metadata
        # check if we can also update name
        # if we are uploading a new version, we have just validated and don't expect a clash,
        # but if a version is being deleted, we want to rollback to a name from an older version,
        # which may by clashing now, and we can't do anything about it
        name = metadata.get('name')
        if not self.__class__.objects.filter(name=name).exclude(pk=self.pk).exists():
            update_fields.add('name')
        for field in ['support', 'website']:
            # if a field is missing from manifest, don't reset the corresponding extension field
            if metadata.get(field):
                update_fields.add(field)
        for field in update_fields:
            setattr(self, field, metadata.get(field))
        self.save(update_fields=update_fields)

    @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]

    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.DRAFT
            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 or is a member of the team."""
        if user is None or user.is_anonymous:
            return False
        if user in self.authors.all():
            return True
        if self.team and user in self.team.users.all():
            return True
        return False

    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

    @transaction.atomic
    def update_latest_version(self, skip_version=None):
        versions = self.versions.select_related('file').order_by('-date_created')
        latest_version = None
        for version in versions:
            if skip_version and version == skip_version:
                continue
            if version.file.status not in self.valid_file_statuses:
                continue
            latest_version = version
            break
        self.latest_version = latest_version
        self.save(update_fields={'latest_version'})
        if latest_version:
            self.update_metadata_from_version(latest_version)

    def update_is_listed(self):
        should_be_listed = (
            self.status == self.STATUSES.APPROVED
            and self.versions.filter(file__status=File.STATUSES.APPROVED).count() > 0
        )

        # this method is called from post_save signal, this early return should prevent a loop
        if self.is_listed == should_be_listed:
            return

        self.is_listed = should_be_listed
        update_fields = {'is_listed'}
        if self.status == self.STATUSES.APPROVED and not should_be_listed:
            self.status = self.STATUSES.DRAFT
            update_fields.add('status')

        self.save(update_fields=update_fields)


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
        licenses = kwargs.pop('licenses', [])
        permissions = kwargs.pop('permissions', [])
        platforms = kwargs.pop('platforms', [])
        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_platforms(platforms)
        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)
    platforms = models.ManyToManyField(Platform, 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_platforms(self, _platforms):
        if not _platforms:
            return

        for slug in _platforms:
            platform = Platform.get_by_slug(slug)
            self.platforms.add(platform)

    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.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 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

    def download_url(self, append_repository=True) -> str:
        filename = f'{self.extension.type_slug_singular}-{self.extension.slug}-v{self.version}.zip'
        download_url = reverse(
            'extensions:version-download',
            kwargs={
                'type_slug': self.extension.type_slug,
                'slug': self.extension.slug,
                'version': self.version,
                'filename': filename,
            },
        )
        if append_repository:
            download_url += '?repository=/api/v1/extensions/'
        return download_url

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

    @transaction.atomic
    def save(self, *args, **kwargs):
        is_new = self.pk is None
        update_fields = kwargs.get('update_fields', None)
        was_changed, old_state = self.pre_save_record(update_fields=update_fields)
        self.record_status_change(was_changed, old_state, **kwargs)

        super().save(*args, **kwargs)

        if not is_new:
            return

        # auto approving our file if extension is already listed (i.e. have been approved)
        if self.extension.is_listed:
            args = {'f_id': self.file.pk, 'pk': self.pk, 's': self.file.source.name}
            log.info('Auto-approving file pk=%(f_id)s of Version pk=%(pk)s source=%(s)s', args)
            self.file.status = File.STATUSES.APPROVED
            self.file.save(update_fields={'status', 'date_modified'})

            ApprovalActivity(
                type=ApprovalActivity.ActivityType.UPLOADED_NEW_VERSION,
                user=self.file.user,
                extension=self.extension,
                message=f'uploaded new version: {self.version}',
            ).save()

        self.extension.update_latest_version()

    @transaction.atomic
    def delete(self, *args, **kwargs):
        if self == self.extension.latest_version:
            # make sure self is not a candidate for latest_version, since it's being deleted
            self.extension.update_latest_version(skip_version=self)
        super().delete(*args, **kwargs)


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)
    # Files can potentially be referenced by different extensions
    file = models.ForeignKey('files.File', on_delete=models.PROTECT)
    caption = models.CharField(max_length=255, default='', null=False, blank=True)
    position = models.IntegerField(default=0)

    class Meta:
        ordering = ('position', 'date_created')
        # We don't want to have duplicate previews on the same extension
        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 []

    def __str__(self) -> str:
        return f'Preview {self.pk} of extension {self.extension_id}: {self.file}'