from typing import List, Dict, Tuple, Union, Optional from sdbus_async.networkmanager import ( NetworkManager, NetworkDeviceGeneric, IPv4Config, IPv6Config, NetworkManagerSettings, NetworkDeviceWireless, AccessPoint, DeviceType, ) from sdbus import sd_bus_open_system from models.network import ( NetworkInterface, IPv4Route, IPv6Route, IPConfig, IndividualIPConfig, NetworkInterfacesContainer, WiFiDevice, ) import subprocess from utils.i18n import get_translations, _ from babel.support import Translations from fastapi import Depends class NetworkService: """Сервис для взаимодействия с NetworkManager через D-Bus. Предоставляет асинхронные методы для получения информации о сетевых интерфейсах, их конфигурациях IP, Wi-Fi и состояниях. """ def __init__(self, translations: Translations = Depends(get_translations("services"))): """Инициализирует сервис NetworkManager. Устанавливает системный D-Bus и создаёт объекты для взаимодействия с NetworkManager и его настройками. """ self.system_bus = sd_bus_open_system() self.nm = NetworkManager(self.system_bus) self.nm_settings = NetworkManagerSettings(self.system_bus) self.translations = translations async def get_ip_method(self, device: NetworkDeviceGeneric, ip_type: str) -> str: """Получает метод IP (например, "auto", "manual") для указанного устройства. Args: device (NetworkDeviceGeneric): Объект устройства NetworkManager. ip_type (str): Тип IP (например, "ipv4" или "ipv6"). Returns: str: Метод IP или "unknown" в случае ошибки. Raises: Exception: Если произошла ошибка при получении настроек соединения. """ try: config = await device.get_applied_connection() settings = config[0].get(ip_type, {}) return str(settings.get("method", ("s", "unknown"))[1]) except Exception: return "unknown" async def get_connection_name(self, device: NetworkDeviceGeneric) -> str: """Извлекает имя соединения (GENERAL.CONNECTION) для устройства. Args: device (NetworkDeviceGeneric): Объект устройства NetworkManager. Returns: str: Имя соединения или "unknown" в случае ошибки. Raises: Exception: Если произошла ошибка при получении настроек соединения. """ try: applied_connection = await device.get_applied_connection() connection_settings = applied_connection[0] connection_info = connection_settings.get("connection", {}) return str(connection_info.get("id", ("s", "unknown"))[1]) except Exception as e: return "unknown" async def get_ip_config(self, device: NetworkDeviceGeneric, ip_version: str) -> List[IndividualIPConfig]: """Получает конфигурацию IP для устройства. Args: device (NetworkDeviceGeneric): Объект устройства NetworkManager. ip_version (str): Версия IP ("4" или "6"). Returns: List[IndividualIPConfig]: Список конфигураций IP, включая адреса и маршруты. Raises: Exception: Если произошла ошибка при получении данных IP. """ configs = [] device_ip = await getattr(device, f"ip{ip_version}_config") if device_ip != "/": ip_config = globals()[f"IPv{ip_version}Config"](device_ip, self.system_bus) address_data = await ip_config.address_data for address in address_data: config = IndividualIPConfig( address=str(address["address"][1]), routes=[ globals()[f"IPv{ip_version}Route"]( destination=route.get("dest", ("s", "0.0.0.0" if ip_version == "4" else "::"))[1], prefix=route.get("prefix", ("u", 0))[1], next_hop=route.get("next-hop", ("s", "0.0.0.0" if ip_version == "4" else "::"))[1], metric=route.get("metric", ("u", 0))[1], ) for route in await ip_config.route_data ], method=await self.get_ip_method(device, f"ipv{ip_version}"), gateway=str(await ip_config.gateway), dns=[ str(ns["address"][1]) if ip_version == "4" else str(ns) for ns in await getattr( ip_config, "nameserver_data" if ip_version == "4" else "nameservers", ) ], ) configs.append(config) return configs async def get_wifi_info(self, wifi_device: NetworkDeviceWireless) -> Dict[str, Optional[Union[str, int]]]: """Получает информацию о Wi-Fi устройстве. Args: wifi_device (NetworkDeviceWireless): Объект Wi-Fi устройства NetworkManager. Returns: Dict[str, Optional[Union[str, int]]]: Словарь с данными Wi-Fi (SSID, BSSID, bitrate, strength, frequency, mode), где значения могут быть None. Raises: Exception: Если произошла ошибка при получении данных точки доступа. """ active_ap = await wifi_device.active_access_point wifi_info = { "ssid": None, "bssid": None, "bitrate": await wifi_device.bitrate, "strength": None, "frequency": None, "mode": None, } if active_ap != "/": ap = AccessPoint(active_ap, self.system_bus) wifi_info.update({ "ssid": await ap.ssid, "bssid": await ap.hw_address, "strength": await ap.strength, "frequency": await ap.frequency, "mode": await ap.mode, }) return wifi_info async def get_network_interfaces( self, initial_positions: Dict[str, Tuple[int, int]] = None ) -> List[Union[NetworkInterface, WiFiDevice]]: """Получает список сетевых интерфейсов. Args: initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных координат (x, y) для интерфейсов. По умолчанию None (используются (100, 100)). Returns: List[Union[NetworkInterface, WiFiDevice]]: Список объектов интерфейсов, включая Wi-Fi устройства. Исключает интерфейсы типа LOOPBACK и неактивные состояния (state == "10"). """ devices_paths = await self.nm.get_devices() network_interfaces = [] initial_positions = initial_positions or {} for device_path in devices_paths: generic_device = NetworkDeviceGeneric(device_path, self.system_bus) interface_name = str(await generic_device.interface) device_type = DeviceType(await generic_device.device_type).name if device_type == "LOOPBACK": continue device_state = str(await generic_device.state) if device_state == "10": continue common_data = { "name": interface_name, "type": device_type, "device_state": device_state, "active_connection": await self.get_connection_name(generic_device), "mac_address": str(await generic_device.hw_address), "device_mtu": str(await generic_device.mtu), "ipv4": IPConfig(configurations=await self.get_ip_config(generic_device, "4")), "ipv6": IPConfig(configurations=await self.get_ip_config(generic_device, "6")), "x": initial_positions.get(interface_name, (100, 100))[0], "y": initial_positions.get(interface_name, (100, 100))[1], } if device_type == "WIFI": wifi_device = NetworkDeviceWireless(device_path, self.system_bus) wifi_info = await self.get_wifi_info(wifi_device) network_interfaces.append(WiFiDevice(**common_data, **wifi_info)) else: network_interfaces.append(NetworkInterface(**common_data)) return network_interfaces async def get_hostname(self) -> Dict[str, str]: """Возвращает имя хоста из команды hostname в формате JSON. Returns: Dict[str, str]: Словарь с ключом "hostname" и значением имени хоста. Raises: RuntimeError: Если произошла ошибка при выполнении команды hostname или обработке вывода. """ try: result = subprocess.run(["hostname"], capture_output=True, text=True, check=True) hostname = result.stdout.strip() return {"hostname": hostname} except subprocess.CalledProcessError as e: raise RuntimeError(self.translations.gettext("Не удалось выполнить команду hostname: {error}").format(error=str(e))) except Exception as e: raise RuntimeError(self.translations.gettext("Ошибка обработки вывода hostname: {error}").format(error=str(e))) async def get_interfaces_by_state( self, state: str, initial_positions: Dict[str, Tuple[int, int]] = None ) -> List[NetworkInterface]: """Получает список интерфейсов по указанному состоянию. Args: state (str): Состояние интерфейса (например, "100" для активных). initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных координат (x, y) для интерфейсов. По умолчанию None. Returns: List[NetworkInterface]: Список интерфейсов с указанным состоянием. """ all_interfaces = await self.get_network_interfaces(initial_positions) return [ interface for interface in all_interfaces if interface.device_state == state ] async def get_interfaces_container( self, state: Optional[str] = None, initial_positions: Dict[str, Tuple[int, int]] = None, container_x: int = 0, container_y: int = 0, ) -> NetworkInterfacesContainer: """Получает объект, содержащий список интерфейсов и координаты контейнера. Args: state (Optional[str], optional): Состояние интерфейсов (например, "100"). По умолчанию None (все интерфейсы). initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных координат (x, y) для интерфейсов. По умолчанию None. container_x (int, optional): Координата X контейнера. По умолчанию 0. container_y (int, optional): Координата Y контейнера. По умолчанию 0. Returns: NetworkInterfacesContainer: Объект с интерфейсами и координатами контейнера. """ if state: interfaces = await self.get_interfaces_by_state(state, initial_positions) else: interfaces = await self.get_network_interfaces(initial_positions) return NetworkInterfacesContainer(interfaces=interfaces, container_x=container_x, container_y=container_y) async def get_network_interfaces_container( self, initial_positions: Optional[Dict[str, Tuple[int, int]]] = None, ) -> NetworkInterfacesContainer: """Получает контейнер с сетевыми интерфейсами и координатами контейнера. Args: initial_positions (Optional[Dict[str, Tuple[int, int]]], optional): Словарь начальных координат (x, y) для интерфейсов. По умолчанию None. Returns: NetworkInterfacesContainer: Объект с интерфейсами и координатами (0, 0). """ interfaces = await self.get_network_interfaces(initial_positions) return NetworkInterfacesContainer(interfaces=interfaces, container_x=0, container_y=0) async def get_activated_interfaces( self, initial_positions: Dict[str, Tuple[int, int]] = None ) -> List[NetworkInterface]: """Получает список активных сетевых интерфейсов. Args: initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных координат (x, y) для интерфейсов. По умолчанию None. Returns: List[NetworkInterface]: Список активных интерфейсов (state == "100"). """ return await self.get_interfaces_by_state("100", initial_positions) async def get_deactivated_interfaces( self, initial_positions: Dict[str, Tuple[int, int]] = None ) -> List[NetworkInterface]: """Получает список неактивных сетевых интерфейсов. Args: initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных координат (x, y) для интерфейсов. По умолчанию None. Returns: List[NetworkInterface]: Список интерфейсов с state != "100". """ all_interfaces = await self.get_network_interfaces(initial_positions) return [ interface for interface in all_interfaces if interface.device_state != "100" ] async def get_activated_interfaces_container( self, initial_positions: Dict[str, Tuple[int, int]] = None, container_x: int = 0, container_y: int = 0, ) -> NetworkInterfacesContainer: """Получает объект, содержащий список активных интерфейсов и координаты контейнера. Args: initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных координат (x, y) для интерфейсов. По умолчанию None. container_x (int, optional): Координата X контейнера. По умолчанию 0. container_y (int, optional): Координата Y контейнера. По умолчанию 0. Returns: NetworkInterfacesContainer: Объект с активными интерфейсами и координатами. """ return await self.get_interfaces_container("100", initial_positions, container_x, container_y) async def get_deactivated_interfaces_container( self, initial_positions: Dict[str, Tuple[int, int]] = None, container_x: int = 0, container_y: int = 0, ) -> NetworkInterfacesContainer: """Получает объект, содержащий список неактивных интерфейсов и координаты контейнера. Args: initial_positions (Dict[str, Tuple[int, int]], optional): Словарь начальных координат (x, y) для интерфейсов. По умолчанию None. container_x (int, optional): Координата X контейнера. По умолчанию 0. container_y (int, optional): Координата Y контейнера. По умолчанию 0. Returns: NetworkInterfacesContainer: Объект с неактивными интерфейсами и координатами. """ interfaces = await self.get_deactivated_interfaces(initial_positions) return NetworkInterfacesContainer(interfaces=interfaces, container_x=container_x, container_y=container_y)