mirror of
https://github.com/deunix-educ/Fail2banMqttActionBanishment.git
synced 2026-08-24 03:11:58 +02:00
27 lines
1.0 KiB
Python
27 lines
1.0 KiB
Python
import json
|
|
from typing import Any
|
|
|
|
from channels.generic.websocket import AsyncWebsocketConsumer
|
|
|
|
GROUP_NAME = 'banevents'
|
|
|
|
|
|
class BanEventConsumer(AsyncWebsocketConsumer):
|
|
"""Pousse chaque nouveau BanEvent aux clients connectés au tableau de bord."""
|
|
|
|
async def connect(self) -> None:
|
|
await self.channel_layer.group_add(GROUP_NAME, self.channel_name)
|
|
await self.accept()
|
|
|
|
async def disconnect(self, close_code: int) -> None:
|
|
await self.channel_layer.group_discard(GROUP_NAME, self.channel_name)
|
|
|
|
async def ban_event(self, event: dict[str, Any]) -> None:
|
|
await self.send(text_data=json.dumps({'kind': 'ban', **event['payload']}))
|
|
|
|
async def command_notification(self, event: dict[str, Any]) -> None:
|
|
"""Issue d'une commande reçue du master (banevents.management.
|
|
commands.master_client) — succès/échec d'un SYNC_BAN, ou commande
|
|
non gérée. Même group_send que ban_event, discriminé par `kind`."""
|
|
await self.send(text_data=json.dumps({'kind': 'command', **event['payload']}))
|