46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
import datetime
|
|
|
|
import requests
|
|
|
|
from app.models import AbstractConfiguration
|
|
from app.extensions import db
|
|
|
|
|
|
class Activity(db.Model):
|
|
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, **kwargs):
|
|
if type(kwargs["activity_type"]) != str or len(kwargs["activity_type"]) > 20 or kwargs["activity_type"] == "":
|
|
raise TypeError("expected string for activity type between 1 and 20 characters")
|
|
if type(kwargs["text"]) != str:
|
|
raise TypeError("expected string for text")
|
|
if "added" not in kwargs:
|
|
kwargs["added"] = datetime.datetime.utcnow()
|
|
super().__init__(**kwargs)
|
|
|
|
def notify(self) -> int:
|
|
count = 0
|
|
hooks = Webhook.query.filter(
|
|
Webhook.destroyed == 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):
|
|
if self.format == "telegram":
|
|
data = {"text": text}
|
|
else:
|
|
# Matrix as default
|
|
data = {"body": text}
|
|
r = requests.post(self.url, json=data)
|