# -*- 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.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 RSSFeedIconPathGenerator: """Icon path generator for RSS Feed.""" def __call__(self, instance, filename: str) -> str: return os.path.join("rssfeed", str(instance.uuid), "icon.webp") @async_to_sync async def validate_rss_url(url: str): log.debug("validating RSS url: %s" % url) try: validator = URLValidator(schemes=["https"]) validator(url) except ValidationError as exc: raise ValidationError("URL scheme must be 'https'") from exc async with httpx.AsyncClient() as client: response = await client.get(url) try: response.raise_for_status() except httpx.HTTPStatusError as error: log.error(error) raise ValidationError(error) content = response.text feed = feedparser.parse(content) if not feed.version: log.error("not a feed") raise ValidationError(f"{url} does not point to a valid RSS feed") log.debug("success") return True class RSSFeed(models.Model): """Represents an RSS Feed.""" uuid = models.UUIDField(primary_key=True, default=uuid4, editable=False) name = models.CharField( verbose_name=_("name"), help_text=_("a human readable nickname for this item"), null=False, blank=False, max_length=120 ) url = models.URLField( verbose_name=_("url"), help_text=_("url to the RSS feed"), validators=[validate_rss_url], ) image = models.ImageField( verbose_name=_("image"), help_text=_("image of the RSS feed"), upload_to=RSSFeedIconPathGenerator(), storage=OverwriteStorage(), null=True, blank=True ) created_at = models.DateTimeField( verbose_name=_("creation date & time"), help_text=_("when this item was created"), default=timezone.now, editable=False ) class Meta: verbose_name = _("RSS Feed") verbose_name_plural = _("RSS Feeds") def __str__(self): return self.name class FeedChannel(models.Model): """Represents a Discord Text Channel to act as a feed.""" uuid = models.UUIDField(primary_key=True, default=uuid4, editable=False) channel_id = models.PositiveBigIntegerField( verbose_name=_("discord channel id"), help_text=_("the id of the discord channel"), null=False, blank=False ) feeds = models.ManyToManyField( to=RSSFeed, verbose_name=_("feeds"), help_text=_("the feeds to include in this item") ) created_at = models.DateTimeField( verbose_name=_("creation date & time"), help_text=_("when this item was created"), default=timezone.now, editable=False ) class Meta: verbose_name = _("Feed Channel") verbose_name_plural = _("Feed Channels") def __str__(self): return str(self.channel_id)