from django.contrib.syndication.views import Feed from django.utils.feedgenerator import Rss201rev2Feed from django.core.cache import cache from django.conf import settings from django.http import HttpResponse import hashlib import re import logging from app.bridge.models import bridge_urls_q from app.feeds.models import Post try: from orgpython import to_html as org_to_html HAS_ORGPYTHON = True except ImportError: HAS_ORGPYTHON = False logger = logging.getLogger(__name__) class CustomRss201rev2Feed(Rss201rev2Feed): """Custom RSS generator that adds author field properly""" def add_item_elements(self, handler, item): """Add item elements including author""" super().add_item_elements(handler, item) # Add author field if provided if item.get("author"): handler.addQuickElement("author", item["author"]) class LatestPostsFeed(Feed): """RSS feed for latest posts from Org Social Relay""" feed_type = CustomRss201rev2Feed def __call__(self, request, *args, **kwargs): """Override to add XML caching""" # Build cache key for the XML response tag = request.GET.get("tag", "") feed_url = request.GET.get("feed", "") cache_parts = ["rss_xml"] if tag: cache_parts.append(f"tag_{hashlib.md5(tag.encode()).hexdigest()[:8]}") elif feed_url: cache_parts.append(f"feed_{hashlib.md5(feed_url.encode()).hexdigest()[:8]}") else: cache_parts.append("all") cache_key = "_".join(cache_parts) # Try to get cached XML cached_xml = cache.get(cache_key) if cached_xml is not None: return HttpResponse( cached_xml, content_type="application/rss+xml; charset=utf-8" ) # Generate feed normally response = super().__call__(request, *args, **kwargs) # Cache the XML response for 5 minutes (300 seconds) if response.status_code == 200: cache.set(cache_key, response.content, 300) return response def get_object(self, request): """Process query parameters""" tag = request.GET.get("tag") feed_url = request.GET.get("feed") return {"tag": tag, "feed": feed_url} def title(self, obj): """Generate feed title based on filters""" if obj["tag"]: return f"Org Social Relay - Posts tagged with '{obj['tag']}'" elif obj["feed"]: return f"Org Social Relay - Posts from {obj['feed']}" return "Org Social Relay - Latest Posts" def link(self, obj): """Generate feed link""" site_domain = settings.SITE_DOMAIN protocol = "https" if not settings.DEBUG else "http" base_url = f"{protocol}://{site_domain}/rss.xml" if obj["tag"]: return f"{base_url}?tag={obj['tag']}" elif obj["feed"]: return f"{base_url}?feed={obj['feed']}" return base_url def description(self, obj): """Generate feed description based on filters""" if obj["tag"]: return f"Latest posts from Org Social Relay tagged with '{obj['tag']}'" elif obj["feed"]: return f"Latest posts from {obj['feed']} on Org Social Relay" return "Latest posts from all registered feeds on Org Social Relay" def items(self, obj): """Return items for the feed, limited to 200 posts""" # Build cache key based on filters cache_parts = ["rss_feed"] if obj["tag"]: cache_parts.append( f"tag_{hashlib.md5(obj['tag'].encode()).hexdigest()[:8]}" ) elif obj["feed"]: cache_parts.append( f"feed_{hashlib.md5(obj['feed'].encode()).hexdigest()[:8]}" ) else: cache_parts.append("all") cache_key = "_".join(cache_parts) cached_posts = cache.get(cache_key) if cached_posts is not None: return cached_posts # Build query - exclude posts with empty content (reactions, votes, etc.) posts_query = ( Post.objects.select_related("profile") .exclude(content__isnull=True) .exclude(content__exact="") .order_by("-created_at") ) # Apply filters if obj["tag"]: # Search by specific tag (exact word match, case insensitive) tag_escaped = re.escape(obj["tag"]) tag_pattern = rf"(^|[\s]){tag_escaped}([\s]|$)" posts_query = posts_query.filter(tags__iregex=tag_pattern).exclude( bridge_urls_q("profile__feed") ) elif obj["feed"]: # Filter by author feed. An explicitly requested bridge feed is # still served: only aggregate feeds must leave bridges out. posts_query = posts_query.filter(profile__feed=obj["feed"]) else: posts_query = posts_query.exclude(bridge_urls_q("profile__feed")) # Fetch more than 200 to account for filtering whitespace-only posts posts_raw = list(posts_query[:250]) # Filter out posts with only whitespace content posts = [p for p in posts_raw if p.content and p.content.strip()] # Limit to 200 posts as per specification posts = posts[:200] # Cache permanently (will be cleared by scan_feeds task) cache.set(cache_key, posts, None) return posts def item_title(self, item): """Generate item title: date + nick""" # Format: "2025-11-15 - username" date_str = item.created_at.strftime("%Y-%m-%d") return f"{date_str} - {item.profile.nick}" def item_description(self, item): """Return full post content as description, converted from Org to HTML""" if not item.content: return "" # Convert Org mode content to HTML if org-python is available if HAS_ORGPYTHON: try: # Posts in Org Social start at level 3 (***), but in RSS context # they should start at level 1. Remove 2 asterisks from headings # before conversion: *** -> *, **** -> **, etc. content = re.sub( r"^(\*{2,})", lambda m: "*" * (len(m.group(1)) - 2), item.content, flags=re.MULTILINE, ) # Convert org-mode to HTML # We disable toc as we're showing individual posts # highlight=True enables syntax highlighting for code blocks html_content = org_to_html(content, toc=False, highlight=True) return html_content except Exception as e: logger.warning( f"Failed to convert Org to HTML for post {item.post_id}: {e}" ) # Fallback to plain text return item.content # If org-python is not available, return plain text return item.content def item_link(self, item): """Generate item link (post URL)""" return f"{item.profile.feed}#{item.post_id}" def item_author_name(self, item): """Return author name""" return item.profile.nick def item_author_email(self, item): """Return author email - using feed URL as identifier""" # RSS 2.0 requires email format for author, but we don't have emails # We'll return None and handle it differently return None def item_extra_kwargs(self, item): """Add custom author field""" return {"author": item.profile.nick} def item_pubdate(self, item): """Return publication date""" return item.created_at def item_categories(self, item): """Return post tags as categories""" if item.tags: return item.tags.split() return [] def item_guid(self, item): """Return unique identifier for the item""" return f"{item.profile.feed}#{item.post_id}" def item_guid_is_permalink(self, item): """The GUID is a permalink""" return True