ayurishchevandClaude Sonnet 5
4c1841b61b
Control Server (FastAPI) с WEB UI (Jinja2 + HTMX) и JSON API для группового администрирования устройств RouterOS 7 через REST. - Устройства: список, статус (модель, канал, версии ROS/FW, uptime, доступные обновления), примечания, пароли шифруются (Fernet); фоновый опрос каждые 30 с и быстрое обнаружение недоступности (таймаут соединения 4 с). - Группы устройств и фильтры (устройства, резервные копии). - Резервные копии: .backup и .rsc (show-sensitive) создаются через REST, скачиваются сервером и загружаются в S3 (Yandex Object Storage). - Обновление ROS/FW и выбор канала, групповые операции задачами. - Интерфейс по утверждённому макету: светлая/тёмная темы, окна, шрифты IBM Plex локально. Docker Compose, SQLite в томе, минимальный набор тестов. - Планы и итоги каждого изменения — в docs/changes/. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
69 lines
2.6 KiB
Python
69 lines
2.6 KiB
Python
from sqlalchemy import func, update
|
|
|
|
from app.db import session_scope
|
|
from app.models import Device, Group
|
|
|
|
|
|
def _clean(name: str) -> str:
|
|
name = (name or "").strip()
|
|
if not name or len(name) > 64:
|
|
raise ValueError("Название группы: от 1 до 64 символов")
|
|
return name
|
|
|
|
|
|
def list_groups() -> list[dict]:
|
|
"""Группы с числом устройств, по алфавиту."""
|
|
with session_scope() as s:
|
|
counts = dict(s.query(Device.group_id, func.count(Device.id)).group_by(Device.group_id))
|
|
return [{"id": g.id, "name": g.name, "device_count": counts.get(g.id, 0)}
|
|
for g in s.query(Group).order_by(Group.name)]
|
|
|
|
|
|
def get_group(group_id: int) -> Group:
|
|
with session_scope() as s:
|
|
g = s.get(Group, group_id)
|
|
if g is None:
|
|
raise LookupError(f"Группа {group_id} не найдена")
|
|
return g
|
|
|
|
|
|
def create_group(name: str) -> Group:
|
|
name = _clean(name)
|
|
with session_scope() as s:
|
|
if s.query(Group).filter_by(name=name).first():
|
|
raise ValueError(f"Группа «{name}» уже существует")
|
|
g = Group(name=name)
|
|
s.add(g)
|
|
s.flush()
|
|
return g
|
|
|
|
|
|
def rename_group(group_id: int, name: str) -> Group:
|
|
name = _clean(name)
|
|
with session_scope() as s:
|
|
g = s.get(Group, group_id)
|
|
if g is None:
|
|
raise LookupError(f"Группа {group_id} не найдена")
|
|
if s.query(Group).filter(Group.name == name, Group.id != group_id).first():
|
|
raise ValueError(f"Группа «{name}» уже существует")
|
|
g.name = name
|
|
return g
|
|
|
|
|
|
def delete_group(group_id: int) -> None:
|
|
"""Удаляет группу; устройства остаются и становятся «Без группы» (FK в SQLite не включены)."""
|
|
with session_scope() as s:
|
|
g = s.get(Group, group_id)
|
|
if g is None:
|
|
raise LookupError(f"Группа {group_id} не найдена")
|
|
s.execute(update(Device).where(Device.group_id == group_id).values(group_id=None))
|
|
s.delete(g)
|
|
|
|
|
|
def move_devices(device_ids: list[int], group_id: int | None) -> None:
|
|
"""Переносит устройства в группу (None — «Без группы»)."""
|
|
with session_scope() as s:
|
|
if group_id is not None and s.get(Group, group_id) is None:
|
|
raise LookupError(f"Группа {group_id} не найдена")
|
|
s.execute(update(Device).where(Device.id.in_(device_ids)).values(group_id=group_id))
|