watchers

1#!/usr/bin/env python32"""Watch an RSS 2.0 or Atom feed; print new items to stdout, silent on empty.3 4Usage (via cron with --no-agent):5 6    hermes cron create my-feed \\7      --schedule "*/15 * * * *" --no-agent \\8      --script "$HERMES_HOME/skills/devops/watchers/scripts/watch_rss.py" \\9      --script-args "--name hn --url https://news.ycombinator.com/rss"10 11First run records a baseline (emits nothing).  Subsequent runs emit only12items whose <guid> / <id> isn't in the watermark.13"""14 15from __future__ import annotations16 17import argparse18import sys19import urllib.error20import urllib.request21from pathlib import Path22from xml.etree import ElementTree as ET23 24sys.path.insert(0, str(Path(__file__).parent))25from _watermark import Watermark, format_items_as_markdown  # type: ignore26 27 28def _strip_ns(tag: str) -> str:29    return tag.split("}", 1)[1] if "}" in tag else tag30 31 32def _parse_feed(xml_bytes: bytes):33    """Return a list of {id, title, url, summary} dicts.34 35    Handles both RSS 2.0 ``<item>`` and Atom ``<entry>``.36    """37    try:38        root = ET.fromstring(xml_bytes)39    except ET.ParseError as e:40        print(f"watch_rss: invalid XML: {e}", file=sys.stderr)41        sys.exit(2)42 43    entries = []44    for item in root.iter():45        tag = _strip_ns(item.tag)46        if tag not in {"item", "entry"}:47            continue48        # ElementTree Elements without children are *falsy* — use `is not None`.49        children = {_strip_ns(c.tag): c for c in item}50 51        guid_el = children.get("guid")52        if guid_el is None:53            guid_el = children.get("id")54        link_el = children.get("link")55        if link_el is not None:56            href = link_el.attrib.get("href") or (link_el.text or "").strip()57        else:58            href = ""59        guid = (guid_el.text or "").strip() if guid_el is not None else ""60        guid = guid or href61        if not guid:62            continue63 64        title_el = children.get("title")65        title = (title_el.text or "").strip() if title_el is not None else ""66 67        summ_el = children.get("description")68        if summ_el is None:69            summ_el = children.get("summary")70        summary = (summ_el.text or "").strip() if summ_el is not None else ""71 72        entries.append(73            {"id": guid, "title": title, "url": href, "summary": summary}74        )75    return entries76 77 78def main() -> int:79    p = argparse.ArgumentParser(description="Watch an RSS/Atom feed.")80    p.add_argument("--name", required=True, help="Watcher name (used for state file)")81    p.add_argument("--url", required=True, help="Feed URL")82    p.add_argument("--max", type=int, default=10,83                   help="Max new items to emit per tick (default: 10)")84    p.add_argument("--with-summary", action="store_true",85                   help="Include <description>/<summary> snippet under each item")86    p.add_argument("--timeout", type=float, default=20.0,87                   help="HTTP timeout in seconds (default: 20)")88    args = p.parse_args()89 90    try:91        req = urllib.request.Request(args.url, headers={"User-Agent": "Hermes-Watcher/1.0"})92        with urllib.request.urlopen(req, timeout=args.timeout) as resp:93            xml_bytes = resp.read()94    except urllib.error.HTTPError as e:95        print(f"watch_rss: HTTP {e.code} from {args.url}", file=sys.stderr)96        return 297    except (urllib.error.URLError, TimeoutError, OSError) as e:98        print(f"watch_rss: network error: {e}", file=sys.stderr)99        return 2100 101    entries = _parse_feed(xml_bytes)102 103    wm = Watermark.load(args.name)104    new_items = wm.filter_new(entries, id_key="id")105    wm.save()106 107    # Cap emitted items (watermark still records all seen IDs so we don't108    # re-emit them next tick).109    if args.max > 0:110        new_items = new_items[: args.max]111 112    body_key = "summary" if args.with_summary else None113    output = format_items_as_markdown(new_items, body_key=body_key)114    if output:115        sys.stdout.write(output)116    # Empty stdout on no-new — cron treats that as silent.117    return 0118 119 120if __name__ == "__main__":121    sys.exit(main())122