| 1 | """Deterministic Atom rendering for the saved research library.""" |
| 2 | |
| 3 | from __future__ import annotations |
| 4 | |
| 5 | from collections.abc import Mapping, Sequence |
| 6 | from datetime import datetime |
| 7 | from xml.etree import ElementTree as ET |
| 8 | |
| 9 | from .library import LibraryEntry |
| 10 | |
| 11 | |
| 12 | ATOM_NS = "http://www.w3.org/2005/Atom" |
| 13 | ET.register_namespace("", ATOM_NS) |
| 14 | |
| 15 | |
| 16 | def render_atom( |
| 17 | entries: Sequence[LibraryEntry], |
| 18 | *, |
| 19 | library_id: str, |
| 20 | entry_urls: Mapping[str, str] | None = None, |
| 21 | feed_url: str | None = None, |
| 22 | title: str = "last30days research library", |
| 23 | author: str = "last30days research library", |
| 24 | ) -> str: |
| 25 | """Render an Atom feed whose IDs and timestamps are stable across runs.""" |
| 26 | urls = entry_urls or {} |
| 27 | feed_id = f"urn:last30days:research-library:{library_id}" |
| 28 | root = ET.Element(_tag("feed")) |
| 29 | ET.SubElement(root, _tag("id")).text = feed_id |
| 30 | ET.SubElement(root, _tag("title")).text = title |
| 31 | author_node = ET.SubElement(root, _tag("author")) |
| 32 | ET.SubElement(author_node, _tag("name")).text = author |
| 33 | updated = max((item.source_updated_at for item in entries), default=None) |
| 34 | ET.SubElement(root, _tag("updated")).text = ( |
| 35 | _format_timestamp(updated) if updated else "1970-01-01T00:00:00Z" |
| 36 | ) |
| 37 | if feed_url: |
| 38 | ET.SubElement(root, _tag("link"), {"rel": "self", "href": feed_url}) |
| 39 | |
| 40 | for item in entries: |
| 41 | node = ET.SubElement(root, _tag("entry")) |
| 42 | entry_id = item.entry_id.removeprefix("urn:last30days:") |
| 43 | ET.SubElement(node, _tag("id")).text = f"{feed_id}:{entry_id}" |
| 44 | ET.SubElement(node, _tag("title")).text = item.headline |
| 45 | ET.SubElement(node, _tag("updated")).text = _format_timestamp(item.source_updated_at) |
| 46 | ET.SubElement(node, _tag("published")).text = f"{item.published_date.isoformat()}T00:00:00Z" |
| 47 | ET.SubElement(node, _tag("category"), {"term": item.topic}) |
| 48 | url = urls.get(item.entry_id, f"briefs/{item.output_name}") |
| 49 | ET.SubElement(node, _tag("link"), {"href": url}) |
| 50 | ET.SubElement(node, _tag("summary"), {"type": "text"}).text = item.summary |
| 51 | |
| 52 | ET.indent(root, space=" ") |
| 53 | return '<?xml version="1.0" encoding="utf-8"?>\n' + ET.tostring(root, encoding="unicode") + "\n" |
| 54 | |
| 55 | |
| 56 | def _tag(name: str) -> str: |
| 57 | return f"{{{ATOM_NS}}}{name}" |
| 58 | |
| 59 | |
| 60 | def _format_timestamp(value: datetime) -> str: |
| 61 | return value.isoformat().replace("+00:00", "Z") |
| 62 |