225 lines
10 KiB
Python
225 lines
10 KiB
Python
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("Не удалось получить имя хоста"))
|