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