gonets/SRV/api/v1/network.py

from fastapi import APIRouter, Body, HTTPException, Depends
import logging
from typing import Dict, Tuple, List, Union
from services.network_service import NetworkService
from models.network import (
    NetworkInterfacesContainer,
    Activation,
    Deactivation,
    NetworkInterface,
    WiFiDevice,
)
from utils.i18n import get_translations, _, _load_translations
from babel.support import Translations

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

router = APIRouter()
network_service = NetworkService()

class ContainerPositionManager:
    """Класс для управления позициями и состоянием контейнеров.

    Хранит и обновляет координаты (x, y) и состояние (isOpen) для контейнеров
    активации и деактивации.
    """

    def __init__(self):
        """Инициализирует менеджер позиций контейнеров.

        Создаёт начальные позиции и состояния для контейнеров "activation" и
        "deactivation" с координатами (0, 0) и закрытым состоянием (False).
        """
        self.positions: Dict[str, Dict[str, Union[int, bool]]] = {
            "activation": {
                "x": 0,
                "y": 0,
                "isOpen": False
            },
            "deactivation": {
                "x": 0,
                "y": 0,
                "isOpen": False
            },
        }

    def get_position_and_state(self, container_name: str, translations: Translations) -> Tuple[int, int, bool]:
        """Получает текущие координаты и состояние указанного контейнера.

        Args:
            container_name (str): Имя контейнера ("activation" или "deactivation").
            translations: Объект переводов для текущего языка.

        Returns:
            Tuple[int, int, bool]: Кортеж с координатами (x, y) и состоянием (isOpen).

        Raises:
            ValueError: Если указан неверный идентификатор контейнера.
        """
        container = self.positions.get(container_name)
        if not container:
            raise ValueError(translations.gettext("Неверное имя контейнера: {container_name}").format(container_name=container_name))
        return container["x"], container["y"], container["isOpen"]

    def update_position(self, container_id: int, x: int, y: int, isOpen: bool) -> None:
        """Обновляет позицию и состояние контейнера по его идентификатору.

        Args:
            container_id (int): Идентификатор контейнера (1 для "activation", 2 для "deactivation").
            x (int): Новая координата X.
            y (int): Новая координата Y.
            isOpen (bool): Новое состояние контейнера (открыт/закрыт).

        Raises:
            ValueError: Если container_id не соответствует ни одному контейнеру.
        """
        container_name = "activation" if container_id == 1 else "deactivation"
        self.positions[container_name] = {"x": x, "y": y, "isOpen": isOpen}

container_manager = ContainerPositionManager()

async def get_network_data(translations: Translations = Depends(get_translations)) -> Dict[str, Dict]:
    """Возвращает текущие сетевые данные в виде словаря.

    Получает данные о сетевых интерфейсах, разделяет их на активные и неактивные,
    и возвращает их в формате Activation и Deactivation с учётом позиций и состояний
    контейнеров.

    Args:
        translations: Объект переводов для текущего языка.

    Returns:
        Dict[str, Dict]: Словарь с ключами "activation" и "deactivation", содержащими
            данные о соответствующих контейнерах.

    Raises:
        HTTPException: Если произошла ошибка при получении данных (status_code=500).
    """
    logger.debug("Calling get_network_data()")
    try:
        network_data = await network_service.get_network_interfaces_container()

        activation_interfaces, deactivation_interfaces = _split_interfaces(network_data.interfaces)

        activation_x, activation_y, activation_is_open = container_manager.get_position_and_state("activation", translations)
        deactivation_x, deactivation_y, deactivation_is_open = container_manager.get_position_and_state("deactivation", translations)

        activation = Activation(
            id=1,
            interfaces=activation_interfaces,
            container_x=activation_x,
            container_y=activation_y,
            isOpen=activation_is_open,
        )
        deactivation = Deactivation(
            id=2,
            interfaces=deactivation_interfaces,
            container_x=deactivation_x,
            container_y=deactivation_y,
            isOpen=deactivation_is_open,
        )

        return {
            "activation": activation.model_dump(),
            "deactivation": deactivation.model_dump(),
        }
    except Exception as e:
        logger.error(f"Error in get_network_data: {e}")
        raise HTTPException(status_code=500, detail=translations.gettext("Не удалось получить сетевые данные"))

def _split_interfaces(
    interfaces: List[Union[NetworkInterface, WiFiDevice]],
) -> Tuple[List[Union[NetworkInterface, WiFiDevice]], List[Union[NetworkInterface, WiFiDevice]]]:
    """Разделяет интерфейсы на активные и неактивные.

    Args:
        interfaces (List[Union[NetworkInterface, WiFiDevice]]): Список сетевых интерфейсов.

    Returns:
        Tuple[List[Union[NetworkInterface, WiFiDevice]], List[Union[NetworkInterface, WiFiDevice]]]:
            Кортеж из списков активных и неактивных интерфейсов.
    """
    activation_interfaces = []
    deactivation_interfaces = []
    for interface in interfaces:
        if interface.device_state == "100":
            activation_interfaces.append(interface)
        else:
            deactivation_interfaces.append(interface)
    return activation_interfaces, deactivation_interfaces

@router.post("/update_container_position")
async def update_container_position(data: Dict[str, Union[int, bool]] = Body(...), translations: Translations = Depends(get_translations)):
    """Обновляет позицию и состояние контейнера.

    Принимает данные о новом положении и состоянии контейнера и обновляет их
    в менеджере позиций.

    Args:
        data (Dict[str, Union[int, bool]]): Словарь с ключами:
            - "container_id" (int): Идентификатор контейнера (1 или 2).
            - "container_x" (int): Новая координата X.
            - "container_y" (int): Новая координата Y.
            - "isOpen" (bool): Новое состояние контейнера (открыт/закрыт).
        translations: Объект переводов для текущего языка.

    Returns:
        dict: Словарь с ключами:
            - "success" (bool): Успешность операции.
            - "message" (str): Сообщение об успешном обновлении.

    Raises:
        HTTPException: Если отсутствуют обязательные поля (status_code=400) или
            типы данных неверны.
    """
    try:
        required_keys = [
            "container_id", "container_x", "container_y", "isOpen"
        ]
        if not all(key in data for key in required_keys):
            raise ValueError(
                translations.gettext("Отсутствуют обязательные поля: container_id, container_x, container_y, isOpen")
            )

        container_id = data["container_id"]
        container_x = data["container_x"]
        container_y = data["container_y"]
        is_open = data["isOpen"]

        if (not isinstance(container_id, int)
                or not isinstance(container_x, int)
                or not isinstance(container_y, int)
                or not isinstance(is_open, bool)):
            raise ValueError(translations.gettext("Неверный тип данных в запросе"))

        container_manager.update_position(container_id, container_x, container_y, is_open)
        return {
            "success": True,
            "message": translations.gettext("Позиция и состояние контейнера обновлены")
        }
    except Exception as e:
        logger.error(f"Error updating container position: {e}")
        raise HTTPException(status_code=400, detail=translations.gettext("Не удалось обновить положение контейнера: {error}").format(error=str(e)))

@router.get("/get_hostname")
async def get_hostname(translations: Translations = Depends(get_translations)):
    """Возвращает имя хоста системы.

    Выполняет запрос к сервису для получения имени хоста.

    Args:
        translations: Объект переводов для текущего языка.

    Returns:
        dict: Словарь с ключом "hostname" и значением имени хоста.

    Raises:
        HTTPException: Если произошла ошибка при получении имени хоста (status_code=500).
    """
    try:
        hostname_info = await network_service.get_hostname()
        return hostname_info
    except Exception as e:
        logger.error(f"Error in get_hostname: {e}")
        raise HTTPException(status_code=500, detail=translations.gettext("Не удалось получить имя хоста"))