Разделение сканирования на разные подсети. Хранение каждой из просканированных подсетей. Исправление багов с заметками и обновленные локали
This commit is contained in:
parent
c326edfb9a
commit
b1306ff66f
10 changed files with 1124 additions and 285 deletions
|
|
@ -27,14 +27,53 @@ logging.basicConfig(
|
|||
)
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def get_subnet_service(translations: Translations = Depends(get_translations())):
|
||||
"""Фабричная функция для создания экземпляра SubnetService.
|
||||
|
||||
Использует зависимости FastAPI для получения объекта переводов и возвращает
|
||||
экземпляр SubnetService с текущей локализацией.
|
||||
|
||||
Args:
|
||||
translations (Translations): Объект переводов для текущего языка.
|
||||
|
||||
Returns:
|
||||
SubnetService: Экземпляр сервиса для работы с подсетями.
|
||||
"""
|
||||
return SubnetService(translations)
|
||||
|
||||
class NetworkScanner:
|
||||
"""Класс для сканирования подсети с использованием ARP-запросов."""
|
||||
def __init__(self, interface: str, translations: Translations = Depends(get_translations())):
|
||||
"""Класс для сканирования подсети с использованием ARP-запросов.
|
||||
|
||||
Выполняет отправку ARP-запросов для обнаружения устройств в указанной подсети.
|
||||
"""
|
||||
def __init__(self, interface: str, translations: Translations):
|
||||
"""Инициализирует сканер сети.
|
||||
|
||||
Устанавливает интерфейс для сканирования и объект переводов.
|
||||
|
||||
Args:
|
||||
interface (str): Имя сетевого интерфейса (например, "wlp0s20f3").
|
||||
translations (Translations): Объект переводов для текущего языка.
|
||||
"""
|
||||
self.interface = interface
|
||||
conf.iface = interface
|
||||
self.translations = translations
|
||||
|
||||
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)
|
||||
ether = Ether(dst="ff:ff:ff:ff:ff:ff")
|
||||
|
|
@ -46,17 +85,48 @@ class NetworkScanner:
|
|||
return []
|
||||
|
||||
class PortScanner:
|
||||
"""Класс для сканирования открытых портов на устройствах."""
|
||||
def __init__(self, interface: str, translations: Translations = Depends(get_translations())):
|
||||
"""Класс для сканирования открытых портов на устройствах.
|
||||
|
||||
Предоставляет функциональность для проверки доступности портов, пока не реализована.
|
||||
"""
|
||||
def __init__(self, interface: str, translations: Translations):
|
||||
"""Инициализирует сканер портов.
|
||||
|
||||
Устанавливает интерфейс для сканирования и объект переводов.
|
||||
|
||||
Args:
|
||||
interface (str): Имя сетевого интерфейса.
|
||||
translations (Translations): Объект переводов для текущего языка.
|
||||
"""
|
||||
self.interface = interface
|
||||
self.translations = translations
|
||||
|
||||
def scan(self, ip: str) -> List[str]:
|
||||
"""Сканирует открытые порты на указанном IP-адресе.
|
||||
|
||||
Пока возвращает пустой список, так как функциональность не реализована.
|
||||
|
||||
Args:
|
||||
ip (str): IP-адрес устройства для сканирования.
|
||||
|
||||
Returns:
|
||||
List[str]: Список открытых портов (пока пустой).
|
||||
"""
|
||||
return []
|
||||
|
||||
class DeviceInfoCollector:
|
||||
"""Класс для сбора информации об устройствах, включая вендоров и хостнеймы."""
|
||||
def __init__(self, translations: Translations = Depends(get_translations())):
|
||||
"""Класс для сбора информации об устройствах, включая вендоров и хостнеймы.
|
||||
|
||||
Использует внешние сервисы (MacLookup, NBTSCAN, DNS) для получения данных.
|
||||
"""
|
||||
def __init__(self, translations: Translations):
|
||||
"""Инициализирует коллектор информации об устройствах.
|
||||
|
||||
Настраивает кэширование и объект переводов.
|
||||
|
||||
Args:
|
||||
translations (Translations): Объект переводов для текущего языка.
|
||||
"""
|
||||
self.mac_lookup = MacLookup()
|
||||
self.nbtscan_cache = {}
|
||||
self.nbtscan_results = {}
|
||||
|
|
@ -64,6 +134,16 @@ class DeviceInfoCollector:
|
|||
|
||||
@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:
|
||||
|
|
@ -71,6 +151,13 @@ class DeviceInfoCollector:
|
|||
return self.translations.gettext("Неизвестный")
|
||||
|
||||
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}")
|
||||
|
|
@ -97,6 +184,16 @@ class DeviceInfoCollector:
|
|||
logger.error(f"NetBIOS: Error for subnet {subnet}: {str(e)}")
|
||||
|
||||
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]
|
||||
|
|
@ -114,20 +211,56 @@ class DeviceInfoCollector:
|
|||
return None
|
||||
|
||||
class DeviceStorage:
|
||||
"""Класс для хранения и управления данными об устройствах."""
|
||||
def __init__(self, storage_path: str = "storage/devices/devices.json", translations: Translations = Depends(get_translations())):
|
||||
self.storage_file = Path(storage_path)
|
||||
self.storage_file.parent.mkdir(parents=True, exist_ok=True)
|
||||
self.offline_timeout = timedelta(days=7)
|
||||
self.translations = translations
|
||||
"""Класс для хранения и управления данными об устройствах.
|
||||
|
||||
def load_devices(self) -> Dict[str, Device]:
|
||||
Обеспечивает загрузку и сохранение данных в JSON-файлы.
|
||||
"""
|
||||
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 на русский
|
||||
|
||||
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"
|
||||
|
||||
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 self.storage_file.exists():
|
||||
with open(self.storage_file, "r", encoding="utf-8") as f:
|
||||
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 {self.storage_file}, expected {'devices': [...]}, got {data}")
|
||||
logger.warning(f"Invalid JSON structure in {file_path}, expected {'devices': [...]}, got {data}")
|
||||
return {}
|
||||
devices = {}
|
||||
for device_data in data.get("devices", []):
|
||||
|
|
@ -137,13 +270,23 @@ class DeviceStorage:
|
|||
return self._clean_old_devices(devices)
|
||||
return {}
|
||||
except json.JSONDecodeError as e:
|
||||
logger.error(f"JSON decode error in {self.storage_file}: {str(e)}")
|
||||
logger.error(f"JSON decode error in {file_path}: {str(e)}")
|
||||
return {}
|
||||
except Exception as e:
|
||||
logger.error(f"Error loading devices: {str(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():
|
||||
|
|
@ -160,28 +303,63 @@ class DeviceStorage:
|
|||
filtered[mac] = device
|
||||
return filtered
|
||||
|
||||
def save_devices(self, devices: List[Device]) -> None:
|
||||
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(self.storage_file, "w", encoding="utf-8") as f:
|
||||
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 {self.storage_file}")
|
||||
logger.info(f"Saved {len(devices)} devices to {file_path}")
|
||||
except Exception as e:
|
||||
logger.error(f"Error saving devices: {str(e)}")
|
||||
raise Exception(self.translations.gettext("Ошибка при сохранении устройств: {error}").format(error=str(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)}")
|
||||
|
||||
class SubnetService:
|
||||
"""Сервис для сканирования подсетей и управления устройствами."""
|
||||
def __init__(self, translations: Translations = Depends(get_translations())):
|
||||
"""Сервис для сканирования подсетей и управления устройствами.
|
||||
|
||||
Объединяет функциональность сканирования, сбора информации и хранения данных.
|
||||
"""
|
||||
def __init__(self, translations: Translations):
|
||||
"""Инициализирует сервис подсети.
|
||||
|
||||
Настраивает интерфейс, сканеры и хранилище с текущей локализацией.
|
||||
|
||||
Args:
|
||||
translations (Translations): Объект переводов для текущего языка.
|
||||
"""
|
||||
self.interface = self._get_default_interface()
|
||||
self.network_scanner = NetworkScanner(self.interface)
|
||||
self.port_scanner = PortScanner(self.interface)
|
||||
self.device_info = DeviceInfoCollector()
|
||||
self.storage = DeviceStorage()
|
||||
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)
|
||||
|
|
@ -203,7 +381,16 @@ class SubnetService:
|
|||
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:
|
||||
|
|
@ -228,13 +415,18 @@ class SubnetService:
|
|||
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 = []
|
||||
|
|
@ -246,9 +438,9 @@ class SubnetService:
|
|||
subnet = self._get_default_subnet()
|
||||
if subnet == "0.0.0.0/0":
|
||||
logger.warning("Default subnet is invalid, scan aborted")
|
||||
return SubnetScanResult(devices=devices)
|
||||
return SubnetScanResult(devices=devices, error=self.translations.gettext("Недопустимая подсеть по умолчанию"))
|
||||
|
||||
saved_devices = self.storage.load_devices()
|
||||
saved_devices = self.storage.load_devices(subnet)
|
||||
|
||||
local_device = self._get_local_device(subnet)
|
||||
if local_device:
|
||||
|
|
@ -264,6 +456,10 @@ class SubnetService:
|
|||
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:
|
||||
|
|
@ -309,7 +505,7 @@ class SubnetService:
|
|||
for device in devices:
|
||||
if device.mac:
|
||||
saved_devices[device.mac] = device
|
||||
self.storage.save_devices(list(saved_devices.values()))
|
||||
self.storage.save_devices(subnet, list(saved_devices.values()))
|
||||
|
||||
logger.info(
|
||||
f"Subnet scan {subnet} completed in {time.time() - start_time:.2f} seconds, "
|
||||
|
|
@ -319,9 +515,19 @@ class SubnetService:
|
|||
|
||||
except Exception as e:
|
||||
logger.error(f"Error scanning subnet {subnet}: {str(e)}")
|
||||
return SubnetScanResult(devices=devices)
|
||||
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:
|
||||
|
|
@ -354,53 +560,115 @@ class SubnetService:
|
|||
logger.warning(f"Failed to get local device: {str(e)}")
|
||||
return None
|
||||
|
||||
def add_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]:
|
||||
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 = self.storage.load_devices()
|
||||
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, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac)
|
||||
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, self.translations.gettext("Заметка для устройства с MAC {mac} уже существует").format(mac=mac)
|
||||
saved_devices[mac].note = note
|
||||
self.storage.save_devices(list(saved_devices.values()))
|
||||
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, None
|
||||
return True, ("success", {"mac": mac, "note": note})
|
||||
except Exception as e:
|
||||
logger.error(f"Error adding note for MAC {mac}: {str(e)}")
|
||||
return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e))
|
||||
return False, ("internal_error", {"error": str(e)})
|
||||
|
||||
def update_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]:
|
||||
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 = self.storage.load_devices()
|
||||
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, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac)
|
||||
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, self.translations.gettext("Заметка для устройства с MAC {mac} не существует").format(mac=mac)
|
||||
saved_devices[mac].note = note
|
||||
self.storage.save_devices(list(saved_devices.values()))
|
||||
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, None
|
||||
return True, ("success", {"mac": mac, "note": note})
|
||||
except Exception as e:
|
||||
logger.error(f"Error updating note for MAC {mac}: {str(e)}")
|
||||
return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e))
|
||||
return False, ("internal_error", {"error": str(e)})
|
||||
|
||||
def delete_note(self, mac: str) -> Tuple[bool, Optional[str]]:
|
||||
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 = self.storage.load_devices()
|
||||
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, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac)
|
||||
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, self.translations.gettext("Заметка для устройства с MAC {mac} не существует").format(mac=mac)
|
||||
saved_devices[mac].note = None
|
||||
self.storage.save_devices(list(saved_devices.values()))
|
||||
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, None
|
||||
return True, ("success", {"mac": mac})
|
||||
except Exception as e:
|
||||
logger.error(f"Error deleting note for MAC {mac}: {str(e)}")
|
||||
return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e))
|
||||
return False, ("internal_error", {"error": str(e)})
|
||||
Loading…
Add table
Add a link
Reference in a new issue