72 lines
2.4 KiB
Python
72 lines
2.4 KiB
Python
from dataclasses import dataclass
|
|
from typing import Sequence
|
|
|
|
from ..chat.base import Target
|
|
from ..globs import glob_matches
|
|
from ..models import KIND_FLAPPING, KIND_PROBLEM, KIND_RECOVERY, KIND_REMINDER, Event
|
|
from .limits import DEFAULT_MIN_STATE, MIN_STATE_RANGE
|
|
|
|
ALL_KINDS = (KIND_PROBLEM, KIND_RECOVERY, KIND_FLAPPING)
|
|
|
|
__all__ = ["ALL_KINDS", "DEFAULT_MIN_STATE", "MIN_STATE_RANGE", "Route", "matches", "matching_rules",
|
|
"resolve_targets", "route_matches"]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Route:
|
|
id: int
|
|
name: str
|
|
position: int
|
|
enabled: bool
|
|
match_host: str
|
|
match_service: str
|
|
min_state: int
|
|
kinds: tuple[str, ...]
|
|
stop_on_match: bool
|
|
targets: tuple[Target, ...]
|
|
delay_minutes: int | None = None
|
|
output_contains: str = ""
|
|
ai_enabled: bool = True
|
|
ai_hint: str = ""
|
|
reminder_hours: int | None = None
|
|
loud: bool = True
|
|
use_default_targets: bool = False
|
|
|
|
|
|
def matches(route: Route, host: str, service: str, kind: str, state: int, output: str) -> bool:
|
|
# Erinnerungen folgen den Regeln für Probleme, damit keine eigene Alarmart nötig ist.
|
|
kind = KIND_PROBLEM if kind == KIND_REMINDER else kind
|
|
if not route.enabled or kind not in route.kinds:
|
|
return False
|
|
if kind == KIND_PROBLEM and state < route.min_state:
|
|
return False
|
|
if route.output_contains and route.output_contains.lower() not in output.lower():
|
|
return False
|
|
return glob_matches(route.match_host, host) and glob_matches(route.match_service, service)
|
|
|
|
|
|
def route_matches(route: Route, event: Event) -> bool:
|
|
return matches(route, event.host, event.service, event.kind, event.state, event.output)
|
|
|
|
|
|
def matching_rules(rules: Sequence[Route], event: Event) -> list[Route]:
|
|
found: list[Route] = []
|
|
for rule in sorted(rules, key=lambda r: (r.position, r.id)):
|
|
if not route_matches(rule, event):
|
|
continue
|
|
found.append(rule)
|
|
if rule.stop_on_match:
|
|
break
|
|
return found
|
|
|
|
|
|
def resolve_targets(rules: Sequence[Route], event: Event, defaults: Sequence[Target]) -> list[Target]:
|
|
found = matching_rules(rules, event)
|
|
if not found:
|
|
return list(dict.fromkeys(defaults))
|
|
resolved: list[Target] = []
|
|
for rule in found:
|
|
extra = defaults if rule.use_default_targets else ()
|
|
resolved.extend(t for t in (*rule.targets, *extra) if t not in resolved)
|
|
return resolved
|