CaAlden/cereal
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Compare commits

base: CaAlden:3891e253aec767733ce69c1fa2ffa117a384021d
Branches Tags
CaAlden:main
..
compare: CaAlden:a7fdd6cbd0a6c05d567b59dc1b5b9094fe215fa5
Branches Tags
CaAlden:main

No commits in common. "3891e253aec767733ce69c1fa2ffa117a384021d" and "a7fdd6cbd0a6c05d567b59dc1b5b9094fe215fa5" have entirely different histories.
3891e253ae ... a7fdd6cbd0

9 changed files with 8 additions and 343 deletions
Download patch file Download diff file Expand all files Collapse all files

4
config.example.json
View file

@ -4,8 +4,4 @@
"logging": {
"level": "info"
}
"email": {
"api_key": "asdf1234",
"sender": "me@example.com"
}
}

14
src/config/__init__.py
View file

@ -1,9 +1,9 @@
import json
from logging import INFO
from dataclasses import dataclass
from typing import Any
from .logging import Logging
from .email import Email
from .parse import ConfigParseError, assert_key_of_type
@ -12,25 +12,21 @@ class Config:
host: str | None
port: int | None
logging: Logging
email: Email
@classmethod
def from_dict(cls, config: dict[str, Any]) -> 'Config':
assert_key_of_type(config, 'logging', dict)
assert_key_of_type(config, 'email', dict)
assert_key_of_type(config, 'host', str)
assert_key_of_type(config, 'port', int)
try:
log_config = Logging.from_dict(config['logging'])
return Config(host=config['host'], port=config['port'], logging=log_config)
except ConfigParseError as e:
raise ConfigParseError(['logging', *e.keypath], e.issue) from e
try:
email_config = Email.from_dict(config['email'])
except ConfigParseError as e:
raise ConfigParseError(['email', *e.keypath], e.issue) from e
return Config(host=config['host'], port=config['port'], email=email_config, logging=log_config)
DEFAULT_CONFIG = Config(logging=Logging(level=INFO), port=None, host=None)
def parse_config(filename: str) -> Config:
@ -39,4 +35,4 @@ def parse_config(filename: str) -> Config:
return Config.from_dict(config)
__all__ = ['Config', 'parse_config']
__all__ = ['Config', 'parse_config', 'DEFAULT_CONFIG']

18
src/config/email.py
View file

@ -1,18 +0,0 @@
from src.config.parse import assert_key_of_type
from typing import Any
from dataclasses import dataclass
EmailAddress = str
@dataclass
class Email:
api_key: str
sender: EmailAddress
@classmethod
def from_dict(cls, config: dict[str, Any]) -> 'Email':
assert_key_of_type(config, 'api_key', str)
assert_key_of_type(config, 'sender', str)
return Email(api_key=config['api_key'], sender=config['sender'])

7
src/main.py
View file

@ -5,7 +5,7 @@ from flask import Flask
import argparse
import logging
from src.config import parse_config, Config
from src.config import parse_config, DEFAULT_CONFIG, Config
logger = logging.getLogger(__name__)
@ -23,10 +23,9 @@ def create_app(name: str, config: Config) -> Flask:
def main():
parser = argparse.ArgumentParser('Cereal', description='run the Cereal server')
parser.add_argument('--config', '-c', help='The path to the configuration file, expects JSON', required=True)
parser.add_argument('--config', '-c', help='The path to the configuration file, expects JSON')
args = parser.parse_args()
config = parse_config(args.config)
config = parse_config(args.config) if args.config else DEFAULT_CONFIG
app = create_app('Cereal', config)
logger.info(str(config))
serve(app, host='0.0.0.0' if config.host is None else config.host, port=config.port)

0
src/notifications/__init__.py
View file

38
src/notifications/data.py
View file

@ -1,38 +0,0 @@
from dataclasses import dataclass, field
URL = str
@dataclass
class NotificationAction:
# A unique identifier for the action
action: str
# The title to show with the action
title: str
# Where to navigate to when clicked.
navigate: URL | None = None
@dataclass
class Notification:
"""
Details to show in a notification
Much of the data here is intended to match the arguments to the [showNotification](https://developer.mozilla.org/en-US/docs/Web/API/ServiceWorkerRegistration/showNotification)
Web API, but this content is ideally displayable in other contexts (for example in Email or RSS)
"""
# The main title content to show
title: str
# The content of the notification
content: str
# The UNIX timestamp in milliseconds since the epoch associated with this notification
timestamp: int
# A list of actions included with this notification
actions: list[NotificationAction] = field(default_factory=list)
# The location of an icon to show as the icon for this notification
icon: URL | None = None
# The location of an image associated with this notification
image: URL | None = None

53
src/notifications/email.py
View file

@ -1,53 +0,0 @@
from typing import Any
from email.utils import formatdate, make_msgid
import abc
from email.message import EmailMessage
from ..config.email import Email as EmailConfig
from .service import NotificationService
from .data import Notification
class EmailService(NotificationService):
def __init__(self, config: EmailConfig):
self._config = config
def notify(self, user: Any, notification: Notification):
email = self._prepare_email(user.email, notification)
self.send_email(email)
@abc.abstractmethod
def send_email(self, email: EmailMessage):
pass
def _render_template_text(self, notification: Notification) -> str:
# TODO
return ''
def _render_template_html(self, notification: Notification) -> str:
# TODO
return ''
def _prepare_email(self, receiver: str, notification: Notification) -> EmailMessage:
msg = EmailMessage()
msg['From'] = self._config.sender
msg['To'] = receiver
msg['Subject'] = notification.title
msg['Date'] = formatdate(localtime=True)
msg['Message-ID'] = make_msgid()
msg.set_content(self._render_template_text(notification))
msg.add_alternative(self._render_template_html(notification), subtype='html')
return msg
class BirdEmailServiceImpl(EmailService):
def send_email(self, email: EmailMessage):
# TODO: Implement Bird Specific Email sending and Error handling
pass
def get_service(config: EmailConfig) -> EmailService:
return BirdEmailServiceImpl(config)

205
src/notifications/rss.py
View file

@ -1,205 +0,0 @@
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)

12
src/notifications/service.py
View file

@ -1,12 +0,0 @@
from typing import Any
from .data import Notification
import abc
class NotificationService(abc.ABC):
"""A service that sends notifications to the user"""
@abc.abstractmethod
def notify(self, user: Any, notification: Notification):
"""Send a notification"""
pass