diff --git a/backend/app/services/feed_generator.py b/backend/app/services/feed_generator.py index e6fb0b8..879b2c5 100644 --- a/backend/app/services/feed_generator.py +++ b/backend/app/services/feed_generator.py @@ -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)