feat(organizations): add optional Telegram notifications for detections

.env.example

@@ -21,6 +21,7 @@ SERVER_NAME=
 POSTHOG_HOST='https://eu.posthog.com'
 POSTHOG_KEY=
 SUPPORT_EMAIL=
+TELEGRAM_TOKEN=
 
 # Production-only
 ACME_EMAIL=

CONTRIBUTING.md

@@ -145,7 +145,7 @@ None :)
 - `S3_REGION`: your S3 bucket is geographically identified by its location's region
 - `S3_ENDPOINT_URL`: the URL providing a S3 endpoint by your cloud provider
 - `S3_PROXY_URL`: the url of the proxy to hide the real s3 url behind, do not use proxy if ""
-
+- `TELEGRAM_TOKEN`: the token of your Telegram bot
 #### Production-only values
 - `ACME_EMAIL`: the email linked to your certificate for HTTPS
 - `BACKEND_HOST`: the subdomain where your users will access your API (e.g "api.mydomain.com")

pyproject.toml

@@ -27,6 +27,7 @@ prometheus-fastapi-instrumentator = "^6.1.0"
 python-multipart = "==0.0.7"
 python-magic = "^0.4.17"
 boto3 = "^1.26.0"
+httpx = "^0.24.0"
 
 [tool.poetry.group.quality]
 optional = true

scripts/dbdiagram.txt

@@ -8,8 +8,8 @@ Table "User" as U {
   "id" int [not null]
   "organization_id" int [ref: > O.id, not null]
   "role" userrole [not null]
-  "login" text [not null]
-  "hashed_password" text [not null]
+  "login" varchar [not null]
+  "hashed_password" varchar [not null]
   "created_at" timestamp [not null]
   Indexes {
     (id, login) [pk]
@@ -19,15 +19,15 @@ Table "User" as U {
 Table "Camera" as C {
   "id" int [not null]
   "organization_id" int [ref: > O.id, not null]
-  "name" text [not null]
+  "name" varchar [not null]
   "angle_of_view" float [not null]
   "elevation" float [not null]
   "lat" float [not null]
   "lon" float [not null]
   "is_trustable" bool [not null]
   "created_at" timestamp [not null]
   "last_active_at" timestamp
-  "last_image" text
+  "last_image" varchar
   Indexes {
     (id) [pk]
   }
@@ -37,8 +37,8 @@ Table "Detection" as D {
   "id" int [not null]
   "camera_id" int [ref: > C.id, not null]
   "azimuth" float [not null]
-  "bucket_key" text [not null]
-  "bboxes" text [not null]
+  "bucket_key" varchar [not null]
+  "bboxes" varchar [not null]
   "is_wildfire" bool
   "created_at" timestamp [not null]
   "updated_at" timestamp [not null]
@@ -49,7 +49,8 @@ Table "Detection" as D {
 
 Table "Organization" as O {
   "id" int [not null]
-  "name" text [not null]
+  "name" varchar [not null]
+  "telegram_id" varchar
   Indexes {
     (id) [pk]
   }
@@ -58,7 +59,7 @@ Table "Organization" as O {
 
 Table "Webhook" as W {
   "id" int [not null]
-  "url" text [not null]
+  "url" varchar [not null]
   Indexes {
     (id) [pk]
   }

src/app/api/api_v1/endpoints/detections.py

@@ -20,10 +20,17 @@
     status,
 )
 
-from app.api.dependencies import dispatch_webhook, get_camera_crud, get_detection_crud, get_jwt, get_webhook_crud
+from app.api.dependencies import (
+    dispatch_webhook,
+    get_camera_crud,
+    get_detection_crud,
+    get_jwt,
+    get_organization_crud,
+    get_webhook_crud,
+)
 from app.core.config import settings
-from app.crud import CameraCRUD, DetectionCRUD, WebhookCRUD
-from app.models import Camera, Detection, Role, UserRole
+from app.crud import CameraCRUD, DetectionCRUD, OrganizationCRUD, WebhookCRUD
+from app.models import Camera, Detection, Organization, Role, UserRole
 from app.schemas.detections import (
     BOXES_PATTERN,
     COMPILED_BOXES_PATTERN,
@@ -34,6 +41,7 @@
 )
 from app.schemas.login import TokenPayload
 from app.services.storage import s3_service, upload_file
+from app.services.telegram import telegram_client
 from app.services.telemetry import telemetry_client
 
 router = APIRouter()
@@ -53,6 +61,7 @@
     file: UploadFile = File(..., alias="file"),
     detections: DetectionCRUD = Depends(get_detection_crud),
     webhooks: WebhookCRUD = Depends(get_webhook_crud),
+    organizations: OrganizationCRUD = Depends(get_organization_crud),
     token_payload: TokenPayload = Security(get_jwt, scopes=[Role.CAMERA]),
 ) -> Detection:
     telemetry_client.capture(f"camera|{token_payload.sub}", event="detections-create")
@@ -74,6 +83,11 @@
     if any(whs):
         for webhook in await webhooks.fetch_all():
             background_tasks.add_task(dispatch_webhook, webhook.url, det)
+    # Telegram notifications
+    if telegram_client.is_enabled:
+        org = cast(Organization, await organizations.get(token_payload.organization_id, strict=True))
+        if org.telegram_id:
+            background_tasks.add_task(telegram_client.notify, org.telegram_id, det.model_dump_json())
     return det
 
 

src/app/api/api_v1/endpoints/organizations.py

@@ -12,8 +12,9 @@
 from app.crud import OrganizationCRUD
 from app.models import Organization, UserRole
 from app.schemas.login import TokenPayload
-from app.schemas.organizations import OrganizationCreate
+from app.schemas.organizations import OrganizationCreate, TelegramChannelId
 from app.services.storage import s3_service
+from app.services.telegram import telegram_client
 from app.services.telemetry import telemetry_client
 
 router = APIRouter()
@@ -73,3 +74,21 @@
     if not (await s3_service.delete_bucket(bucket_name)):
         raise HTTPException(status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Failed to create bucket")
     await organizations.delete(organization_id)
+
+
+@router.patch(
+    "/{organization_id}", status_code=status.HTTP_200_OK, summary="Update telegram channel ID of an organization"
+)
+async def update_telegram_id(
+    payload: TelegramChannelId,
+    organization_id: int = Path(..., gt=0),
+    organizations: OrganizationCRUD = Depends(get_organization_crud),
+    token_payload: TokenPayload = Security(get_jwt, scopes=[UserRole.ADMIN]),
+) -> Organization:
+    telemetry_client.capture(
+        token_payload.sub, event="organizations-update-telegram-id", properties={"organization_id": organization_id}
+    )
+    # Check if the telegram channel ID is valid
+    if payload.telegram_id and not telegram_client.has_channel_access(payload.telegram_id):
+        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Unable to access Telegram channel")
+    return await organizations.update(organization_id, payload)

src/app/core/config.py

@@ -58,6 +58,9 @@ def sqlachmey_uri(cls, v: str) -> str:
     S3_PROXY_URL: str = os.environ.get("S3_PROXY_URL", "")
     S3_URL_EXPIRATION: int = int(os.environ.get("S3_URL_EXPIRATION") or 24 * 3600)
 
+    # Notifications
+    TELEGRAM_TOKEN: Union[str, None] = os.environ.get("TELEGRAM_TOKEN")
+
     # Error monitoring
     SENTRY_DSN: Union[str, None] = os.environ.get("SENTRY_DSN")
     SERVER_NAME: str = os.environ.get("SERVER_NAME", socket.gethostname())

src/app/crud/crud_organization.py

@@ -7,11 +7,11 @@
 
 from app.crud.base import BaseCRUD
 from app.models import Organization
-from app.schemas.organizations import OrganizationCreate, OrganizationUpdate
+from app.schemas.organizations import OrganizationCreate, TelegramChannelId
 
 __all__ = ["OrganizationCRUD"]
 
 
-class OrganizationCRUD(BaseCRUD[Organization, OrganizationCreate, OrganizationUpdate]):
+class OrganizationCRUD(BaseCRUD[Organization, OrganizationCreate, TelegramChannelId]):
     def __init__(self, session: AsyncSession) -> None:
         super().__init__(session, Organization)

src/app/models.py

@@ -69,6 +69,7 @@ class Organization(SQLModel, table=True):
     __tablename__ = "organizations"
     id: int = Field(None, primary_key=True)
     name: str = Field(..., min_length=5, max_length=100, nullable=False, unique=True)
+    telegram_id: Union[str, None] = Field(None, nullable=True)
 
 
 class Webhook(SQLModel, table=True):

src/app/schemas/organizations.py

@@ -3,6 +3,8 @@
 # This program is licensed under the Apache License 2.0.
 # See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.
 
+from typing import Union
+
 from pydantic import BaseModel, Field
 
 __all__ = ["OrganizationCreate", "OrganizationUpdate"]
@@ -26,3 +28,7 @@ class OrganizationUpdate(BaseModel):
         description="name of the organization",
         json_schema_extra={"examples": ["pyro-org-01"]},
     )
+
+
+class TelegramChannelId(BaseModel):
+    telegram_id: Union[str, None] = Field(None, pattern=r"^@[a-zA-Z0-9_-]+$")

src/app/services/telegram.py

@@ -0,0 +1,58 @@
+# Copyright (C) 2024, Pyronear.
+
+# This program is licensed under the Apache License 2.0.
+# See LICENSE or go to <https://opensource.org/licenses/Apache-2.0> for full license details.
+
+import logging
+from typing import Union
+
+import requests
+
+from app.core.config import settings
+
+logger = logging.getLogger("uvicorn.error")
+
+__all__ = ["telegram_client"]
+
+
+class TelegramClient:
+    BASE_URL = "https://api.telegram.org/bot{token}"
+
+    def __init__(self, token: Union[str, None] = None) -> None:
+        self.is_enabled = isinstance(token, str)
+        if isinstance(token, str):
+            self.token = token
+            # Validate token
+            response = requests.get(
+                f"{self.BASE_URL.format(token=self.token)}/getMe",
+                timeout=1,
+            )
+            if response.status_code != 200:
+                raise ValueError(f"Invalid Telegram Bot token: {response.text}")
+            logger.info("Telegram notifications enabled")
+
+    def has_channel_access(self, channel_id: str) -> bool:
+        if not self.is_enabled:
+            raise AssertionError("Telegram notifications are not enabled")
+        response = requests.get(
+            f"{self.BASE_URL.format(token=self.token)}/getChat",
+            json={"chat_id": channel_id},
+            timeout=1,
+        )
+        return response.status_code == 200
+
+    def notify(self, channel_id: str, message: str) -> requests.Response:
+        if not self.is_enabled:
+            raise AssertionError("Telegram notifications are not enabled")
+        response = requests.post(
+            f"{self.BASE_URL.format(token=self.token)}/sendMessage",
+            json={"chat_id": channel_id, "text": message},
+            timeout=2,
+        )
+        if response.status_code != 200:
+            logger.error(f"Failed to send message to Telegram: {response.text}")
+
+        return response
+
+
+telegram_client = TelegramClient(token=settings.TELEGRAM_TOKEN)

src/tests/conftest.py

@@ -53,10 +53,12 @@
     {
         "id": 1,
         "name": "organization-1",
+        "telegram_id": None,
     },
     {
         "id": 2,
         "name": "organization-2",
+        "telegram_id": None,
     },
 ]
 

src/tests/endpoints/test_organizations.py

@@ -55,9 +55,7 @@ async def test_create_organization(
     if isinstance(status_detail, str):
         assert response.json()["detail"] == status_detail
     if response.status_code // 100 == 2:
-        assert {
-            k: v for k, v in response.json().items() if k not in {"id", "created_at", "last_active_at", "is_trustable"}
-        } == payload
+        assert {k: v for k, v in response.json().items() if k not in {"id", "telegram_id"}} == payload
 
 
 @pytest.mark.parametrize(

src/tests/services/test_telegram.py

@@ -0,0 +1,24 @@
+import pytest
+
+from app.core.config import settings
+from app.services.telegram import TelegramClient
+
+
+def test_telegram_client():
+    with pytest.raises(ValueError, match="Invalid Telegram Bot token"):
+        TelegramClient("invalid-token")
+
+    client = TelegramClient(None)
+    assert not client.is_enabled
+
+    client = TelegramClient(settings.TELEGRAM_TOKEN)
+    assert client.is_enabled == isinstance(settings.TELEGRAM_TOKEN, str)
+
+    if isinstance(settings.TELEGRAM_TOKEN, str):
+        assert not client.has_channel_access("invalid-channel-id")
+        assert client.notify("invalid-channel-id", "test").status_code == 404
+    else:
+        with pytest.raises(AssertionError, match="Telegram notifications are not enabled"):
+            client.has_channel_access("invalid-channel-id")
+        with pytest.raises(AssertionError, match="Telegram notifications are not enabled"):
+            client.notify("invalid-channel-id", "test")