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 class NetworkService: """Сервис для взаимодействия с NetworkManager через D-Bus""" def __init__(self): self.system_bus = sd_bus_open_system() self.nm = NetworkManager(self.system_bus) self.nm_settings = NetworkManagerSettings(self.system_bus) async def get_ip_method(self, device: NetworkDeviceGeneric, ip_type: str) -> str: """Получает метод IP для устройства""" 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) для устройства""" 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: # print(f"Ошибка при получении соединения для устройства {await device.interface}: {e}") # срёт ошибки дл отключенных интерфейсов # добавить потом условия для отключенных интерфейсов return 'unknown' async def get_ip_config(self, device: NetworkDeviceGeneric, ip_version: str) -> List[IndividualIPConfig]: """Получает конфигурацию 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 устройстве""" 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]]: """Получает список сетевых интерфейсов""" 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""" 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(f"Failed to execute hostname: {e}") except Exception as e: raise RuntimeError(f"Error processing hostname output: {e}") async def get_interfaces_by_state(self, state: str, initial_positions: Dict[str, Tuple[int, int]] = None) -> 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: """ Получает объект, содержащий список интерфейсов и координаты контейнера """ 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: """ Получает контейнер с сетевыми интерфейсами и координатами контейнера """ 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]: """ Получает список активных сетевых интерфейсов """ 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]: """ Получает список неактивных сетевых интерфейсов """ 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: """ Получает объект, содержащий список активных интерфейсов и координаты контейнера """ 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: """ Получает объект, содержащий список неактивных интерфейсов и координаты контейнера """ interfaces = await self.get_deactivated_interfaces(initial_positions) return NetworkInterfacesContainer( interfaces=interfaces, container_x=container_x, container_y=container_y )