79 lines
3.4 KiB
Python
79 lines
3.4 KiB
Python
from fastapi import Request
|
|
|
|
from ..chat import Target, default_targets
|
|
from ..models import KIND_FLAPPING, KIND_PROBLEM, KIND_RECOVERY, STATE_NAMES
|
|
from ..routing import DEFAULT_MIN_STATE, Route, RouteInput
|
|
|
|
CHOICE_SEP = "|"
|
|
EXTRA_SEP = ":"
|
|
KIND_LABELS = {KIND_PROBLEM: "Problem", KIND_RECOVERY: "Recovery", KIND_FLAPPING: "Flapping"}
|
|
|
|
|
|
def _choice_value(target: Target) -> str:
|
|
return f"{target.backend}{CHOICE_SEP}{target.target}"
|
|
|
|
|
|
def _optional_int(form, name: str, label: str, errors: list[str]) -> int | None:
|
|
raw = str(form.get(name, "")).strip()
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return int(raw)
|
|
except ValueError:
|
|
errors.append(f"{label} muss eine ganze Zahl sein.")
|
|
return None
|
|
|
|
|
|
def _parse_targets(form, errors: list[str]) -> tuple[Target, ...]:
|
|
targets: list[Target] = []
|
|
for value in form.getlist("targets"):
|
|
backend, _, target = str(value).partition(CHOICE_SEP)
|
|
targets.append(Target(backend, target))
|
|
for line in str(form.get("extra_targets", "")).splitlines():
|
|
if not line.strip():
|
|
continue
|
|
backend, sep, target = line.strip().partition(EXTRA_SEP)
|
|
if not sep:
|
|
errors.append(f"Zielzeile ungültig: {line.strip()[:60]} (erwartet backend:ziel).")
|
|
continue
|
|
targets.append(Target(backend.strip().lower(), target.strip()))
|
|
return tuple(dict.fromkeys(targets))
|
|
|
|
|
|
def parse_form(form) -> tuple[RouteInput, list[str]]:
|
|
errors: list[str] = []
|
|
min_state = _optional_int(form, "min_state", "Mindest-Schweregrad", errors)
|
|
data = RouteInput(
|
|
name=str(form.get("name", "")).strip(), match_host=str(form.get("match_host", "")).strip(),
|
|
match_service=str(form.get("match_service", "")).strip(),
|
|
min_state=DEFAULT_MIN_STATE if min_state is None else min_state,
|
|
kinds=tuple(str(k) for k in form.getlist("kinds")), stop_on_match=bool(form.get("stop_on_match")),
|
|
enabled=bool(form.get("enabled")), targets=_parse_targets(form, errors),
|
|
delay_minutes=_optional_int(form, "delay_minutes", "Verzögerung", errors),
|
|
output_contains=str(form.get("output_contains", "")).strip(),
|
|
ai_enabled=bool(form.get("ai_enabled")), ai_hint=str(form.get("ai_hint", "")).strip(),
|
|
reminder_hours=_optional_int(form, "reminder_hours", "Erinnerung", errors),
|
|
loud=bool(form.get("loud")), use_default_targets=bool(form.get("use_default_targets")),
|
|
)
|
|
return data, errors
|
|
|
|
|
|
def form_context(request: Request, data: RouteInput) -> dict:
|
|
choices = default_targets(request.app.state.store)
|
|
free = [t for t in data.targets if t not in choices]
|
|
return {
|
|
"form": data,
|
|
"choices": [{"value": _choice_value(t), "label": f"{t.backend}: {t.target}"} for t in choices],
|
|
"selected": {_choice_value(t) for t in data.targets if t in choices},
|
|
"extra_text": "\n".join(f"{t.backend}{EXTRA_SEP}{t.target}" for t in free),
|
|
"state_options": sorted(STATE_NAMES.items()),
|
|
"kind_options": KIND_LABELS,
|
|
}
|
|
|
|
|
|
def to_input(route: Route) -> RouteInput:
|
|
return RouteInput(route.name, route.match_host, route.match_service, route.min_state, route.kinds,
|
|
route.stop_on_match, route.enabled, route.targets, route.delay_minutes,
|
|
route.output_contains, route.ai_enabled, route.ai_hint, route.reminder_hours,
|
|
route.loud, route.use_default_targets)
|