refactor: reduce redundancy in feed gen

This commit is contained in:
Leon
2025-09-21 19:27:18 +02:00
parent 7de80c6d57
commit a5d511eaf3

View File

@@ -1,3 +1,5 @@
from typing import List
from dateutil import tz
from feedgen.feed import FeedGenerator
from sqlalchemy.orm import Session
@@ -5,6 +7,45 @@ from sqlalchemy.orm import Session
from app.core.config import settings
from app.crud.entries import get_all_entries, get_entries_by_newsletter
from app.crud.newsletters import get_newsletter_by_identifier
from app.models.entries import Entry
def _create_feed_generator(
feed_id: str, title: str, feed_url: str, description: str
) -> FeedGenerator:
"""Initialize and configures a FeedGenerator instance."""
logo_url = f"{settings.app_base_url}/logo.png"
icon_url = f"{settings.app_base_url}/favicon.ico"
fg = FeedGenerator()
fg.id(feed_id)
fg.title(title)
fg.logo(logo_url)
fg.icon(icon_url)
fg.link(href=feed_url, rel="self")
fg.link(href=f"{settings.app_base_url}/", rel="alternate")
fg.description(description)
return fg
def _add_entries_to_feed(
fg: FeedGenerator, entries: List[Entry], is_master_feed: bool = False
):
"""Add a list of entries to a FeedGenerator instance."""
for entry in entries:
fe = fg.add_entry()
fe.id(f"urn:letterfeed:entry:{entry.id}")
fe.title(
f"[{entry.newsletter.name}] {entry.subject}"
if is_master_feed
else entry.subject
)
fe.content(entry.body, type="html")
if entry.received_at.tzinfo is None:
timezone_aware_received_at = entry.received_at.replace(tzinfo=tz.tzutc())
fe.published(timezone_aware_received_at)
else:
fe.published(entry.received_at)
def generate_feed(db: Session, feed_identifier: str):
@@ -16,29 +57,17 @@ def generate_feed(db: Session, feed_identifier: str):
entries = get_entries_by_newsletter(db, newsletter.id)
feed_url = f"{settings.app_base_url}/feeds/{newsletter.slug or newsletter.id}"
logo_url = f"{settings.app_base_url}/logo.png"
icon_url = f"{settings.app_base_url}/favicon.ico"
fg = FeedGenerator()
fg.id(f"urn:letterfeed:newsletter:{newsletter.id}")
fg.title(newsletter.name)
fg.logo(logo_url)
fg.icon(icon_url)
fg.link(href=feed_url, rel="self")
fg.link(href=f"{settings.app_base_url}/", rel="alternate")
sender_emails = ", ".join([s.email for s in newsletter.senders])
fg.description(f"A feed of newsletters from {sender_emails}")
description = f"A feed of newsletters from {sender_emails}"
for entry in entries:
fe = fg.add_entry()
fe.id(f"urn:letterfeed:entry:{entry.id}")
fe.title(entry.subject)
fe.content(entry.body, type="html")
if entry.received_at.tzinfo is None:
timezone_aware_received_at = entry.received_at.replace(tzinfo=tz.tzutc())
fe.published(timezone_aware_received_at)
else:
fe.published(entry.received_at)
fg = _create_feed_generator(
feed_id=f"urn:letterfeed:newsletter:{newsletter.id}",
title=newsletter.name,
feed_url=feed_url,
description=description,
)
_add_entries_to_feed(fg, entries)
return fg.atom_str(pretty=True)
@@ -48,27 +77,14 @@ def generate_master_feed(db: Session):
entries = get_all_entries(db)
feed_url = f"{settings.app_base_url}/feeds/all"
logo_url = f"{settings.app_base_url}/logo.png"
icon_url = f"{settings.app_base_url}/favicon.ico"
fg = FeedGenerator()
fg.id("urn:letterfeed:master")
fg.title("LetterFeed: All Newsletters")
fg.logo(logo_url)
fg.icon(icon_url)
fg.link(href=feed_url, rel="self")
fg.link(href=f"{settings.app_base_url}/", rel="alternate")
fg.description("A master feed of all your newsletters.")
fg = _create_feed_generator(
feed_id="urn:letterfeed:master",
title="LetterFeed: All Newsletters",
feed_url=feed_url,
description="A master feed of all your newsletters.",
)
for entry in entries:
fe = fg.add_entry()
fe.id(f"urn:letterfeed:entry:{entry.id}")
fe.title(f"[{entry.newsletter.name}] {entry.subject}")
fe.content(entry.body, type="html")
if entry.received_at.tzinfo is None:
timezone_aware_received_at = entry.received_at.replace(tzinfo=tz.tzutc())
fe.published(timezone_aware_received_at)
else:
fe.published(entry.received_at)
_add_entries_to_feed(fg, entries, is_master_feed=True)
return fg.atom_str(pretty=True)