How to turn any RSS or Atom feed into JSON and CSV with Python

#python #rss #webdev #tutorial

RSS and Atom still power podcasts, changelogs, and "what's new" pages. The formats differ, dates are inconsistent, and descriptions are often HTML. Here's a small Python script (stdlib + requests) that discovers a feed, normalizes items, and writes a CSV.

1. Discover the feed from a website

Many pages advertise their feed in <head>:

import re, requests
from urllib.parse import urljoin, urlparse

UA = {"User-Agent": "rss-tutorial/1.0"}

def discover_feeds(page_url):
    html = requests.get(page_url, headers=UA, timeout=20).text
    found = []
    for m in re.finditer(r'''<link[^>]+rel=["']alternate["'][^>]*>''', html, re.I):
        tag = m.group(0)
        if not re.search(r'type=["\']application/(rss|atom)\+xml["\']', tag, re.I):
            continue
        href = re.search(r'href=["\']([^"\']+)["\']', tag, re.I)
        if href:
            found.append(urljoin(page_url, href.group(1)))
    origin = f"{urlparse(page_url).scheme}://{urlparse(page_url).netloc}"
    for path in ("/feed", "/rss", "/atom.xml", "/feed.xml", "/rss.xml"):
        found.append(origin + path)
    return list(dict.fromkeys(found))

On https://www.nasa.gov/ this finds https://www.nasa.gov/feed/ via <link rel="alternate">. If you already have a feed URL (e.g. https://hnrss.org/frontpage), skip discovery and fetch it directly.

2. Parse RSS 2.0 / Atom into one shape

import gzip, xml.etree.ElementTree as ET
from datetime import datetime, timezone
from email.utils import parsedate_to_datetime
from html import unescape

def _local(tag):
    return tag.split("}", 1)[-1].lower() if tag else ""

def _text(el):
    if el is None: return ""
    parts = [el.text or ""] + [ET.tostring(c, encoding="unicode") + (c.tail or "") for c in el]
    return unescape("".join(parts)).strip()

def _strip_html(s):
    return re.sub(r"\s+", " ", re.sub(r"<[^>]+>", " ", s or "")).strip()

def parse_date(s):
    if not s: return ""
    try: return parsedate_to_datetime(s).astimezone(timezone.utc).isoformat()
    except Exception: pass
    try: return datetime.fromisoformat(s.strip().replace("Z", "+00:00")).astimezone(timezone.utc).isoformat()
    except Exception: return s.strip()

def fetch_xml(url):
    body = requests.get(url, headers=UA, timeout=30, allow_redirects=True).content
    if url.endswith(".gz") or body[:2] == b"\x1f\x8b":
        body = gzip.decompress(body)
    return body

def parse_feed(body):
    root = ET.fromstring(body)
    items = []
    if _local(root.tag) in ("rss", "rdf"):
        channel = root.find("channel") if _local(root.tag) == "rss" else root
        entries = (channel.findall("item") if channel is not None else []) or root.findall(".//item")
        for it in entries:
            get = lambda *names: _text(next((c for c in it if _local(c.tag) in names), None))
            items.append({"title": get("title"), "link": get("link"),
                          "published": parse_date(get("pubdate", "date")),
                          "author": get("author", "creator"),
                          "summary": _strip_html(get("description", "summary", "encoded"))[:300]})
    elif _local(root.tag) == "feed":  # Atom
        for it in [c for c in root if _local(c.tag) == "entry"]:
            title = _text(next((c for c in it if _local(c.tag) == "title"), None))
            link = next((c.attrib.get("href", "") for c in it
                         if _local(c.tag) == "link" and c.attrib.get("rel", "alternate") in ("alternate", "")), "")
            pub = _text(next((c for c in it if _local(c.tag) in ("published", "updated")), None))
            author_el = next((c for c in it if _local(c.tag) == "author"), None)
            author = _text(next((c for c in author_el if _local(c.tag) == "name"), None)) if author_el is not None else ""
            summary = _text(next((c for c in it if _local(c.tag) in ("summary", "content")), None))
            items.append({"title": title, "link": link, "published": parse_date(pub),
                          "author": author, "summary": _strip_html(summary)[:300]})
    else:
        raise ValueError(f"Unknown feed root: {root.tag}")
    return items

https://hnrss.org/frontpage (RSS 2.0) yields 20 normalized rows; https://github.blog/feed/atom/ (Atom) uses the same fields from <entry>, <link href>, and <published>.

3. Write CSV (or JSON)

import csv
rows = parse_feed(fetch_xml("https://hnrss.org/frontpage"))
with open("feed.csv", "w", newline="", encoding="utf-8") as f:
    w = csv.DictWriter(f, fieldnames=["title", "link", "published", "author", "summary"])
    w.writeheader()
    w.writerows(rows)

For JSON: json.dump(rows, open("feed.json", "w"), ensure_ascii=False, indent=2).

Gotchas you'll hit

HTML in descriptions — RSS description / Atom content often contain markup; strip or keep it on purpose.
Missing dates — RSS uses RFC 822 (parsedate_to_datetime); Atom uses ISO-8601; some items have neither.
Atom vs RSS links — Atom: <link href="...">; RSS: element text.
Gzip and redirects — decompress on magic bytes; follow redirects.
XML namespaces — strip {http://...} with _local() so dc:creator and plain title both work.
Guessed paths that 404 — prefer <link rel="alternate"> over /feed fallbacks.

If you don't want to maintain it

Disclosure: I built a hosted version of this, RSS Feed Scraper on Apify, which also handles JSON Feed, keyword/date filters, cross-feed dedupe, and monitoring. Pricing is about $0.30 per 1,000 feed items; failed feeds, filtered items, duplicates, and already-seen items are free. The script above is enough for a handful of feeds on a cron job.

DEV Community