rss-feeds

1#!/usr/bin/env python32"""Read RSS / Atom / JSON Feed sources and discover feeds behind a page URL.3 4Standard library only, so it runs in any Hermes environment without an install5step. Output is JSON (``--json``) or a compact text listing.6 7    python3 feed.py read https://example.com/feed.xml [--limit N] [--since 2026-09-01]8    python3 feed.py discover https://example.com/9    python3 feed.py read https://example.com/  ->  discovers, then reads the first feed10"""11 12from __future__ import annotations13 14import argparse15import html16import json17import re18import sys19import urllib.error20import urllib.parse21import urllib.request22import xml.etree.ElementTree as ET23from datetime import datetime, timezone24from email.utils import parsedate_to_datetime25 26USER_AGENT = "hermes-agent/1.0 (rss-feeds skill; +https://github.com/NousResearch/hermes-agent)"27TIMEOUT = 2028NS = {29    "atom": "http://www.w3.org/2005/Atom",30    "dc": "http://purl.org/dc/elements/1.1/",31    "content": "http://purl.org/rss/1.0/modules/content/",32    "media": "http://search.yahoo.com/mrss/",33}34FEED_TYPES = ("application/rss+xml", "application/atom+xml", "application/feed+json", "application/json")35COMMON_FEED_PATHS = ("/feed", "/feed.xml", "/rss", "/rss.xml", "/atom.xml", "/index.xml", "/feed.json", "/blog/feed", "/blog/rss.xml")36_TAG_RE = re.compile(r"<[^>]+>")37_WS_RE = re.compile(r"\s+")38 39 40def fetch(url: str) -> tuple[bytes, str]:41    req = urllib.request.Request(url, headers={"User-Agent": USER_AGENT, "Accept": "*/*"})42    with urllib.request.urlopen(req, timeout=TIMEOUT) as resp:43        return resp.read(), resp.headers.get("Content-Type", "")44 45 46def strip_html(text: str | None) -> str:47    if not text:48        return ""49    return _WS_RE.sub(" ", html.unescape(_TAG_RE.sub(" ", text))).strip()50 51 52def parse_date(value: str | None) -> str | None:53    """Normalise RFC 822 (RSS) and ISO 8601 (Atom/JSON Feed) dates to UTC ISO."""54    if not value:55        return None56    value = value.strip()57    try:58        dt = parsedate_to_datetime(value)59    except (TypeError, ValueError):60        try:61            dt = datetime.fromisoformat(value.replace("Z", "+00:00"))62        except ValueError:63            return value64    if dt.tzinfo is None:65        dt = dt.replace(tzinfo=timezone.utc)66    return dt.astimezone(timezone.utc).isoformat()67 68 69def _text(el, *paths) -> str | None:70    for p in paths:71        found = el.find(p, NS)72        if found is not None and (found.text or "").strip():73            return found.text74    return None75 76 77def _atom_link(entry) -> str | None:78    alternate = None79    for link in entry.findall("atom:link", NS) + entry.findall("link"):80        href = link.get("href")81        if not href:82            continue83        rel = link.get("rel", "alternate")84        if rel == "alternate":85            return href86        alternate = alternate or href87    return alternate88 89 90def parse_xml(data: bytes) -> dict:91    root = ET.fromstring(data)92    tag = root.tag.rsplit("}", 1)[-1].lower()93    if tag == "feed":  # Atom94        title = strip_html(_text(root, "atom:title"))95        entries = []96        for e in root.findall("atom:entry", NS):97            entries.append({98                "title": strip_html(_text(e, "atom:title")),99                "link": _atom_link(e),100                "published": parse_date(_text(e, "atom:published", "atom:updated")),101                "author": strip_html(_text(e, "atom:author/atom:name", "dc:creator")),102                "summary": strip_html(_text(e, "atom:summary", "atom:content"))[:2000],103            })104        return {"format": "atom", "title": title, "entries": entries}105    channel = root.find("channel") if tag == "rss" else root  # RSS 2.0 vs RDF/RSS 1.0106    if channel is None:107        raise ValueError(f"unrecognised XML root <{tag}>")108    entries = []109    for item in channel.iter("item") if tag == "rss" else root.iter("{http://purl.org/rss/1.0/}item"):110        entries.append({111            "title": strip_html(_text(item, "title", "{http://purl.org/rss/1.0/}title")),112            "link": (_text(item, "link", "{http://purl.org/rss/1.0/}link") or "").strip() or None,113            "published": parse_date(_text(item, "pubDate", "dc:date")),114            "author": strip_html(_text(item, "dc:creator", "author")),115            "summary": strip_html(_text(item, "content:encoded", "description", "{http://purl.org/rss/1.0/}description"))[:2000],116        })117    return {"format": "rss", "title": strip_html(_text(channel, "title", "{http://purl.org/rss/1.0/}title")), "entries": entries}118 119 120def parse_json_feed(data: bytes) -> dict:121    doc = json.loads(data)122    entries = []123    for item in doc.get("items", []):124        authors = item.get("authors") or ([item["author"]] if item.get("author") else [])125        entries.append({126            "title": strip_html(item.get("title")),127            "link": item.get("url") or item.get("external_url"),128            "published": parse_date(item.get("date_published") or item.get("date_modified")),129            "author": ", ".join(a.get("name", "") for a in authors if isinstance(a, dict)) or None,130            "summary": strip_html(item.get("summary") or item.get("content_text") or item.get("content_html"))[:2000],131        })132    return {"format": "jsonfeed", "title": strip_html(doc.get("title")), "entries": entries}133 134 135def parse_feed(data: bytes, content_type: str = "") -> dict:136    head = data.lstrip()[:1]137    if head == b"{" or "json" in content_type:138        return parse_json_feed(data)139    return parse_xml(data)140 141 142def discover(page_url: str, page_html: bytes | None = None) -> list[str]:143    """Return candidate feed URLs for a page: <link rel=alternate> first, then well-known paths."""144    if page_html is None:145        page_html, _ = fetch(page_url)146    text = page_html.decode("utf-8", "replace")147    found: list[str] = []148    for m in re.finditer(r"<link\b[^>]*>", text, re.I):149        tag = m.group(0)150        type_m = re.search(r"""type\s*=\s*["']([^"']+)""", tag, re.I)151        href_m = re.search(r"""href\s*=\s*["']([^"']+)""", tag, re.I)152        rel_m = re.search(r"""rel\s*=\s*["']([^"']+)""", tag, re.I)153        if not href_m or not type_m or type_m.group(1).lower() not in FEED_TYPES:154            continue155        if rel_m and "alternate" not in rel_m.group(1).lower():156            continue157        url = urllib.parse.urljoin(page_url, html.unescape(href_m.group(1)))158        if url not in found:159            found.append(url)160    if found:161        return found162    parsed = urllib.parse.urlsplit(page_url)163    base = f"{parsed.scheme}://{parsed.netloc}"164    return [base + p for p in COMMON_FEED_PATHS]165 166 167def looks_like_feed(data: bytes, content_type: str) -> bool:168    head = data.lstrip()[:300].lower()169    return head.startswith(b"{") and b"items" in data[:2000] or b"<rss" in head or b"<feed" in head or b"<rdf" in head170 171 172def read(url: str) -> dict:173    data, ctype = fetch(url)174    if looks_like_feed(data, ctype):175        feed = parse_feed(data, ctype)176        feed["url"] = url177        return feed178    candidates = discover(url, data)179    errors = []180    for cand in candidates:181        try:182            cdata, cctype = fetch(cand)183        except (urllib.error.URLError, OSError) as exc:184            errors.append(f"{cand}: {exc}")185            continue186        if looks_like_feed(cdata, cctype):187            feed = parse_feed(cdata, cctype)188            feed["url"] = cand189            feed["discovered_from"] = url190            return feed191    raise SystemExit(f"no feed found at {url}; tried {len(candidates)} candidates\n" + "\n".join(errors))192 193 194def filter_entries(entries: list[dict], limit: int, since: str | None) -> list[dict]:195    if since:196        cutoff = datetime.fromisoformat(since).replace(tzinfo=timezone.utc) if "T" not in since else datetime.fromisoformat(since.replace("Z", "+00:00"))197        if cutoff.tzinfo is None:198            cutoff = cutoff.replace(tzinfo=timezone.utc)199        entries = [e for e in entries if e["published"] and datetime.fromisoformat(e["published"]) >= cutoff]200    entries.sort(key=lambda e: e["published"] or "", reverse=True)201    return entries[:limit]202 203 204def render_text(feed: dict) -> str:205    lines = [f"{feed.get('title') or '(untitled feed)'}  [{feed['format']}]  {feed['url']}"]206    for e in feed["entries"]:207        when = (e["published"] or "")[:10]208        by = f"  — {e['author']}" if e.get("author") else ""209        lines.append(f"- {when}  {e['title'] or '(no title)'}{by}\n  {e['link'] or ''}")210        if e.get("summary"):211            lines.append(f"  {e['summary'][:300]}")212    return "\n".join(lines)213 214 215def main(argv: list[str] | None = None) -> int:216    ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)217    sub = ap.add_subparsers(dest="cmd", required=True)218    r = sub.add_parser("read", help="read a feed (or discover one behind a page URL)")219    r.add_argument("url")220    r.add_argument("--limit", type=int, default=20)221    r.add_argument("--since", help="ISO date/datetime; drop older entries")222    r.add_argument("--json", action="store_true")223    d = sub.add_parser("discover", help="list feed URLs advertised by a page")224    d.add_argument("url")225    d.add_argument("--json", action="store_true")226    args = ap.parse_args(argv)227 228    try:229        if args.cmd == "discover":230            urls = discover(args.url)231            print(json.dumps(urls, indent=2) if args.json else "\n".join(urls))232            return 0233        feed = read(args.url)234        feed["entries"] = filter_entries(feed["entries"], args.limit, args.since)235        print(json.dumps(feed, indent=2, ensure_ascii=False) if args.json else render_text(feed))236        return 0237    except urllib.error.HTTPError as exc:238        print(f"HTTP {exc.code} for {exc.url}", file=sys.stderr)239        return 2240    except (urllib.error.URLError, ET.ParseError, ValueError, json.JSONDecodeError) as exc:241        print(f"error: {exc}", file=sys.stderr)242        return 2243 244 245if __name__ == "__main__":246    sys.exit(main())247