Python script to convert a news sitemap file to a RSS feed

sitemap_to_rss.py

#!/usr/bin/env python3
"""
sitemap_to_rss.py
=================
Convert a sitemap XML file (URL or local path) into an RSS 2.0 feed.

Supported namespaces
--------------------
  • http://www.sitemaps.org/schemas/sitemap/0.9        (standard sitemap)
  • http://www.google.com/schemas/sitemap-news/0.9     (Google News extension)
  • http://www.google.com/schemas/sitemap-image/1.1    (Google Image extension)

Sitemap index files (sitemapindex) are detected automatically; use
--follow-index to recursively fetch and merge all child sitemaps.

Images are emitted as both <enclosure> (first image, RSS 2.0 core) and
<media:content> elements (Media RSS, all images).  Keywords / genres from
Google News appear in <category> elements.

Usage examples
--------------
  python sitemap_to_rss.py https://www.usatoday.com/news-sitemap.xml
  python sitemap_to_rss.py sitemap.xml -o feed.rss
  python sitemap_to_rss.py https://example.com/sitemap.xml \\
      --title "My Feed" --link https://example.com --max-items 50
  python sitemap_to_rss.py https://example.com/sitemap-index.xml --follow-index
"""

from __future__ import annotations

import argparse
import mimetypes
import re
import sys
import urllib.request
import urllib.error
from datetime import datetime, timezone
from email.utils import format_datetime
from pathlib import Path
from typing import Optional
from urllib.parse import urlparse
from xml.dom import minidom
from xml.etree import ElementTree as ET


# ---------------------------------------------------------------------------
# Namespace constants
# ---------------------------------------------------------------------------

NS_SITEMAP = "http://www.sitemaps.org/schemas/sitemap/0.9"
NS_NEWS    = "http://www.google.com/schemas/sitemap-news/0.9"
NS_IMAGE   = "http://www.google.com/schemas/sitemap-image/1.1"
NS_MEDIA   = "http://search.yahoo.com/mrss/"
NS_DC      = "http://purl.org/dc/elements/1.1/"

# Clark-notation helpers
SM  = f"{{{NS_SITEMAP}}}"
N   = f"{{{NS_NEWS}}}"
IMG = f"{{{NS_IMAGE}}}"


# ---------------------------------------------------------------------------
# HTTP / file fetching
# ---------------------------------------------------------------------------

_USER_AGENT = (
    "Mozilla/5.0 (compatible; SitemapToRSS/1.0; "
    "+https://github.com/)"
)


def fetch_xml(source: str) -> bytes:
    """Return raw bytes from a URL or local file path."""
    if source.startswith(("http://", "https://")):
        req = urllib.request.Request(source, headers={"User-Agent": _USER_AGENT})
        try:
            with urllib.request.urlopen(req, timeout=30) as resp:
                return resp.read()
        except urllib.error.HTTPError as exc:
            raise SystemExit(f"HTTP {exc.code} fetching {source}: {exc.reason}") from exc
        except urllib.error.URLError as exc:
            raise SystemExit(f"Network error fetching {source}: {exc.reason}") from exc
    else:
        path = Path(source)
        if not path.is_file():
            raise SystemExit(f"File not found: {source}")
        return path.read_bytes()


def parse_root(source: str) -> ET.Element:
    """Fetch *source* and return the parsed XML root element."""
    raw = fetch_xml(source)
    try:
        return ET.fromstring(raw)
    except ET.ParseError as exc:
        raise SystemExit(f"XML parse error in {source}: {exc}") from exc


# ---------------------------------------------------------------------------
# Date helpers
# ---------------------------------------------------------------------------

_ISO_PATTERNS = [
    "%Y-%m-%dT%H:%M:%S%z",
    "%Y-%m-%dT%H:%M:%S.%f%z",
    "%Y-%m-%d",
]


def parse_date(date_str: str) -> Optional[datetime]:
    """Parse an ISO 8601 / RFC 3339 string into a tz-aware datetime."""
    if not date_str:
        return None
    # Normalise trailing Z → +00:00 for broad Python version support
    s = date_str.strip().replace("Z", "+00:00")
    for fmt in _ISO_PATTERNS:
        try:
            dt = datetime.strptime(s, fmt)
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=timezone.utc)
            return dt
        except ValueError:
            continue
    return None


def to_rss_date(dt: Optional[datetime]) -> str:
    """Format *dt* as an RFC 2822 string suitable for RSS <pubDate>."""
    if dt is None:
        dt = datetime.now(timezone.utc)
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return format_datetime(dt)


# ---------------------------------------------------------------------------
# XML helpers
# ---------------------------------------------------------------------------

def _text(el: Optional[ET.Element], *tags: str) -> str:
    """
    Return stripped text of the first matching child tag, trying each tag in
    *tags* until one is found.  Falls back to "" if none match.

    Supports Clark-notation tags, e.g. "{http://...}name".
    """
    if el is None:
        return ""
    for tag in tags:
        child = el.find(tag)
        if child is not None and child.text:
            return child.text.strip()
    return ""


# ---------------------------------------------------------------------------
# Sitemap parsing
# ---------------------------------------------------------------------------

def _is_index(root: ET.Element) -> bool:
    """Return True when *root* is a <sitemapindex> element."""
    local = root.tag.split("}")[-1] if "}" in root.tag else root.tag
    return local == "sitemapindex"


def _child_sitemap_urls(root: ET.Element) -> list[str]:
    """Extract <loc> values from a sitemapindex root."""
    urls = []
    for sm_el in root.findall(f"{SM}sitemap") + root.findall("sitemap"):
        loc = _text(sm_el, f"{SM}loc", "loc")
        if loc:
            urls.append(loc)
    return urls


def _extract_url_elements(root: ET.Element) -> list[ET.Element]:
    """Return all <url> child elements regardless of namespace prefix."""
    els = root.findall(f"{SM}url")
    if not els:
        els = root.findall("url")
    return els


_EXTRA_MIME: dict[str, str] = {
    # Types not always registered in the system mimetypes database
    ".webp": "image/webp",
    ".avif": "image/avif",
    ".heic": "image/heic",
    ".heif": "image/heif",
    ".jxl":  "image/jxl",
    ".svg":  "image/svg+xml",
}


def _image_mime(url: str) -> str:
    """Guess a MIME type from an image URL extension; fall back to image/jpeg."""
    path = url.split("?")[0].lower()
    for ext, mime in _EXTRA_MIME.items():
        if path.endswith(ext):
            return mime
    mime, _ = mimetypes.guess_type(path)
    return mime if mime and mime.startswith("image/") else "image/jpeg"


def extract_items(root: ET.Element) -> list[dict]:
    """
    Parse all <url> entries in *root* into a list of dicts with keys:

    Standard sitemap
    ----------------
    loc, lastmod, changefreq, priority

    Google News (news:*)
    --------------------
    news_title, news_pub_date, news_keywords, news_genres,
    news_pub_name, news_pub_language

    Google Image (image:*)
    ----------------------
    images  →  list of {"loc", "caption", "title", "license"}
    """
    items: list[dict] = []

    for url_el in _extract_url_elements(root):
        item: dict = {
            # Standard sitemap
            "loc":        _text(url_el, f"{SM}loc",        "loc"),
            "lastmod":    _text(url_el, f"{SM}lastmod",    "lastmod"),
            "changefreq": _text(url_el, f"{SM}changefreq", "changefreq"),
            "priority":   _text(url_el, f"{SM}priority",   "priority"),
            # Google News (populated below)
            "news_title":        "",
            "news_pub_date":     "",
            "news_keywords":     "",
            "news_genres":       "",
            "news_pub_name":     "",
            "news_pub_language": "",
            # Google Image (populated below)
            "images": [],
        }

        # ---- Google News extension ----
        news_el = url_el.find(f"{N}news")
        if news_el is not None:
            pub_el = news_el.find(f"{N}publication")
            item["news_title"]        = _text(news_el, f"{N}title")
            item["news_pub_date"]     = _text(news_el, f"{N}publication_date")
            item["news_keywords"]     = _text(news_el, f"{N}keywords")
            item["news_genres"]       = _text(news_el, f"{N}genres")
            item["news_pub_name"]     = _text(pub_el,  f"{N}name")     if pub_el is not None else ""
            item["news_pub_language"] = _text(pub_el,  f"{N}language") if pub_el is not None else ""

        # ---- Google Image extension ----
        for img_el in url_el.findall(f"{IMG}image"):
            item["images"].append({
                "loc":     _text(img_el, f"{IMG}loc"),
                "caption": _text(img_el, f"{IMG}caption"),
                "title":   _text(img_el, f"{IMG}title"),
                "license": _text(img_el, f"{IMG}license"),
            })

        if item["loc"]:
            items.append(item)

    return items


def collect_items(
    source: str,
    follow_index: bool = False,
    visited: Optional[set] = None,
) -> list[dict]:
    """
    Fetch *source*, parse it, and return all URL items.

    If *follow_index* is True and the sitemap is a sitemapindex, recursively
    fetches each child sitemap and merges their items.
    """
    if visited is None:
        visited = set()
    if source in visited:
        return []
    visited.add(source)

    print(f"  Fetching: {source}", file=sys.stderr)
    root = parse_root(source)

    if _is_index(root):
        child_urls = _child_sitemap_urls(root)
        print(
            f"  → Sitemap index with {len(child_urls)} child sitemaps.",
            file=sys.stderr,
        )
        if not follow_index:
            print(
                "    (Use --follow-index to fetch child sitemaps.)",
                file=sys.stderr,
            )
            return []
        all_items: list[dict] = []
        for url in child_urls:
            all_items.extend(collect_items(url, follow_index, visited))
        return all_items

    return extract_items(root)


# ---------------------------------------------------------------------------
# Channel metadata auto-detection
# ---------------------------------------------------------------------------

def _infer_channel_meta(
    source: str,
    items: list[dict],
    title_override: str,
    link_override: str,
    desc_override: str,
    lang_override: str,
) -> tuple[str, str, str, str]:
    """Return (title, link, description, language) from overrides + heuristics."""

    title = title_override
    link  = link_override
    lang  = lang_override
    desc  = desc_override

    # Infer from Google News publication fields
    for item in items:
        if not title and item.get("news_pub_name"):
            title = item["news_pub_name"]
        if lang == "en" and item.get("news_pub_language"):
            lang = item["news_pub_language"]
        if not link and item.get("loc"):
            parsed = urlparse(item["loc"])
            link = f"{parsed.scheme}://{parsed.netloc}"
        if title and link:
            break

    # Final fallbacks
    if not title:
        parsed = urlparse(source)
        title = parsed.netloc or source
    if not link:
        parsed = urlparse(source)
        link = f"{parsed.scheme}://{parsed.netloc}" if parsed.netloc else source
    if not desc:
        desc = f"RSS feed generated from {source}"

    return title, link, desc, lang


# ---------------------------------------------------------------------------
# RSS 2.0 builder
# ---------------------------------------------------------------------------

def _sort_key(item: dict) -> datetime:
    """Return a tz-aware datetime used to sort items newest-first."""
    ds = item.get("news_pub_date") or item.get("lastmod") or ""
    return parse_date(ds) or datetime.min.replace(tzinfo=timezone.utc)


def build_rss(
    items: list[dict],
    channel_title: str,
    channel_link: str,
    channel_description: str,
    channel_language: str,
    max_items: Optional[int] = None,
    source: str = "",
) -> ET.Element:
    """
    Build and return an <rss> ElementTree element (RSS 2.0).

    Mapping summary
    ---------------
    news:title            → <title> + <description> fallback
    news:publication_date → <pubDate>
    news:publication/name → <dc:creator> + channel <title> heuristic
    news:keywords         → <category> elements
    news:genres           → <category> elements
    image:image/loc       → first: <enclosure>; all: <media:content>
    image:image/caption   → <media:description>
    image:image/title     → <media:title>
    sitemap:lastmod       → <pubDate> fallback
    sitemap:loc           → <link> + <guid isPermaLink="true">
    """
    # Register namespaces so ElementTree emits them with clean prefixes.
    # Do NOT also set xmlns:* attributes manually — that would duplicate them.
    ET.register_namespace("media", NS_MEDIA)
    ET.register_namespace("dc",    NS_DC)

    # Root element
    rss = ET.Element("rss")
    rss.set("version", "2.0")

    channel = ET.SubElement(rss, "channel")

    # ---- Channel metadata ----
    ET.SubElement(channel, "title").text        = channel_title
    ET.SubElement(channel, "link").text         = channel_link
    ET.SubElement(channel, "description").text  = channel_description
    ET.SubElement(channel, "language").text     = channel_language
    ET.SubElement(channel, "generator").text    = "sitemap_to_rss.py"
    ET.SubElement(channel, "lastBuildDate").text = to_rss_date(datetime.now(timezone.utc))
    if source:
        ET.SubElement(channel, "docs").text = "https://www.rssboard.org/rss-specification"

    # ---- Sort and optionally truncate ----
    sorted_items = sorted(items, key=_sort_key, reverse=True)
    if max_items and max_items > 0:
        sorted_items = sorted_items[:max_items]

    # ---- Items ----
    for item in sorted_items:
        entry = ET.SubElement(channel, "item")

        # Title
        title = item.get("news_title") or item.get("loc", "Untitled")
        ET.SubElement(entry, "title").text = title

        # Link
        link = item.get("loc", "")
        ET.SubElement(entry, "link").text = link

        # GUID
        guid = ET.SubElement(entry, "guid")
        guid.set("isPermaLink", "true")
        guid.text = link

        # Publication date (news date preferred; lastmod as fallback)
        ds = item.get("news_pub_date") or item.get("lastmod")
        ET.SubElement(entry, "pubDate").text = to_rss_date(parse_date(ds))

        # Description
        desc_parts: list[str] = []
        if item.get("news_keywords"):
            desc_parts.append(f"Keywords: {item['news_keywords']}")
        if item.get("news_genres"):
            desc_parts.append(f"Genres: {item['news_genres']}")
        if item.get("changefreq"):
            desc_parts.append(f"Change frequency: {item['changefreq']}")
        ET.SubElement(entry, "description").text = (
            " | ".join(desc_parts) if desc_parts else title
        )

        # dc:creator  (news publication name)
        if item.get("news_pub_name"):
            ET.SubElement(entry, f"{{{NS_DC}}}creator").text = item["news_pub_name"]

        # dc:language
        if item.get("news_pub_language"):
            ET.SubElement(entry, f"{{{NS_DC}}}language").text = item["news_pub_language"]

        # <category> — keywords
        if item.get("news_keywords"):
            for kw in re.split(r"[,;]\s*", item["news_keywords"]):
                kw = kw.strip()
                if kw:
                    ET.SubElement(entry, "category").text = kw

        # <category> — genres
        if item.get("news_genres"):
            for genre in re.split(r"[,;]\s*", item["news_genres"]):
                genre = genre.strip()
                if genre:
                    cat = ET.SubElement(entry, "category")
                    cat.set("domain", "genre")
                    cat.text = genre

        # Images
        images = item.get("images", [])
        for idx, img in enumerate(images):
            img_url = img.get("loc", "")
            if not img_url:
                continue
            mime = _image_mime(img_url)

            # First image → RSS 2.0 <enclosure>
            if idx == 0:
                enc = ET.SubElement(entry, "enclosure")
                enc.set("url",    img_url)
                enc.set("type",   mime)
                enc.set("length", "0")   # length required by spec; 0 = unknown

            # All images → <media:content> (Media RSS)
            mc = ET.SubElement(entry, f"{{{NS_MEDIA}}}content")
            mc.set("url",    img_url)
            mc.set("medium", "image")
            mc.set("type",   mime)

            if img.get("title"):
                ET.SubElement(mc, f"{{{NS_MEDIA}}}title").text = img["title"]
            if img.get("caption"):
                ET.SubElement(mc, f"{{{NS_MEDIA}}}description").text = img["caption"]
            if img.get("license"):
                ET.SubElement(mc, f"{{{NS_MEDIA}}}license").text = img["license"]

    return rss


# ---------------------------------------------------------------------------
# Output formatting
# ---------------------------------------------------------------------------

def _pretty_xml(element: ET.Element) -> str:
    """Return a pretty-printed, UTF-8 XML string with a proper declaration."""
    raw = ET.tostring(element, encoding="unicode")
    dom = minidom.parseString(raw)
    pretty = dom.toprettyxml(indent="  ", encoding=None)
    # toprettyxml adds its own <?xml?> declaration; keep it but normalise it
    # to an explicit UTF-8 declaration
    lines = pretty.splitlines()
    output_lines: list[str] = ['<?xml version="1.0" encoding="utf-8"?>']
    for line in lines:
        stripped = line.rstrip()
        # Skip the declaration emitted by toprettyxml (first line)
        if stripped.startswith("<?xml") and len(output_lines) == 1:
            continue
        # Skip blank lines introduced by toprettyxml
        if stripped:
            output_lines.append(stripped)
    return "\n".join(output_lines) + "\n"


def _compact_xml(element: ET.Element) -> str:
    """Return a compact XML string (no extra whitespace)."""
    raw = ET.tostring(element, encoding="unicode")
    return '<?xml version="1.0" encoding="utf-8"?>\n' + raw + "\n"


# ---------------------------------------------------------------------------
# CLI
# ---------------------------------------------------------------------------

def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="sitemap_to_rss.py",
        description="Convert a sitemap XML (standard / Google News / Image) to RSS 2.0.",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Examples
--------
  # Fetch a live news sitemap and print RSS to stdout
  python sitemap_to_rss.py https://www.usatoday.com/news-sitemap.xml

  # Save to a file, limit to 25 items
  python sitemap_to_rss.py https://www.bbc.com/sitemap.xml -o bbc.rss --max-items 25

  # Process a local sitemap file with custom channel metadata
  python sitemap_to_rss.py sitemap.xml -o feed.rss \\
      --title "My News Site" --link https://example.com

  # Follow a sitemap index and merge all child sitemaps
  python sitemap_to_rss.py https://example.com/sitemap-index.xml --follow-index -o all.rss
        """,
    )
    parser.add_argument(
        "source",
        help="Sitemap URL (http/https) or local file path.",
    )
    parser.add_argument(
        "-o", "--output",
        metavar="FILE",
        help="Write RSS output to FILE instead of stdout.",
    )
    parser.add_argument(
        "--title",
        default="",
        metavar="TEXT",
        help="RSS channel <title>  (auto-detected from news:publication if omitted).",
    )
    parser.add_argument(
        "--link",
        default="",
        metavar="URL",
        help="RSS channel <link>  (auto-detected from item URLs if omitted).",
    )
    parser.add_argument(
        "--description",
        default="",
        metavar="TEXT",
        help="RSS channel <description>.",
    )
    parser.add_argument(
        "--language",
        default="en",
        metavar="LANG",
        help="RSS channel <language>  (default: en; auto-detected from news:language).",
    )
    parser.add_argument(
        "--follow-index",
        action="store_true",
        help="If source is a sitemap index, fetch and merge all child sitemaps.",
    )
    parser.add_argument(
        "--max-items",
        type=int,
        default=0,
        metavar="N",
        help="Keep only the N most-recent items (default: all).",
    )
    parser.add_argument(
        "--no-pretty",
        action="store_true",
        help="Compact output (no indentation / extra whitespace).",
    )
    return parser


def main(argv: Optional[list[str]] = None) -> int:
    parser = build_parser()
    args   = parser.parse_args(argv)

    # ---- Collect items ----
    print(f"Processing sitemap: {args.source}", file=sys.stderr)
    items = collect_items(args.source, follow_index=args.follow_index)

    if not items:
        print(
            "Warning: No URL entries found. "
            "If this is a sitemap index, try --follow-index.",
            file=sys.stderr,
        )

    print(f"Extracted {len(items)} URL entries.", file=sys.stderr)

    # ---- Infer channel metadata ----
    title, link, desc, lang = _infer_channel_meta(
        source=args.source,
        items=items,
        title_override=args.title,
        link_override=args.link,
        desc_override=args.description,
        lang_override=args.language,
    )

    # ---- Build RSS ----
    rss = build_rss(
        items=items,
        channel_title=title,
        channel_link=link,
        channel_description=desc,
        channel_language=lang,
        max_items=args.max_items or None,
        source=args.source,
    )

    # ---- Serialise ----
    xml_text = _compact_xml(rss) if args.no_pretty else _pretty_xml(rss)

    if args.output:
        Path(args.output).write_text(xml_text, encoding="utf-8")
        print(f"RSS feed written to: {args.output}", file=sys.stderr)
    else:
        sys.stdout.write(xml_text)

    return 0


if __name__ == "__main__":
    sys.exit(main())