PYRSS-Website/apps/home/models.py

# -*- encoding: utf-8 -*-

import logging
from uuid import uuid4
from pathlib import Path

from django.db import models
from django.utils import timezone
from django.dispatch import receiver
from django.db.utils import IntegrityError
from django.db.models.signals import pre_save
from django.utils.translation import gettext_lazy as _
from django.core.files.storage import FileSystemStorage
from django.utils.deconstruct import deconstructible

log = logging.getLogger(__name__)


class OverwriteStorage(FileSystemStorage):
    """
    Storage class that allows overriding files, instead of django appending random
    characters to prevent conflicts.
    """

    def get_available_name(self, name, max_length=None) -> str:
        if self.exists(name):
            (Path(self.location) / name).unlink()

        return name


@deconstructible
class IconPathGenerator:
    """
    Icon path generator.
    """

    def __call__(self, instance, filename: str) -> Path:
        return Path(instance.__class__.__name__.lower()) / str(instance.uuid) / "icon.webp"


class Subscription(models.Model):
    """
    Represents a stored RSS Feed.
    """

    uuid = models.UUIDField(
        primary_key=True,
        default=uuid4,
        editable=False
    )

    # Name attribute acts as a human readable identification and search option.
    name = models.CharField(
        verbose_name=_("name"),
        help_text=_("Reference name for this subscription (max 32 chars)."),
        max_length=32,
        null=False,
        blank=False
    )

    rss_url = models.URLField(
        verbose_name=_("rss url"),
        help_text=_("URL of the subscribed to RSS feed.")
    )

    image = models.ImageField(
        verbose_name=_("image"),
        help_text=_("image of the RSS feed."),
        upload_to=IconPathGenerator(),
        storage=OverwriteStorage(),
        default="../static/images/defaultuser.webp",
        null=True,
        blank=True
    )

    # Discord Server ID
    server = models.PositiveBigIntegerField(
        verbose_name=_("server id"),
        help_text=_("Identifier for the discord server that owns this subscription.")
    )

    creation_datetime = models.DateTimeField(
        verbose_name=_("creation datetime"),
        help_text=_("when this instance was created."),
        default=timezone.now,
        editable=False
    )

    class Meta:
        verbose_name = "subscription"
        verbose_name_plural = "subscriptions"
        get_latest_by = "-creation_datetime"
        constraints = [
            # Prevent servers from having subscriptions with duplicate names
            models.UniqueConstraint(fields=["name", "server"], name="unique name & server pair")
        ]

    def __str__(self):
        return self.name

    def save(self, *args, **kwargs):
        new_text = "New " if self._state.adding else ""
        log.debug("%sSubscription Saved %s", new_text, self.uuid)
        super().save(*args, **kwargs)

    @property
    def channels(self):
        """
        Returns all SubscriptionChannel objects linked to this Subscription.
        """

        return SubscriptionChannel.objects.filter(subscription=self)

    def save(self, *args, **kwargs):
        if Subscription.objects.filter(server=self.server).count() >= 1:
            raise IntegrityError(f"Subscription limit reached for server '{self.server}'")

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


class SubscriptionChannel(models.Model):
    """
    Represents a Discord TextChannel that should be subject to the content of a Subscription.
    """

    uuid = models.UUIDField(
        primary_key=True,
        default=uuid4,
        editable=False
    )

    # The ID is not auto generated, but rather be obtained from discord.
    id = models.PositiveBigIntegerField(
        verbose_name=_("id"),
        help_text=_("Identifier of the channel, provided by Discord.")
    )

    subscription = models.ForeignKey(
        verbose_name=_("subscription"),
        help_text=_("The subscription of this instance."),
        to=Subscription,
        on_delete=models.CASCADE
    )

    creation_datetime = models.DateTimeField(
        verbose_name=_("creation datetime"),
        help_text=_("when this instance was created."),
        default=timezone.now,
        editable=False
    )

    class Meta:
        verbose_name = "subscription channel"
        verbose_name_plural = "subscription channels"
        get_latest_by = "-creation_date"
        constraints = [
            models.UniqueConstraint(fields=["id", "subscription"], name="unique id & sub pair")
        ]

    def __str__(self):
        return str(self.id)

    def save(self, *args, **kwargs):
        if SubscriptionChannel.objects.filter(subscription=self.subscription).count() >= 4:
            raise IntegrityError(
                f"SubscriptionChannel limit reached for subscription '{self.subscription}'"
            )

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


class TrackedContent(models.Model):
    """
    Tracks content shared from an RSS Feed Subscription.
    Content is tracked to help prevent duplicate content.
    """

    uuid = models.UUIDField(
        primary_key=True,
        default=uuid4,
        editable=False
    )

    subscription = models.ForeignKey(
        verbose_name=_("subscription"),
        help_text=_("The subscription that this content originated from."),
        to=Subscription,
        on_delete=models.CASCADE,
        related_name="tracked_content"
    )

    content_url = models.URLField(
        verbose_name=_("content url"),
        help_text=_("URL of the tracked content.")
    )

    creation_datetime = models.DateTimeField(
        verbose_name=_("creation datetime"),
        help_text=_("when this instance was created."),
        default=timezone.now,
        editable=False
    )

    def __str__(self):
        return str(self.content_url)


@receiver(pre_save, sender=TrackedContent)
def maintain_item_cap(sender, instance, **kwargs):
    """"
    Delete the latest tracked content, if the total under
    the same subscription reaches 100.
    """

    log.debug("checking if tracked content can be deleted.")

    queryset = sender.objects.filter(subscription=instance.subscription)
    total_tracked = queryset.count()

    if total_tracked < 100:
        log.debug("tracked content cannot be deleted, less than 100 items: %s", total_tracked)
        return

    oldest = queryset.earliest()

    log.info("tracked content limit exceeded, deleting oldest item with uuid: %s", oldest.uuid)

    oldest.delete()