+from scapy.config import conf
+from mac_vendor_lookup import MacLookup
+from typing import List, Optional, Tuple, Dict
+import socket
+import logging
+import netifaces
+import ipaddress
+from concurrent.futures import ThreadPoolExecutor
+import time
+from functools import lru_cache
+import json
+from pathlib import Path
+from datetime import datetime, timedelta
+import subprocess
+
+from scapy.layers.l2 import ARP, Ether
+from scapy.sendrecv import srp
+from utils.i18n import get_translations
+from babel.support import Translations
+from fastapi import Depends
+
+from models.subnet import SubnetScanResult, Device
+
+logging.basicConfig(
+ level=logging.INFO,
+ format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
+)
+logger = logging.getLogger(__name__)
+
+
+
[docs]
+
def get_subnet_service(translations: Translations = Depends(get_translations())):
+
"""Фабричная функция для создания экземпляра SubnetService.
+
+
Использует зависимости FastAPI для получения объекта переводов и возвращает
+
экземпляр SubnetService с текущей локализацией.
+
+
Args:
+
translations (Translations): Объект переводов для текущего языка.
+
+
Returns:
+
SubnetService: Экземпляр сервиса для работы с подсетями.
+
"""
+
return SubnetService(translations)
+
[docs]
+
class NetworkScanner:
+
"""Класс для сканирования подсети с использованием ARP-запросов.
+
+
Назначение:
+
Обнаруживает активные устройства в локальной сети, отправляя широковещательные
+
ARP‑запросы по указанной подсети и собирая ответы (IP, MAC).
+
+
Детали реализации:
+
- Использует `scapy` (`Ether` + `ARP`) для генерации пакетов и `srp` для
+
отправки/получения на канальном уровне от имени заданного интерфейса.
+
- Таймаут ожидания ответа короткий (2 секунды), ориентирован на интерактивные
+
сценарии UI.
+
+
Ограничения и замечания:
+
- Требуются права для отправки низкоуровневых пакетов (capabilities/root).
+
- В сетях с фильтрацией широковещания количество обнаруженных устройств
+
может быть ограничено.
+
"""
+
def __init__(self, interface: str, translations: Translations):
+
"""Инициализирует сканер сети.
+
+
Устанавливает интерфейс для сканирования и объект переводов.
+
+
Args:
+
interface (str): Имя сетевого интерфейса (например, "wlp0s20f3").
+
translations (Translations): Объект переводов для текущего языка.
+
"""
+
self.interface = interface
+
conf.iface = interface
+
self.translations = translations
+
+
+
[docs]
+
def scan(self, subnet: str) -> List[Tuple[str, str]]:
+
"""Сканирует подсеть с использованием ARP-запросов.
+
+
Отправляет ARP-пакеты для обнаружения активных устройств и возвращает
+
список кортежей (IP, MAC).
+
+
Args:
+
subnet (str): Подсеть в формате CIDR (например, "10.1.1.0/24").
+
+
Returns:
+
List[Tuple[str, str]]: Список кортежей (IP-адрес, MAC-адрес) обнаруженных устройств.
+
+
Raises:
+
Exception: Если произошла ошибка сканирования.
+
"""
+
try:
+
arp = ARP(pdst=subnet)
+
except Exception as e:
+
logger.error(f"Error checking status: {str(e)}")
+
try:
+
arp = ARP(pdst=subnet)
+
ether = Ether(dst="ff:ff:ff:ff:ff:ff")
+
packet = ether / arp
+
result = srp(packet, timeout=2, verbose=False, iface=self.interface)[0]
+
return [(received.psrc, received.hwsrc) for sent, received in result]
+
except Exception as e:
+
logger.error(f"Error scanning subnet {subnet}: {str(e)}")
+
return []
+
+
[docs]
+
class PortScanner:
+
"""Класс для сканирования открытых портов на устройствах.
+
+
Назначение:
+
Плейсхолдер для будущего функционала активного сканирования TCP/UDP‑портов
+
у обнаруженных узлов. В текущей версии возвращает пустой список.
+
+
План развития:
+
- Добавить неблокирующее сканирование наиболее распространённых портов
+
(22, 80, 443 и т.д.) с ограничением по времени.
+
- Настраиваемые профили глубины и скорости сканирования.
+
"""
+
def __init__(self, interface: str, translations: Translations):
+
"""Инициализирует сканер портов.
+
+
Устанавливает интерфейс для сканирования и объект переводов.
+
+
Args:
+
interface (str): Имя сетевого интерфейса.
+
translations (Translations): Объект переводов для текущего языка.
+
"""
+
self.interface = interface
+
self.translations = translations
+
+
+
[docs]
+
def scan(self, ip: str) -> List[str]:
+
"""Сканирует открытые порты на указанном IP-адресе.
+
+
Пока возвращает пустой список, так как функциональность не реализована.
+
+
Args:
+
ip (str): IP-адрес устройства для сканирования.
+
+
Returns:
+
List[str]: Список открытых портов (пока пустой).
+
"""
+
return []
+
+
[docs]
+
class DeviceInfoCollector:
+
"""Класс для сбора информации об устройствах, включая вендоров и хостнеймы.
+
+
Источники данных:
+
- MacLookup: сопоставление префикса MAC (OUI) с названием производителя.
+
- DNS: `gethostbyaddr` с малым таймаутом для обратного разрешения имени.
+
- NetBIOS (nbtscan): опрос сегмента сети для извлечения NetBIOS‑имён.
+
+
Политика кэширования:
+
- Результаты `get_vendor` кэшируются с помощью `functools.lru_cache`.
+
- Результаты `nbtscan` аккумулируются в словарях `nbtscan_cache`/`nbtscan_results`
+
на время жизни экземпляра.
+
"""
+
def __init__(self, translations: Translations):
+
"""Инициализирует коллектор информации об устройствах.
+
+
Настраивает кэширование и объект переводов.
+
+
Args:
+
translations (Translations): Объект переводов для текущего языка.
+
"""
+
self.mac_lookup = MacLookup()
+
self.nbtscan_cache = {}
+
self.nbtscan_results = {}
+
self.translations = translations
+
+
+
[docs]
+
@lru_cache(maxsize=1000)
+
def get_vendor(self, mac: str) -> str:
+
"""Получает вендора устройства по MAC-адресу.
+
+
Использует кэширование для повышения производительности.
+
+
Args:
+
mac (str): MAC-адрес устройства.
+
+
Returns:
+
str: Название вендора или "Неизвестный" в случае ошибки.
+
"""
+
try:
+
return self.mac_lookup.lookup(mac)
+
except Exception as e:
+
logger.warning(f"Failed to determine vendor for MAC {mac}: {str(e)}")
+
return self.translations.gettext("Неизвестный")
+
+
+
+
[docs]
+
def scan_nbtscan(self, subnet: str) -> None:
+
"""Сканирует подсеть с помощью nbtscan для получения хостнеймов.
+
+
Сохраняет результаты в кэш и логирует процесс.
+
+
Args:
+
subnet (str): Подсеть в формате CIDR.
+
"""
+
if subnet not in self.nbtscan_results:
+
try:
+
logger.debug(f"Starting nbtscan for subnet {subnet}")
+
result = subprocess.run(["sudo", "nbtscan", "-r", str(subnet)],
+
capture_output=True,
+
text=True,
+
check=False,
+
timeout=10)
+
logger.debug(f"Full nbtscan output: {result.stdout}")
+
if result.returncode == 0:
+
for line in result.stdout.splitlines():
+
logger.debug(f"Processing line: {line}")
+
if not line.startswith(("IP address", "-", "Doing NBT")):
+
parts = [p for p in line.split() if p]
+
if len(parts) >= 2:
+
ip_addr, hostname = parts[0], parts[1]
+
if hostname != "<unknown>":
+
self.nbtscan_cache[ip_addr] = hostname
+
logger.info(f"NetBIOS: Found hostname {hostname} for IP {ip_addr}")
+
self.nbtscan_results[subnet] = True
+
except subprocess.TimeoutExpired:
+
logger.error(f"NetBIOS: Timeout for subnet {subnet}")
+
except Exception as e:
+
logger.error(f"NetBIOS: Error for subnet {subnet}: {str(e)}")
+
+
+
+
[docs]
+
def get_hostname(self, ip: str) -> Optional[str]:
+
"""Получает хостнейм для указанного IP-адреса.
+
+
Пробует сначала DNS, затем NetBIOS.
+
+
Args:
+
ip (str): IP-адрес устройства.
+
+
Returns:
+
Optional[str]: Хостнейм или None, если не удалось определить.
+
"""
+
try:
+
socket.setdefaulttimeout(2)
+
return socket.gethostbyaddr(ip)[0]
+
except (socket.herror, socket.gaierror, socket.timeout):
+
logger.debug(f"DNS: Failed to resolve name for IP {ip}")
+
+
subnet = ipaddress.ip_network(f"{ip}/24", strict=False)
+
if subnet not in self.nbtscan_results:
+
self.scan_nbtscan(str(subnet))
+
+
if ip in self.nbtscan_cache:
+
return self.nbtscan_cache[ip]
+
+
logger.debug(f"Failed to determine hostname for IP {ip} via DNS or NetBIOS")
+
return None
+
+
[docs]
+
class DeviceStorage:
+
"""Класс для хранения и управления данными об устройствах.
+
+
Формат файла подсети:
+
JSON‑структура вида
+
{
+
"devices": [Device, ...]
+
}
+
+
Политика очистки:
+
- Метод `_clean_old_devices` удаляет оффлайн‑устройства, не наблюдавшиеся
+
более 7 дней, чтобы поддерживать актуальность списка.
+
"""
+
def __init__(self, base_path: str = "storage/subnets", translations: Translations = None):
+
"""Инициализирует хранилище устройств.
+
+
Создаёт директорию, если её нет, и задаёт параметры очистки.
+
+
Args:
+
base_path (str): Путь к директории хранения данных.
+
translations (Translations, optional): Объект переводов.
+
"""
+
self.base_path = Path(base_path)
+
self.base_path.mkdir(parents=True, exist_ok=True)
+
self.offline_timeout = timedelta(days=7)
+
self.translations = translations or Translations.load("locales", ["ru"], "messages") # Fallback на русский
+
+
+
[docs]
+
def get_subnet_file(self, subnet: str) -> Path:
+
"""Возвращает путь к файлу для конкретной подсети.
+
+
Заменяет символы '/' на '_' для корректного имени файла.
+
+
Args:
+
subnet (str): Подсеть в формате CIDR.
+
+
Returns:
+
Path: Путь к файлу.
+
"""
+
sanitized_subnet = subnet.replace("/", "_")
+
return self.base_path / f"{sanitized_subnet}.json"
+
+
+
+
[docs]
+
def load_devices(self, subnet: str) -> Dict[str, Device]:
+
"""Загружает устройства для указанной подсети.
+
+
Читает данные из JSON-файла и фильтрует устаревшие устройства.
+
+
Args:
+
subnet (str): Подсеть в формате CIDR.
+
+
Returns:
+
Dict[str, Device]: Словарь устройств с MAC как ключом.
+
"""
+
file_path = self.get_subnet_file(subnet)
+
try:
+
if file_path.exists():
+
with open(file_path, "r", encoding="utf-8") as f:
+
data = json.load(f)
+
if not isinstance(data, dict) or "devices" not in data:
+
logger.warning(f"Invalid JSON structure in {file_path}, expected {'devices': [...]}, got {data}")
+
return {}
+
devices = {}
+
for device_data in data.get("devices", []):
+
device = Device(**device_data)
+
if device.mac:
+
devices[device.mac] = device
+
return self._clean_old_devices(devices)
+
return {}
+
except json.JSONDecodeError as e:
+
logger.error(f"JSON decode error in {file_path}: {str(e)}")
+
return {}
+
except Exception as e:
+
logger.error(f"Error loading devices for subnet {subnet}: {str(e)}")
+
return {}
+
+
+
def _clean_old_devices(self, devices: Dict[str, Device]) -> Dict[str, Device]:
+
"""Очищает список устройств от устаревших оффлайн-устройств.
+
+
Удаляет устройства, которые оффлайн более 7 дней.
+
+
Args:
+
devices (Dict[str, Device]): Словарь устройств.
+
+
Returns:
+
Dict[str, Device]: Отфильтрованный словарь устройств.
+
"""
+
now = datetime.now()
+
filtered = {}
+
for mac, device in devices.items():
+
if device.status == "offline" and device.last_seen:
+
try:
+
last_seen = datetime.fromisoformat(device.last_seen)
+
if now - last_seen <= self.offline_timeout:
+
filtered[mac] = device
+
else:
+
logger.info(f"Removed device {mac} (offline > 7 days)")
+
except ValueError:
+
logger.warning(f"Invalid last_seen format for device {mac}")
+
else:
+
filtered[mac] = device
+
return filtered
+
+
+
[docs]
+
def save_devices(self, subnet: str, devices: List[Device]) -> None:
+
"""Сохраняет устройства для указанной подсети.
+
+
Записывает данные в JSON-файл.
+
+
Args:
+
subnet (str): Подсеть в формате CIDR.
+
devices (List[Device]): Список устройств для сохранения.
+
+
Raises:
+
Exception: Если произошла ошибка при сохранении.
+
"""
+
file_path = self.get_subnet_file(subnet)
+
try:
+
data = SubnetScanResult(devices=devices).model_dump()
+
with open(file_path, "w", encoding="utf-8") as f:
+
json.dump(data, f, indent=2, ensure_ascii=False)
+
logger.info(f"Saved {len(devices)} devices to {file_path}")
+
except Exception as e:
+
logger.error(f"Error saving devices for subnet {subnet}: {str(e)}")
+
if self.translations:
+
raise Exception(self.translations.gettext("Ошибка при сохранении устройств: {error}").format(error=str(e)))
+
else:
+
raise Exception(f"Ошибка при сохранении устройств: {str(e)}")
+
+
[docs]
+
class SubnetService:
+
"""Сервис для сканирования подсетей и управления устройствами.
+
+
Роль сервиса:
+
Объединяет сетевое сканирование (ARP), сбор дополнительной информации
+
(вендор, хостнейм), слияние с кэшем предыдущих результатов и персистентное
+
хранение данных по подсетям. Предоставляет вспомогательные операции для
+
локального узла и работы с заметками устройств.
+
+
Поток выполнения скана (`scan_subnet`):
+
1) Определение подсети (вручную или авто по интерфейсу).
+
2) Загрузка ранее сохранённых устройств подсети.
+
3) Добавление локального устройства (если принадлежит подсети).
+
4) ARP‑сканирование и параллельная обработка узлов (обогащение данных).
+
5) Маркировка оффлайн‑узлов из кэша, не встреченных в текущем скане.
+
6) Сортировка по IP и сохранение объединённого списка в JSON.
+
"""
+
def __init__(self, translations: Translations):
+
"""Инициализирует сервис подсети.
+
+
Настраивает интерфейс, сканеры и хранилище с текущей локализацией.
+
+
Args:
+
translations (Translations): Объект переводов для текущего языка.
+
"""
+
self.interface = self._get_default_interface()
+
self.network_scanner = NetworkScanner(self.interface, translations)
+
self.port_scanner = PortScanner(self.interface, translations)
+
self.device_info = DeviceInfoCollector(translations)
+
self.storage = DeviceStorage(translations=translations)
+
self.translations = translations
+
logger.info(f"Initialized SubnetService with interface: {self.interface}")
+
+
def _get_default_interface(self) -> str:
+
"""Определяет интерфейс по умолчанию.
+
+
Использует информацию о шлюзе или первый доступный интерфейс.
+
+
Returns:
+
str: Имя интерфейса (например, "wlp0s20f3" или "eth0").
+
+
Raises:
+
Exception: Если не удалось определить интерфейс.
+
"""
+
try:
+
gateways = netifaces.gateways()
+
default_gateway = gateways.get("default", {}).get(netifaces.AF_INET)
+
if default_gateway:
+
iface = default_gateway[1]
+
logger.info(f"Default interface from gateway: {iface}")
+
return iface
+
interfaces = netifaces.interfaces()
+
for iface in interfaces:
+
if iface != "lo":
+
addrs = netifaces.ifaddresses(iface)
+
if netifaces.AF_INET in addrs:
+
logger.info(f"Selected interface: {iface}")
+
return iface
+
logger.warning("No suitable interface found, defaulting to eth0")
+
return "eth0"
+
except Exception as e:
+
logger.warning(f"Failed to determine interface: {str(e)}, defaulting to eth0")
+
return "eth0"
+
+
def _get_default_subnet(self) -> str:
+
"""Автоматически определяет подсеть локального устройства.
+
+
Использует информацию о текущем интерфейсе для определения подсети.
+
+
Returns:
+
str: Подсеть в формате CIDR (например, "10.1.1.0/24").
+
+
Raises:
+
Exception: Если не удалось определить подсеть.
+
"""
+
try:
+
interfaces = netifaces.interfaces()
+
for iface in interfaces:
+
if iface == "lo" or iface != self.interface:
+
continue
+
addrs = netifaces.ifaddresses(iface)
+
if netifaces.AF_INET in addrs:
+
ip_info = addrs[netifaces.AF_INET][0]
+
ip = ip_info["addr"]
+
netmask = ip_info["netmask"]
+
# Преобразуем маску из строки (например, "255.255.255.0") в CIDR
+
mask = ipaddress.ip_network(f"0.0.0.0/{netmask}", strict=False).prefixlen
+
network = ipaddress.ip_network(f"{ip}/{mask}", strict=False)
+
logger.info(f"Auto-detected subnet: {network} (IP: {ip}, Mask: {netmask} -> /{mask})")
+
return str(network)
+
logger.warning("No suitable subnet found, defaulting to 0.0.0.0/0")
+
return "0.0.0.0/0"
+
except Exception as e:
+
logger.error(f"Error determining default subnet: {str(e)}")
+
return "0.0.0.0/0"
+
+
+
[docs]
+
def scan_subnet(self, subnet: Optional[str] = None) -> SubnetScanResult:
+
"""Сканирует указанную подсеть и возвращает результаты.
+
+
Выполняет ARP-сканирование, собирает информацию о устройствах и сохраняет данные.
+
+
Args:
+
subnet (Optional[str]): Подсеть в формате CIDR (например, "10.1.1.0/24").
+
Если None или пустая строка, определяется автоматически.
+
+
Returns:
+
SubnetScanResult: Объект с результатами сканирования, содержащий список
+
устройств или сообщение об ошибке.
+
+
Raises:
+
Exception: Если произошла ошибка сканирования.
+
"""
+
start_time = time.time()
+
devices = []
+
now = datetime.now().astimezone()
+
formatted_now = now.strftime("%d-%m-%Y, %H:%M")
+
try:
+
# Определяем подсеть, если не указана или пустая
+
if not subnet or subnet.strip() == "":
+
subnet = self._get_default_subnet()
+
if subnet == "0.0.0.0/0":
+
logger.warning("Default subnet is invalid, scan aborted")
+
return SubnetScanResult(devices=devices, error=self.translations.gettext("Недопустимая подсеть по умолчанию"))
+
+
saved_devices = self.storage.load_devices(subnet)
+
+
local_device = self._get_local_device(subnet)
+
if local_device:
+
local_device.status = "online"
+
local_device.last_seen = formatted_now
+
devices.append(local_device)
+
if local_device.mac:
+
saved_devices[local_device.mac] = local_device
+
logger.info(f"Detected local device: {local_device.ip}, MAC: {local_device.mac}")
+
+
logger.info(f"Scanning subnet: {subnet} with interface: {self.interface}")
+
discovered_devices = self.network_scanner.scan(subnet)
+
logger.info(f"Discovered devices: {len(discovered_devices)}")
+
self.device_info.scan_nbtscan(subnet)
+
+
if not discovered_devices and not local_device:
+
logger.warning(f"No devices found in subnet {subnet}")
+
return SubnetScanResult(devices=[], error=self.translations.gettext("В данной подсети не обнаружено устройств"))
+
+
def process_device(ip_mac: Tuple[str, str]) -> Optional[Device]:
+
ip, mac = ip_mac
+
if local_device and ip == local_device.ip:
+
return None
+
+
existing_device = saved_devices.get(mac)
+
if existing_device:
+
return Device(
+
ip=ip,
+
mac=mac,
+
hostname=self.device_info.get_hostname(ip) or existing_device.hostname,
+
vendor=self.device_info.get_vendor(mac),
+
open_ports=existing_device.open_ports,
+
status="online",
+
last_seen=formatted_now,
+
note=existing_device.note,
+
)
+
vendor = self.device_info.get_vendor(mac)
+
hostname = self.device_info.get_hostname(ip)
+
return Device(
+
ip=ip,
+
mac=mac,
+
hostname=hostname,
+
vendor=vendor,
+
open_ports=[],
+
status="online",
+
last_seen=formatted_now,
+
note=None,
+
)
+
+
with ThreadPoolExecutor(max_workers=10) as executor:
+
device_results = executor.map(process_device, discovered_devices)
+
devices.extend([d for d in device_results if d is not None])
+
+
active_macs = {d.mac for d in devices if d.mac}
+
for mac, device in saved_devices.items():
+
if mac not in active_macs:
+
device.status = "offline"
+
devices.append(device)
+
+
devices.sort(key=lambda x: int(ipaddress.IPv4Address(x.ip)))
+
+
for device in devices:
+
if device.mac:
+
saved_devices[device.mac] = device
+
self.storage.save_devices(subnet, list(saved_devices.values()))
+
+
logger.info(
+
f"Subnet scan {subnet} completed in {time.time() - start_time:.2f} seconds, "
+
f"found {len(devices)} devices (active: {len(active_macs)})"
+
)
+
return SubnetScanResult(devices=devices)
+
+
except Exception as e:
+
logger.error(f"Error scanning subnet {subnet}: {str(e)}")
+
return SubnetScanResult(devices=devices, error=self.translations.gettext("Ошибка сканирования: {error}").format(error=str(e)))
+
+
+
def _get_local_device(self, subnet: str) -> Optional[Device]:
+
"""Определяет локальное устройство в указанной подсети.
+
+
Использует информацию об интерфейсе для обнаружения.
+
+
Args:
+
subnet (str): Подсеть в формате CIDR.
+
+
Returns:
+
Optional[Device]: Объект устройства или None, если не найдено.
+
"""
+
now = datetime.now().astimezone()
+
formatted_now = now.strftime("%d-%m-%Y, %H:%M")
+
try:
+
subnet_network = ipaddress.ip_network(subnet, strict=False)
+
interfaces = netifaces.interfaces()
+
for iface in interfaces:
+
if iface == "lo":
+
continue
+
addrs = netifaces.ifaddresses(iface)
+
if netifaces.AF_INET in addrs and netifaces.AF_LINK in addrs:
+
ip = addrs[netifaces.AF_INET][0]["addr"]
+
if ipaddress.ip_address(ip) in subnet_network:
+
mac = addrs[netifaces.AF_LINK][0]["addr"]
+
vendor = self.device_info.get_vendor(mac)
+
hostname = self.device_info.get_hostname(ip)
+
logger.info(f"Local device detected: IP={ip}, MAC={mac}, Interface={iface}")
+
return Device(
+
ip=ip,
+
mac=mac,
+
hostname=hostname,
+
vendor=vendor,
+
open_ports=[],
+
status="online",
+
last_seen=formatted_now,
+
note=None,
+
)
+
logger.warning(f"Local device not found in subnet {subnet}")
+
return None
+
except Exception as e:
+
logger.warning(f"Failed to get local device: {str(e)}")
+
return None
+
+
+
[docs]
+
def add_note(self, mac: str, note: str, translations: Translations) -> Tuple[bool, Optional[Tuple[str, Dict]]]:
+
"""Добавляет заметку для устройства по MAC-адресу.
+
+
Привязывает заметку к устройству, если оно существует.
+
+
Args:
+
mac (str): MAC-адрес устройства.
+
note (str): Текст заметки.
+
translations (Translations): Объект переводов для текущего языка.
+
+
Returns:
+
Tuple[bool, Optional[Tuple[str, Dict]]]: Кортеж (успех, результат),
+
где результат содержит код ошибки/успеха и данные.
+
"""
+
try:
+
saved_devices = {}
+
for subnet in self.storage.base_path.glob("*.json"):
+
devices = self.storage.load_devices(subnet.stem.replace("_", "/"))
+
saved_devices.update(devices)
+
if mac not in saved_devices:
+
logger.warning(f"Device with MAC {mac} not found")
+
return False, ("not_found", {"mac": mac})
+
if saved_devices[mac].note is not None:
+
logger.error(f"Note for device with MAC {mac} already exists")
+
return False, ("note_exists", {"mac": mac})
+
for subnet_file in self.storage.base_path.glob("*.json"):
+
subnet = subnet_file.stem.replace("_", "/")
+
devices = self.storage.load_devices(subnet)
+
if mac in devices:
+
devices[mac].note = note
+
self.storage.save_devices(subnet, list(devices.values()))
+
break
+
logger.info(f"Added note '{note}' for device with MAC {mac}")
+
return True, ("success", {"mac": mac, "note": note})
+
except Exception as e:
+
logger.error(f"Error adding note for MAC {mac}: {str(e)}")
+
return False, ("internal_error", {"error": str(e)})
+
+
+
+
[docs]
+
def update_note(self, mac: str, note: str, translations: Translations) -> Tuple[bool, Optional[Tuple[str, Dict]]]:
+
"""Обновляет заметку для устройства по MAC-адресу.
+
+
Изменяет существующую заметку, если она есть.
+
+
Args:
+
mac (str): MAC-адрес устройства.
+
note (str): Новый текст заметки.
+
translations (Translations): Объект переводов для текущего языка.
+
+
Returns:
+
Tuple[bool, Optional[Tuple[str, Dict]]]: Кортеж (успех, результат),
+
где результат содержит код ошибки/успеха и данные.
+
"""
+
try:
+
saved_devices = {}
+
for subnet in self.storage.base_path.glob("*.json"):
+
devices = self.storage.load_devices(subnet.stem.replace("_", "/"))
+
saved_devices.update(devices)
+
if mac not in saved_devices:
+
logger.warning(f"Device with MAC {mac} not found")
+
return False, ("not_found", {"mac": mac})
+
if saved_devices[mac].note is None:
+
logger.error(f"Note for device with MAC {mac} does not exist")
+
return False, ("note_not_exists", {"mac": mac})
+
for subnet_file in self.storage.base_path.glob("*.json"):
+
subnet = subnet_file.stem.replace("_", "/")
+
devices = self.storage.load_devices(subnet)
+
if mac in devices:
+
devices[mac].note = note
+
self.storage.save_devices(subnet, list(devices.values()))
+
break
+
logger.info(f"Updated note '{note}' for device with MAC {mac}")
+
return True, ("success", {"mac": mac, "note": note})
+
except Exception as e:
+
logger.error(f"Error updating note for MAC {mac}: {str(e)}")
+
return False, ("internal_error", {"error": str(e)})
+
+
+
+
[docs]
+
def delete_note(self, mac: str, translations: Translations) -> Tuple[bool, Optional[Tuple[str, Dict]]]:
+
"""Удаляет заметку для устройства по MAC-адресу.
+
+
Удаляет привязанную заметку, если она существует.
+
+
Args:
+
mac (str): MAC-адрес устройства.
+
translations (Translations): Объект переводов для текущего языка.
+
+
Returns:
+
Tuple[bool, Optional[Tuple[str, Dict]]]: Кортеж (успех, результат),
+
где результат содержит код ошибки/успеха и данные.
+
"""
+
try:
+
saved_devices = {}
+
for subnet in self.storage.base_path.glob("*.json"):
+
devices = self.storage.load_devices(subnet.stem.replace("_", "/"))
+
saved_devices.update(devices)
+
if mac not in saved_devices:
+
logger.warning(f"Device with MAC {mac} not found")
+
return False, ("not_found", {"mac": mac})
+
if saved_devices[mac].note is None:
+
logger.error(f"Note for device with MAC {mac} does not exist")
+
return False, ("note_not_exists", {"mac": mac})
+
for subnet_file in self.storage.base_path.glob("*.json"):
+
subnet = subnet_file.stem.replace("_", "/")
+
devices = self.storage.load_devices(subnet)
+
if mac in devices:
+
devices[mac].note = None
+
self.storage.save_devices(subnet, list(devices.values()))
+
break
+
logger.info(f"Deleted note for device with MAC {mac}")
+
return True, ("success", {"mac": mac})
+
except Exception as e:
+
logger.error(f"Error deleting note for MAC {mac}: {str(e)}")
+
return False, ("internal_error", {"error": str(e)})
+