majuna/app/models/activity.py

import datetime
from typing import Any, Optional

import requests

from app.models import AbstractConfiguration
from app.extensions import db


class Activity(db.Model):  # type: ignore
    id = db.Column(db.Integer(), primary_key=True)
    group_id = db.Column(db.Integer(), nullable=True)
    activity_type = db.Column(db.String(20), nullable=False)
    text = db.Column(db.Text(), nullable=False)
    added = db.Column(db.DateTime(), nullable=False)

    def __init__(self, *,
                 id: Optional[int] = None,
                 group_id: Optional[int] = None,
                 activity_type: str,
                 text: str,
                 added: Optional[datetime.datetime] = None,
                 **kwargs: Any) -> None:
        if type(activity_type) != str or len(activity_type) > 20 or activity_type == "":
            raise TypeError("expected string for activity type between 1 and 20 characters")
        if type(text) != str:
            raise TypeError("expected string for text")
        super().__init__(id=id,
                         group_id=group_id,
                         activity_type=activity_type,
                         text=text,
                         added=added,
                         **kwargs)
        if self.added is None:
            self.added = datetime.datetime.utcnow()

    def notify(self) -> int:
        count = 0
        hooks = Webhook.query.filter(
            Webhook.destroyed.is_(None)
        )
        for hook in hooks:
            hook.send(self.text)
            count += 1
        return count


class Webhook(AbstractConfiguration):
    format = db.Column(db.String(20))
    url = db.Column(db.String(255))

    def send(self, text: str) -> None:
        if self.format == "telegram":
            data = {"text": text}
        else:
            # Matrix as default
            data = {"body": text}
        requests.post(self.url, json=data)