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, _ 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 = Depends(get_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")) deactivation_x, deactivation_y, deactivation_is_open = (container_manager.get_position_and_state("deactivation")) 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("Не удалось получить имя хоста"))