26 lines
729 B
Python
26 lines
729 B
Python
|
from urllib.parse import urljoin
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
from django.conf import settings
|
||
|
|
|
||
|
|
|
||
|
|
def notify_admins(message: str) -> None:
|
||
|
|
"""Notify admins on their own Matrix channel."""
|
||
|
|
return notify_matrix_channel(settings.MATRIX_SERVICE_REQUEST_ADMIN_ROOM, message)
|
||
|
|
|
||
|
|
|
||
|
|
def notify_matrix_channel(channel_url: str, message: str) -> None:
|
||
|
|
"""Send a message to a matrix channel."""
|
||
|
|
result = httpx.post(
|
||
|
|
urljoin(channel_url, "send/m.room.message"),
|
||
|
|
json={
|
||
|
|
"msgtype": "m.text",
|
||
|
|
"body": message,
|
||
|
|
},
|
||
|
|
params={
|
||
|
|
"access_token": settings.MATRIX_ACCESS_TOKEN,
|
||
|
|
},
|
||
|
|
)
|
||
|
|
if settings.MATRIX_ACCESS_TOKEN and not settings.DEBUG:
|
||
|
|
result.raise_for_status()
|