ros_control/app/services/groups.py

from sqlalchemy import func, update

from app import ids
from app.db import session_scope
from app.models import Device, Group
from app.services import events


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: str) -> Group:
    ids.check(group_id, "grp")
    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()
        events.record("group.created", "group", g.id, f"Группа «{name}» создана", data={"name": name}, s=s)
        return g


def rename_group(group_id: str, name: str) -> Group:
    ids.check(group_id, "grp")
    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}» уже существует")
        old, g.name = g.name, name
        if old != name:
            events.record("group.renamed", "group", g.id, f"Группа «{old}» переименована в «{name}»",
                          data={"old": old, "new": name}, s=s)
        return g


def delete_group(group_id: str) -> None:
    """Удаляет группу; устройства остаются и становятся «Без группы» (FK в SQLite не включены)."""
    ids.check(group_id, "grp")
    with session_scope() as s:
        g = s.get(Group, group_id)
        if g is None:
            raise LookupError(f"Группа {group_id} не найдена")
        detached = [d for (d,) in s.query(Device.id).filter(Device.group_id == group_id)]
        s.execute(update(Device).where(Device.group_id == group_id).values(group_id=None))
        events.record("group.deleted", "group", g.id, f"Группа «{g.name}» удалена", data={"name": g.name, "detached_devices": detached}, s=s)
        s.delete(g)


def move_devices(device_ids: list[str], group_id: str | None) -> None:
    """Переносит устройства в группу (None — «Без группы»)."""
    for i in device_ids:
        ids.check(i, "dev")
    with session_scope() as s:
        if group_id is not None:
            ids.check(group_id, "grp")
            if 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))
        events.record("devices.moved", "group", group_id, f"Перенесено устройств: {len(device_ids)}",
                      data={"device_ids": device_ids, "group_id": group_id}, s=s)