mirrors/lego-monitoring
1
0
Fork 0
mirror of https://forgejo.altau.su/lego/lego-monitoring.git synced 2026-03-10 04:41:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

creds file -> config file

Browse source
This commit is contained in:
Alex 2024-10-27 14:47:02 +03:00
parent c653055134
commit f199292aba
5 changed files with 70 additions and 49 deletions
Download patch file Download diff file Expand all files Collapse all files

28
alerting/alerts.py
View file

@ -6,7 +6,8 @@ from typing import Optional
import aiofiles
import nio
from alerting.common import CREDS_FILE, ROOM_ID
from alerting.common import CONFIG_FILE
from misc import cvars
class AlertType(StrEnum):
@ -41,13 +42,11 @@ async def get_client() -> nio.AsyncClient:
Returns a Matrix client.
It is better to call get_client once and use it for multiple send_alert calls
"""
async with aiofiles.open(CREDS_FILE) as f:
contents = await f.read()
creds = json.loads(contents)
client = nio.AsyncClient(creds["homeserver"])
client.access_token = creds["access_token"]
client.user_id = creds["user_id"]
client.device_id = creds["device_id"]
matrix_cfg = cvars.config.get()["matrix"]
client = nio.AsyncClient(matrix_cfg["homeserver"])
client.access_token = matrix_cfg["access_token"]
client.user_id = matrix_cfg["user_id"]
client.device_id = matrix_cfg["device_id"]
return client
@ -67,12 +66,19 @@ def format_message(alert: Alert) -> str:
return message, None
async def send_alert(alert: Alert, client: Optional[nio.AsyncClient] = None) -> None:
if client is None:
async def send_alert(alert: Alert) -> None:
try:
client = cvars.matrix_client.get()
except LookupError: # being called standalone
async with aiofiles.open(CONFIG_FILE) as f:
contents = await f.read()
cvars.config.set(json.loads(contents))
temp_client = True
client = await get_client()
cvars.matrix_client.set(client)
else:
temp_client = False
room_id = cvars.config.get()["matrix"]["room_id"]
message, html_message = format_message(alert)
content = {
"msgtype": "m.text",
@ -82,7 +88,7 @@ async def send_alert(alert: Alert, client: Optional[nio.AsyncClient] = None) ->
content["format"] = "org.matrix.custom.html"
content["formatted_body"] = html_message
await client.room_send(
room_id=ROOM_ID,
room_id=room_id,
message_type="m.room.message",
content=content,
)