patrickandClaude Sonnet 5.5
347f55cf7c
Includes atomic migrations, per-rule reminder semantics, first-run helper script and the note on the undecided UI chat idea. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
93 lines
3.6 KiB
Python
93 lines
3.6 KiB
Python
from .. import audit
|
|
from ..db import Database
|
|
from .repository import RouteRepository
|
|
from .rules import Route
|
|
from .validation import RouteInput, validate_route
|
|
|
|
DIRECTION_UP = "up"
|
|
DIRECTION_DOWN = "down"
|
|
|
|
|
|
class RouteError(ValueError):
|
|
def __init__(self, errors: list[str]):
|
|
super().__init__(" ".join(errors))
|
|
self.errors = errors
|
|
|
|
|
|
def describe(route: Route | RouteInput) -> str:
|
|
targets = ", ".join(f"{t.backend}:{t.target}" for t in route.targets) or "keine"
|
|
state = "an" if route.enabled else "aus"
|
|
delay = "std" if route.delay_minutes is None else route.delay_minutes
|
|
reminder = "std" if route.reminder_hours is None else route.reminder_hours
|
|
return (f"{route.name} [{state}] host={route.match_host or '*'} service={route.match_service or '*'} "
|
|
f"min_state={route.min_state} kinds={','.join(route.kinds)} "
|
|
f"stop={int(route.stop_on_match)} delay={delay} contains={route.output_contains!r} "
|
|
f"ai={int(route.ai_enabled)} hint_len={len(route.ai_hint)} "
|
|
f"reminder_h={reminder} loud={int(route.loud)} "
|
|
f"default_targets={int(route.use_default_targets)} -> {targets}")
|
|
|
|
|
|
class RouteService:
|
|
def __init__(self, db: Database):
|
|
self._db = db
|
|
self._repo = RouteRepository(db)
|
|
|
|
def list_all(self) -> list[Route]:
|
|
return self._repo.list_all()
|
|
|
|
def enabled_routes(self) -> list[Route]:
|
|
return [route for route in self._repo.list_all() if route.enabled]
|
|
|
|
def get(self, route_id: int) -> Route | None:
|
|
return self._repo.get(route_id)
|
|
|
|
def create(self, data: RouteInput, actor: str) -> int:
|
|
self._check(data)
|
|
route_id = self._repo.insert(data)
|
|
audit.record(self._db, actor, "route.create", data.name.strip(), None, describe(data))
|
|
return route_id
|
|
|
|
def update(self, route_id: int, data: RouteInput, actor: str) -> bool:
|
|
old = self._repo.get(route_id)
|
|
if old is None:
|
|
return False
|
|
self._check(data)
|
|
self._repo.update(route_id, data)
|
|
audit.record(self._db, actor, "route.update", data.name.strip(), describe(old), describe(data))
|
|
return True
|
|
|
|
def set_enabled(self, route_id: int, enabled: bool, actor: str) -> bool:
|
|
old = self._repo.get(route_id)
|
|
if old is None:
|
|
return False
|
|
self._repo.set_enabled(route_id, enabled)
|
|
audit.record(self._db, actor, "route.toggle", old.name, "an" if old.enabled else "aus",
|
|
"an" if enabled else "aus")
|
|
return True
|
|
|
|
def delete(self, route_id: int, actor: str) -> bool:
|
|
old = self._repo.get(route_id)
|
|
if old is None or not self._repo.delete(route_id):
|
|
return False
|
|
audit.record(self._db, actor, "route.delete", old.name, describe(old), None)
|
|
return True
|
|
|
|
def move(self, route_id: int, direction: str, actor: str) -> bool:
|
|
routes = self._repo.list_all()
|
|
index = next((i for i, r in enumerate(routes) if r.id == route_id), None)
|
|
if index is None:
|
|
return False
|
|
other = index - 1 if direction == DIRECTION_UP else index + 1
|
|
if direction not in (DIRECTION_UP, DIRECTION_DOWN) or not 0 <= other < len(routes):
|
|
return False
|
|
self._repo.swap_positions(routes[index], routes[other])
|
|
audit.record(self._db, actor, "route.move", routes[index].name,
|
|
str(routes[index].position), str(routes[other].position))
|
|
return True
|
|
|
|
@staticmethod
|
|
def _check(data: RouteInput) -> None:
|
|
errors = validate_route(data)
|
|
if errors:
|
|
raise RouteError(errors)
|