diff --git a/src/notifications/rss.py b/src/notifications/rss.py new file mode 100644 index 0000000..6421a30 --- /dev/null +++ b/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)