majuna/app/models/activity.py

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)