50 lines
1.4 KiB
Python
50 lines
1.4 KiB
Python
from dataclasses import dataclass, replace
|
|
from typing import Iterable, Sequence
|
|
|
|
from ..globs import glob_matches
|
|
from ..models import SUPPRESS_MUTE, Event
|
|
|
|
SOURCE_UI = "ui"
|
|
SOURCE_CHAT = "chat"
|
|
SOURCES = (SOURCE_UI, SOURCE_CHAT)
|
|
|
|
MATCH_ALL = "*"
|
|
MAX_MUTE_MINUTES = 7 * 24 * 60
|
|
DURATION_CHOICES = ((30, "30 Minuten"), (60, "1 Stunde"), (120, "2 Stunden"), (240, "4 Stunden"),
|
|
(480, "8 Stunden"), (1440, "24 Stunden"))
|
|
DEFAULT_DURATION_MINUTES = 120
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Mute:
|
|
id: int
|
|
host_glob: str
|
|
service_glob: str
|
|
expires_at: int
|
|
reason: str
|
|
created_by: str
|
|
created_at: int
|
|
source: str
|
|
|
|
|
|
def is_active(mute: Mute, now: float) -> bool:
|
|
return mute.expires_at > now
|
|
|
|
|
|
def find_active_mute(mutes: Iterable[Mute], host: str, service: str, now: float) -> Mute | None:
|
|
for mute in mutes:
|
|
if is_active(mute, now) and glob_matches(mute.host_glob, host) and glob_matches(mute.service_glob, service):
|
|
return mute
|
|
return None
|
|
|
|
|
|
def apply_mutes(events: Sequence[Event], mutes: Sequence[Mute], now: float) -> list[Event]:
|
|
if not mutes:
|
|
return list(events)
|
|
result = []
|
|
for event in events:
|
|
if not event.suppressed and find_active_mute(mutes, event.host, event.service, now):
|
|
event = replace(event, suppressed=True, suppress_reason=SUPPRESS_MUTE)
|
|
result.append(event)
|
|
return result
|