Steal RSS abstractions from one of my other projects

src/notifications/rss.py

@@ -0,0 +1,205 @@
+from xml.etree.ElementTree import TreeBuilder, tostring as to_xml_string
+from typing import Any
+from email import utils
+from datetime import datetime
+from time import mktime
+from dataclasses import dataclass, field
+
+Email = str
+Language = str
+URL = str
+
+
+@dataclass
+class Image:
+    """https://www.rssboard.org/rss-specification#ltimagegtSubelementOfLtchannelgt"""
+
+    url: URL  # The URL of the image
+    title: str  # The channel title
+    link: URL  # The channel link
+    width: int | None = None  # in pixels max 144, default: 88.
+    height: int | None = None  # in pixels max 400, default: 31.
+    description: str | None = None
+
+
+@dataclass
+class Enclosure:
+    """https://www.rssboard.org/rss-specification#ltenclosuregtSubelementOfLtitemgt"""
+
+    url: URL
+    type: str  # MIME Type
+    length: int | None = None  # no. of bytes
+
+
+@dataclass
+class Item:
+    # One of Title or Description must be set
+    title: str | None = None
+    description: str | None = None
+    link: URL | None = None
+    author: Email | None = None
+    categories: list[str] = field(default_factory=list)
+    comments: URL | None = None
+    enclosures: list[Enclosure] = field(default_factory=list)
+    guid: str | None = None
+    pubDate: datetime | None = None
+
+
+@dataclass
+class Channel:
+    """https://www.rssboard.org/rss-specification"""
+
+    title: str
+    link: str
+    description: str
+    categories: list[str] = field(default_factory=list)
+    items: list[Item] = field(default_factory=list)
+    language: Language | None = None
+    copyright: str | None = None
+    managingEditor: Email | None = None
+    webMaster: Email | None = None
+    pubDate: datetime | None = None
+    lastBuildDate: datetime | None = None
+    generator: str | None = None  # A string indicating the program used to generate the channel.
+    docs: URL | None = None
+    ttl: int | None = None  # Time to cache in minutes
+    image: Image | None = None
+
+
+def convert_date(item: Any, key: str) -> datetime | None:
+    if key not in item:
+        return None
+
+    return datetime.fromtimestamp(mktime(item[key]))
+
+
+def maybe_convert_enclosure(enclosure) -> Enclosure | None:
+    href = enclosure.get('href')
+    mime = enclosure.get('type')
+    length = enclosure.get('length')
+    if any([x is None for x in [href, mime]]):
+        return None
+
+    kwargs = {
+        'url': href,
+        'type': mime,
+    }
+
+    if length is not None:
+        kwargs['length'] = int(length)
+
+    return Enclosure(**kwargs)
+
+
+def convert_item(item) -> Item:
+    tags = item['tags'] if 'tags' in item else []
+    enclosures: list[Any] = list(filter(lambda x: x is not None, [maybe_convert_enclosure(e) for e in item.enclosures]))
+
+    return Item(
+        title=item.get('title'),
+        description=item.get('summary'),
+        link=item.get('link'),
+        author=item.get('author'),
+        categories=[tag['term'] for tag in tags],
+        comments=item.get('comments'),
+        guid=item.get('id'),
+        pubDate=convert_date(item, 'published_parsed'),
+        enclosures=enclosures,
+    )
+
+
+def encode_inline(builder: TreeBuilder, tag: str, data: str | None, **kwargs):
+    if data is None:
+        return
+    builder.start(tag, kwargs)
+    builder.data(data)
+    builder.end(tag)
+
+
+def encode_date(date: datetime | None) -> str | None:
+    if date is None:
+        return
+
+    return utils.format_datetime(date)
+
+
+def encode_enclosure(builder: TreeBuilder, enclosure: Enclosure):
+    kwargs = {
+        'url': enclosure.url,
+        'type': enclosure.type,
+    }
+    if enclosure.length is not None:
+        kwargs['length'] = str(enclosure.length)
+
+    encode_inline(builder, 'enclosure', '', **kwargs)
+
+
+def encode_image(builder: TreeBuilder, image: Image):
+    builder.start('image', {})
+    encode_inline(builder, 'url', image.url)
+    encode_inline(builder, 'title', image.title)
+    encode_inline(builder, 'link', image.link)
+    encode_inline(builder, 'width', str(image.width) if image.width is not None else None)
+    encode_inline(builder, 'height', str(image.height) if image.height is not None else None)
+    encode_inline(builder, 'description', image.description)
+    builder.end('image')
+
+
+def encode_item(builder: TreeBuilder, item: Item):
+    builder.start('item', {})
+    encode_inline(builder, 'title', item.title)
+    encode_inline(builder, 'description', item.description)
+    encode_inline(builder, 'link', item.link)
+    encode_inline(builder, 'author', item.author)
+    encode_inline(builder, 'comments', item.comments)
+    encode_inline(builder, 'guid', item.guid)
+    encode_inline(builder, 'pubDate', encode_date(item.pubDate))
+
+    for cat in item.categories:
+        encode_inline(builder, 'category', cat)
+
+    for e in item.enclosures:
+        encode_enclosure(builder, e)
+
+    builder.end('item')
+
+
+def encode(channel: Channel) -> bytes:
+    """Given a Channel, emit RSS"""
+    # https://docs.python.org/3/library/xml.etree.elementtree.html#treebuilder-objects
+
+    builder = TreeBuilder()
+    headers = {
+        'version': '2.0',
+    }
+
+    builder.start('rss', headers)
+    builder.start('channel', {})
+
+    encode_inline(builder, 'title', channel.title)
+    encode_inline(builder, 'link', channel.link)
+    encode_inline(builder, 'language', channel.language)
+    encode_inline(builder, 'description', channel.description)
+    encode_inline(builder, 'copyright', channel.copyright)
+    encode_inline(builder, 'managingEditor', channel.managingEditor)
+    encode_inline(builder, 'pubDate', encode_date(channel.pubDate))
+    encode_inline(builder, 'lastBuildDate', encode_date(channel.lastBuildDate))
+    encode_inline(builder, 'webMaster', channel.webMaster)
+    encode_inline(builder, 'generator', channel.generator)
+    encode_inline(builder, 'docs', channel.docs)
+    encode_inline(builder, 'ttl', str(channel.ttl) if channel.ttl is not None else None)
+
+    if channel.image:
+        encode_image(builder, channel.image)
+
+    for cat in channel.categories:
+        encode_inline(builder, 'category', cat)
+
+    for i in channel.items:
+        encode_item(builder, i)
+
+    builder.end('channel')
+    builder.end('rss')
+
+    rss = builder.close()
+    return to_xml_string(rss)