80 lines
2.1 KiB
Python
80 lines
2.1 KiB
Python
from datetime import datetime, timezone
|
|
from typing import Any, Optional
|
|
|
|
import requests
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from app.brm.brn import BRN
|
|
from app.extensions import db
|
|
from app.models import AbstractConfiguration
|
|
from app.models.types import AwareDateTime
|
|
|
|
|
|
class Activity(db.Model): # type: ignore
|
|
id: Mapped[int] = mapped_column(primary_key=True)
|
|
group_id: Mapped[Optional[int]]
|
|
activity_type: Mapped[str]
|
|
text: Mapped[str]
|
|
added: Mapped[datetime] = mapped_column(AwareDateTime())
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
id: Optional[int] = None,
|
|
group_id: Optional[int] = None,
|
|
activity_type: str,
|
|
text: str,
|
|
added: Optional[datetime] = None,
|
|
**kwargs: Any
|
|
) -> None:
|
|
if (
|
|
not isinstance(activity_type, str)
|
|
or len(activity_type) > 20
|
|
or activity_type == ""
|
|
):
|
|
raise TypeError(
|
|
"expected string for activity type between 1 and 20 characters"
|
|
)
|
|
if not isinstance(text, str):
|
|
raise TypeError("expected string for text")
|
|
if added is None:
|
|
added = datetime.now(tz=timezone.utc)
|
|
super().__init__(
|
|
id=id,
|
|
group_id=group_id,
|
|
activity_type=activity_type,
|
|
text=text,
|
|
added=added,
|
|
**kwargs
|
|
)
|
|
|
|
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: Mapped[str]
|
|
url: Mapped[str]
|
|
|
|
@property
|
|
def brn(self) -> BRN:
|
|
return BRN(
|
|
group_id=0,
|
|
product="notify",
|
|
provider=self.format,
|
|
resource_type="conf",
|
|
resource_id=str(self.id),
|
|
)
|
|
|
|
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, timeout=30)
|