rss_processor.py

import uuid
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from xml.sax.saxutils import escape

import feedparser
import listparser

from config import RSS_FEED_CONFIG


@dataclass
class Article:
    title: str
    url: str
    guid: str
    published_at: datetime
    feed_name: str
    feed_url: str
    summary: str | None = None
    relevance_score: float = 0.0

    def __post_init__(self):
        self.title = self.title.strip()
        self.url = self.url.strip()

    @classmethod
    def from_feed_entry(cls, entry, feed_name: str, feed_url: str) -> "Article":
        return cls(
            title=escape(entry.get("title", "Untitled")),
            url=escape(entry.get("link", "")),
            guid=entry.get("id") or entry.get("link") or str(uuid.uuid4()),
            published_at=cls._parse_date(entry),
            feed_name=feed_name,
            feed_url=feed_url,
            summary=cls._parse_summary(entry),
        )

    @staticmethod
    def _parse_date(entry) -> datetime:
        # Prefer published over updated; fall back to now if neither present
        t = entry.get("published_parsed") or entry.get("updated_parsed")
        if t:
            return datetime(*t[:6], tzinfo=timezone.utc)
        return datetime.now(timezone.utc)

    @staticmethod
    def _parse_summary(entry) -> str | None:
        # content[] (full body) takes priority over summary (excerpt)
        if entry.get("content"):
            return entry["content"][0].get("value")
        return entry.get("summary") or entry.get("description")


def fetch_data_from_opml(
    article_age_limit: timedelta = timedelta(days=1),
    opml_file: str = "./feeds.opml",
):
    """
    Parses an OPML file to fetch and filter RSS feed entries based on age.

    This function separates feeds into standard and priority categories. It filters out
    older articles from both categories based on the `article_age_limit`. Standard feed
    headlines are logged to `feeds_to_filter.txt`.

    Args:
        priority_feed_urls (list[str]): A list of feed URLs to be treated with priority.
        article_age_limit (timedelta): The maximum age of an article to be included.
            Defaults to 1 day.
        opml_file (str): Path to the OPML file containing feed URLs.

    Returns:
        dict: A dictionary containing:
            - "feeds": List of parsed feed objects for standard feeds.
            - "priority_feeds": List of parsed feed objects for priority feeds.
            - "titles": List of entry titles from standard feeds within the age limit.
    """
    result = listparser.parse(open(opml_file).read())

    # Filter out priority feeds from the main list parsed from OPML
    feed_urls = [feed["url"] for feed in result["feeds"]]

    data = []

    # Process standard feeds and log recent headlines to a file
    with open("feeds_to_filter.txt", "w") as f:
        for feed_url in feed_urls:
            feed = feedparser.parse(feed_url)
            feed_name = feed.feed.get("title", feed_url)  # "Ars Technica - All content"

            for entry in feed.entries:
                is_recent = True
                if entry.get("updated_parsed"):
                    published_time = entry.updated_parsed
                    # Check if the article is within the allowed age limit
                    if (
                        datetime.now() - datetime(*published_time[0:6])
                    ) >= article_age_limit:
                        is_recent = False

                if is_recent:
                    f.write(f"{feed_url} {entry.description}\n")
                    entry = Article.from_feed_entry(
                        entry, feed_name=feed_name, feed_url=feed_url
                    )
                    data.append(entry)

    print("Number of feeds processed:", len(feed_urls))
    print("Number of headlines fetched:", len(data))

    return data


def generate_rss_feed(
    articles: list[Article], output_file: str = "filtered_articles.rss"
):
    """
    Generate an RSS feed from a list of Article instances and save it to a file.

    Args:
        articles (list[Article]): List of Article instances to include in the RSS feed
        output_file (str): Path to the output RSS file (default: "filtered_articles.rss")
    """
    # RSS feed header
    rss_content = f"""<?xml version="1.0" encoding="UTF-8"?>
<rss version="2.0" xmlns:atom="http://www.w3.org/2005/Atom">
<channel>
<atom:link href="{RSS_FEED_CONFIG["url"]}" rel="self" type="application/rss+xml" />
<title>{RSS_FEED_CONFIG["title"]}</title>
<description>{RSS_FEED_CONFIG["description"]}</description>
<link>{RSS_FEED_CONFIG["website_url"]}</link>
<image>
    <url>{RSS_FEED_CONFIG["image_url"]}</url>
    <title>{RSS_FEED_CONFIG["title"]}</title>
    <link>{RSS_FEED_CONFIG["website_url"]}</link>
    <width>144</width>
    <height>144</height>
</image>
<language>en-us</language>
<lastBuildDate>{{}}</lastBuildDate>
""".format(datetime.now(timezone.utc).strftime("%a, %d %b %Y %H:%M:%S %z"))

    # Add each article as an item in the RSS feed
    for article in articles:
        # Format the publication date
        pub_date = (
            article.published_at.strftime("%a, %d %b %Y %H:%M:%S %z")
            if article.published_at
            else datetime.now(timezone.utc).strftime("%a, %d %b %Y %H:%M:%S %z")
        )

        # Escape special XML characters
        title = escape(article.title or "")
        guid = escape(article.guid or "")
        url = escape(article.url or "")

        raw_summary = article.summary or ""
        if "<![CDATA[" in raw_summary:
            # Remove the opening and closing CDATA tags if they exist
            raw_summary = raw_summary.replace("<![CDATA[", "").replace("]]>", "")

        rss_content += f"""
<item>
<title>{title}</title>
<link>{url}</link>
<guid isPermaLink="false">{guid}</guid>
<pubDate>{pub_date}</pubDate>
<description><![CDATA[{raw_summary}]]></description>
</item>"""

    # RSS feed footer
    rss_content += """
</channel>
</rss>"""

    # Write the RSS feed to a file
    with open(output_file, "w", encoding="utf-8") as f:
        f.write(rss_content)

    print(f"RSS feed generated and saved to {output_file}")