PYRSS-Website/apps/home/models.py

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

import os
import logging
from uuid import uuid4

import httpx
import feedparser
from django.db import models
from django.utils import timezone
from django.dispatch import receiver
from django.db.models.signals import pre_save
from django.utils.translation import gettext_lazy as _
from django.core.exceptions import ValidationError
from django.core.validators import URLValidator
from django.core.files.storage import FileSystemStorage
from django.utils.deconstruct import deconstructible
from asgiref.sync import sync_to_async, async_to_sync

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):
        if self.exists(name):
            os.remove(os.path.join(self.location, name))

        return name


@deconstructible
class Subscription_IconPathGenerator:
    """Icon path generator for Subscriptions."""

    def __call__(self, instance, filename: str) -> str:
        return os.path.join("subscriptions", str(instance.uuid), "icon.webp")


class DiscordChannel(models.Model):
    """Represents a Discord Channel."""

    id = models.PositiveBigIntegerField(
        verbose_name=_("id"),
        help_text=_("Unique identifier of the channel, provided by Discord."),
        primary_key=True
    )
    creation_datetime = models.DateTimeField(
        verbose_name=_("creation datetime"),
        help_text=_("when this instance was created."),
        default=timezone.now,
        editable=False
    )

    class Meta:
        verbose_name = _("discord channel")
        verbose_name_plural = _("discord channels")
        get_latest_by = "-creation_datetime"

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


class Subscription(models.Model):
    """Stores relevant data for a user submitted RSS Feed."""

    uuid = models.UUIDField(
        primary_key=True,
        default=uuid4,
        editable=False
    )
    name = models.CharField(
        verbose_name=_("name"),
        help_text=_("Reference name for this subscription (max %(max_length)s 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=Subscription_IconPathGenerator(),
        storage=OverwriteStorage(),
        default="../static/images/defaultuser.webp",
        null=True,
        blank=True
    )
    creation_datetime = models.DateTimeField(
        verbose_name=_("creation datetime"),
        help_text=_("when this instance was created."),
        default=timezone.now,
        editable=False
    )
    server = models.PositiveBigIntegerField(
        verbose_name=_("server id"),
        help_text=_("Identifier for the discord server that owns this subscription.")
    )
    channels = models.ManyToManyField(
        verbose_name=_("channels"),
        help_text=_("List of Discord Channels acting as targets for subscription content."),
        to=DiscordChannel,
    )

    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)


class TrackedContent(models.Model):
    """Tracks content shared from an RSS Feed, to prevent duplicates."""

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