ros_control/app/services/devices.py

import re
from dataclasses import dataclass

from app import security
from app.config import get_settings
from app.db import session_scope
from app.models import Device, Group
from app.ros.operations import version_newer
from app.ros.client import RosClient

NAME_RE = re.compile(r"^[A-Za-z0-9._-]{1,64}$")  # имя попадает в ключи S3 — только безопасные символы


@dataclass
class Conn:
    id: int
    name: str
    host: str
    port: int
    username: str
    password: str
    verify_tls: bool
    use_tls: bool


def validate_name(name: str) -> None:
    if not NAME_RE.match(name):
        raise ValueError("Имя: латиница, цифры, '.', '_', '-' (до 64 символов)")


def list_devices() -> list[Device]:
    with session_scope() as s:
        return list(s.query(Device).order_by(Device.name))


def get_device(device_id: int) -> Device:
    with session_scope() as s:
        d = s.get(Device, device_id)
        if d is None:
            raise LookupError(f"Устройство {device_id} не найдено")
        return d


def _check_group(s, group_id: int | None) -> None:
    if group_id is not None and s.get(Group, group_id) is None:
        raise LookupError(f"Группа {group_id} не найдена")


NOTE_MAX = 500


def _clean_note(note: str | None) -> str | None:
    note = (note or "").strip()
    if len(note) > NOTE_MAX:
        raise ValueError(f"Примечание: не более {NOTE_MAX} символов")
    return note or None


def _resolve_group(s, group_id: int | None, new_group: str | None) -> int | None:
    """Группа устройства. new_group создаёт группу (или берёт существующую с таким именем)
    в той же транзакции, что и устройство: при ошибке не остаётся лишней группы."""
    name = (new_group or "").strip()
    if name:
        if len(name) > 64:
            raise ValueError("Название группы: от 1 до 64 символов")
        g = s.query(Group).filter_by(name=name).first()
        if g is None:
            g = Group(name=name)
            s.add(g)
            s.flush()
        return g.id
    _check_group(s, group_id)
    return group_id


def create_device(name, host, port, username, password, verify_tls=False, use_tls=True,
                  group_id: int | None = None, new_group: str | None = None, note: str | None = None) -> Device:
    validate_name(name)
    note = _clean_note(note)
    with session_scope() as s:
        group_id = _resolve_group(s, group_id, new_group)
        if s.query(Device).filter_by(name=name).first():
            raise ValueError(f"Устройство с именем {name} уже существует")
        d = Device(name=name, host=host, port=port, username=username,
                   password_enc=security.encrypt(password), verify_tls=verify_tls, use_tls=use_tls,
                   group_id=group_id, note=note)
        s.add(d)
        s.flush()
        return d


def update_device(device_id: int, **fields) -> Device:
    with session_scope() as s:
        d = s.get(Device, device_id)
        if d is None:
            raise LookupError(f"Устройство {device_id} не найдено")
        if fields.get("name") and fields["name"] != d.name:
            validate_name(fields["name"])
            d.name = fields["name"]
        for k in ("host", "port", "username", "verify_tls", "use_tls"):
            if fields.get(k) is not None:
                setattr(d, k, fields[k])
        if "group_id" in fields or fields.get("new_group"):  # group_id=None — открепить от группы
            d.group_id = _resolve_group(s, fields.get("group_id"), fields.get("new_group"))
        if "note" in fields:  # пустое примечание — очистить
            d.note = _clean_note(fields["note"])
        if fields.get("password"):  # пустой пароль = не менять
            d.password_enc = security.encrypt(fields["password"])
        return d


def delete_device(device_id: int) -> None:
    with session_scope() as s:
        d = s.get(Device, device_id)
        if d is None:
            raise LookupError(f"Устройство {device_id} не найдено")
        s.delete(d)


def get_conn(device_id: int) -> Conn:
    d = get_device(device_id)
    return Conn(d.id, d.name, d.host, d.port, d.username,
                security.decrypt(d.password_enc), d.verify_tls, d.use_tls)


def open_client(conn: Conn) -> RosClient:
    return RosClient(conn.host, conn.port, conn.username, conn.password,
                     conn.verify_tls, get_settings().ros_timeout, conn.use_tls,
                     connect_timeout=get_settings().ros_connect_timeout)


def has_ros_update(st: dict) -> bool:
    return version_newer(st.get("ros_latest"), st.get("ros_installed"))


def has_fw_update(st: dict) -> bool:
    return bool(st.get("fw_current") and st.get("fw_upgrade") and st["fw_upgrade"] != st["fw_current"])


def filter_devices(devs: list[Device], group: str = "", q: str = "", status: str = "",
                   updates: str = "", channel: str = "") -> list[Device]:
    """Общий фильтр списка устройств для UI и API (по кэшу статуса).

    group: "" — все, "none" — без группы, иначе id группы. status: online/offline.
    updates: ros / fw / any (есть обновление) / none (обновлений нет).
    """
    q = q.strip().lower()
    out = []
    for d in devs:
        st = d.status
        if group == "none" and d.group_id is not None:
            continue
        if group not in ("", "none") and str(d.group_id) != group:
            continue
        if q and q not in " ".join((d.name, d.host, st.get("model") or "")).lower():
            continue
        if status == "online" and d.online is not True:
            continue
        if status == "offline" and d.online is True:
            continue
        ros_new, fw_new = has_ros_update(st), has_fw_update(st)
        if (updates == "ros" and not ros_new) or (updates == "fw" and not fw_new):
            continue
        if (updates == "any" and not (ros_new or fw_new)) or (updates == "none" and (ros_new or fw_new)):
            continue
        if channel and st.get("channel") != channel:
            continue
        out.append(d)
    return out