Интернационализация для всех сервисов сервера
This commit is contained in:
parent
7491ce455d
commit
ba8cea01e7
13 changed files with 1011 additions and 206 deletions
|
|
@ -12,6 +12,9 @@ import json
|
|||
from pathlib import Path
|
||||
from datetime import datetime, timedelta
|
||||
import subprocess
|
||||
from utils.i18n import get_translations, _
|
||||
from babel.support import Translations
|
||||
from fastapi import Depends
|
||||
|
||||
from models.subnet import SubnetScanResult, Device
|
||||
from config.ports import port_config
|
||||
|
|
@ -30,7 +33,7 @@ class NetworkScanner:
|
|||
Выполняет обнаружение устройств в указанной подсети через ARP-пакеты.
|
||||
"""
|
||||
|
||||
def __init__(self, interface: str):
|
||||
def __init__(self, interface: str, translations: Translations = Depends(get_translations("services"))):
|
||||
"""Инициализирует сканер сети с указанным сетевым интерфейсом.
|
||||
|
||||
Args:
|
||||
|
|
@ -38,6 +41,7 @@ class NetworkScanner:
|
|||
"""
|
||||
self.interface = interface
|
||||
conf.iface = interface
|
||||
self.translations = translations
|
||||
|
||||
def scan(self, subnet: str) -> List[Tuple[str, str]]:
|
||||
"""Сканирует подсеть с использованием ARP-запросов.
|
||||
|
|
@ -63,7 +67,7 @@ class NetworkScanner:
|
|||
return [(received.psrc, received.hwsrc)
|
||||
for sent, received in result]
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка ARP-сканирования подсети {subnet}: {str(e)}")
|
||||
logger.error(f"Error scanning subnet {subnet}: {str(e)}")
|
||||
return []
|
||||
|
||||
class PortScanner:
|
||||
|
|
@ -72,13 +76,14 @@ class PortScanner:
|
|||
Выполняет сканирование портов с использованием TCP SYN-запросов.
|
||||
"""
|
||||
|
||||
def __init__(self, interface: str):
|
||||
def __init__(self, interface: str, translations: Translations = Depends(get_translations("services"))):
|
||||
"""Инициализирует сканер портов с указанным сетевым интерфейсом.
|
||||
|
||||
Args:
|
||||
interface (str): Название сетевого интерфейса (например, "eth0").
|
||||
"""
|
||||
self.interface = interface
|
||||
self.translations = translations
|
||||
|
||||
def scan(self, ip: str) -> List[str]:
|
||||
"""Сканирует открытые порты на указанном IP-адресе.
|
||||
|
|
@ -114,7 +119,7 @@ class PortScanner:
|
|||
return None
|
||||
except Exception as e:
|
||||
logger.debug(
|
||||
f"Ошибка проверки порта {port} для IP {ip}: {str(e)}")
|
||||
f"Error checking port {port} for IP {ip}: {str(e)}")
|
||||
return None
|
||||
|
||||
try:
|
||||
|
|
@ -122,11 +127,11 @@ class PortScanner:
|
|||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||||
results = executor.map(check_port, port_config.get_ports())
|
||||
open_ports = [port for port in results if port is not None]
|
||||
logger.debug(f"Открытые порты для IP {ip}: {open_ports}")
|
||||
logger.debug(f"Open ports for IP {ip}: {open_ports}")
|
||||
return open_ports
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Ошибка при сканировании портов для IP {ip}: {str(e)}")
|
||||
f"Error scanning ports for IP {ip}: {str(e)}")
|
||||
return []
|
||||
|
||||
class DeviceInfoCollector:
|
||||
|
|
@ -135,7 +140,7 @@ class DeviceInfoCollector:
|
|||
Управляет кэшированием данных NetBIOS и DNS.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self, translations: Translations = Depends(get_translations("services"))):
|
||||
"""Инициализирует коллектор информации об устройствах.
|
||||
|
||||
Создаёт объект MacLookup для определения вендоров и инициализирует кэши.
|
||||
|
|
@ -143,6 +148,7 @@ class DeviceInfoCollector:
|
|||
self.mac_lookup = MacLookup()
|
||||
self.nbtscan_cache = {}
|
||||
self.nbtscan_results = {}
|
||||
self.translations = translations
|
||||
|
||||
@lru_cache(maxsize=1000)
|
||||
def get_vendor(self, mac: str) -> str:
|
||||
|
|
@ -161,8 +167,8 @@ class DeviceInfoCollector:
|
|||
return self.mac_lookup.lookup(mac)
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Не удалось определить производителя для MAC {mac}: {str(e)}")
|
||||
return "Unknown"
|
||||
f"Failed to determine vendor for MAC {mac}: {str(e)}")
|
||||
return self.translations.gettext("Неизвестный")
|
||||
|
||||
def scan_nbtscan(self, subnet: str) -> None:
|
||||
"""Выполняет сканирование NetBIOS для указанной подсети.
|
||||
|
|
@ -174,28 +180,28 @@ class DeviceInfoCollector:
|
|||
"""
|
||||
if subnet not in self.nbtscan_results:
|
||||
try:
|
||||
logger.debug(f"Запуск nbtscan для подсети {subnet}")
|
||||
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"Полный вывод nbtscan: {result.stdout}")
|
||||
logger.debug(f"Full nbtscan output: {result.stdout}")
|
||||
if result.returncode == 0:
|
||||
for line in result.stdout.splitlines():
|
||||
logger.debug(f"Обработка строки: {line}")
|
||||
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: Найден хостнейм {hostname} для IP {ip_addr}")
|
||||
logger.info(f"NetBIOS: Found hostname {hostname} for IP {ip_addr}")
|
||||
self.nbtscan_results[subnet] = True
|
||||
except subprocess.TimeoutExpired:
|
||||
logger.error(f"NetBIOS: Таймаут для подсети {subnet}")
|
||||
logger.error(f"NetBIOS: Timeout for subnet {subnet}")
|
||||
except Exception as e:
|
||||
logger.error(f"NetBIOS: Ошибка для подсети {subnet}: {str(e)}")
|
||||
logger.error(f"NetBIOS: Error for subnet {subnet}: {str(e)}")
|
||||
|
||||
def get_hostname(self, ip: str) -> Optional[str]:
|
||||
"""Определяет хостнейм для указанного IP-адреса.
|
||||
|
|
@ -212,7 +218,7 @@ class DeviceInfoCollector:
|
|||
socket.setdefaulttimeout(2)
|
||||
return socket.gethostbyaddr(ip)[0]
|
||||
except (socket.herror, socket.gaierror, socket.timeout):
|
||||
logger.debug(f"DNS: Не удалось разрешить имя для IP {ip}")
|
||||
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:
|
||||
|
|
@ -221,7 +227,7 @@ class DeviceInfoCollector:
|
|||
if ip in self.nbtscan_cache:
|
||||
return self.nbtscan_cache[ip]
|
||||
|
||||
logger.debug(f"Не удалось определить хостнейм для IP {ip} через DNS или NetBIOS")
|
||||
logger.debug(f"Failed to determine hostname for IP {ip} via DNS or NetBIOS")
|
||||
return None
|
||||
|
||||
class DeviceStorage:
|
||||
|
|
@ -230,7 +236,7 @@ class DeviceStorage:
|
|||
Хранит устройства в JSON-файле и удаляет устаревшие записи.
|
||||
"""
|
||||
|
||||
def __init__(self, storage_path: str = "storage/devices/devices.json"):
|
||||
def __init__(self, storage_path: str = "storage/devices/devices.json", translations: Translations = Depends(get_translations("services"))):
|
||||
"""Инициализирует хранилище устройств с указанным путём.
|
||||
|
||||
Args:
|
||||
|
|
@ -242,6 +248,7 @@ class DeviceStorage:
|
|||
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]:
|
||||
"""Загружает устройства из файла хранения.
|
||||
|
|
@ -264,7 +271,7 @@ class DeviceStorage:
|
|||
return self._clean_old_devices(devices)
|
||||
return {}
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка при загрузке устройств: {str(e)}")
|
||||
logger.error(f"Error loading devices: {str(e)}")
|
||||
return {}
|
||||
|
||||
def _clean_old_devices(self, devices: Dict[str, Device]) -> Dict[str, Device]:
|
||||
|
|
@ -286,9 +293,9 @@ class DeviceStorage:
|
|||
if now - last_seen <= self.offline_timeout:
|
||||
filtered[mac] = device
|
||||
else:
|
||||
logger.info(f"Удалено устройство {mac} (offline > 7 дней)")
|
||||
logger.info(f"Removed device {mac} (offline > 7 days)")
|
||||
except ValueError:
|
||||
logger.warning(f"Неверный формат last_seen для устройства {mac}")
|
||||
logger.warning(f"Invalid last_seen format for device {mac}")
|
||||
else:
|
||||
filtered[mac] = device
|
||||
return filtered
|
||||
|
|
@ -306,9 +313,10 @@ class DeviceStorage:
|
|||
data = SubnetScanResult(devices=devices).model_dump()
|
||||
with open(self.storage_file, "w", encoding="utf-8") as f:
|
||||
json.dump(data, f, indent=2, ensure_ascii=False)
|
||||
logger.info(f"Сохранено {len(devices)} устройств в {self.storage_file}")
|
||||
logger.info(f"Saved {len(devices)} devices to {self.storage_file}")
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка при сохранении устройств: {str(e)}")
|
||||
logger.error(f"Error saving devices: {str(e)}")
|
||||
raise Exception(self.translations.gettext("Ошибка при сохранении устройств: {error}").format(error=str(e)))
|
||||
|
||||
class SubnetService:
|
||||
"""Сервис для сканирования подсетей и управления устройствами.
|
||||
|
|
@ -316,7 +324,7 @@ class SubnetService:
|
|||
Объединяет функционал сканирования сети, портов и хранения данных об устройствах.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self, translations: Translations = Depends(get_translations("services"))):
|
||||
"""Инициализирует сервис подсети с настройкой всех компонентов.
|
||||
|
||||
Автоматически определяет интерфейс и инициализирует сканеры и хранилище.
|
||||
|
|
@ -326,6 +334,7 @@ class SubnetService:
|
|||
self.port_scanner = PortScanner(self.interface)
|
||||
self.device_info = DeviceInfoCollector()
|
||||
self.storage = DeviceStorage()
|
||||
self.translations = translations
|
||||
|
||||
def _get_default_interface(self) -> str:
|
||||
"""Определяет имя сетевого интерфейса по умолчанию.
|
||||
|
|
@ -346,7 +355,7 @@ class SubnetService:
|
|||
return iface
|
||||
return "eth0"
|
||||
except Exception as e:
|
||||
logger.warning(f"Не удалось определить интерфейс: {str(e)}")
|
||||
logger.warning(f"Failed to determine interface: {str(e)}")
|
||||
return "eth0"
|
||||
|
||||
def scan_subnet(self, subnet: str, ports: List[int] = None) -> SubnetScanResult:
|
||||
|
|
@ -438,13 +447,13 @@ class SubnetService:
|
|||
self.storage.save_devices(list(saved_devices.values()))
|
||||
|
||||
logger.info(
|
||||
f"Сканирование подсети {subnet} завершено за {time.time() - start_time:.2f} секунд, "
|
||||
f"найдено {len(devices)} устройств (активных: {len(active_macs)})"
|
||||
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"Ошибка при сканировании подсети {subnet}: {str(e)}")
|
||||
logger.error(f"Error scanning subnet {subnet}: {str(e)}")
|
||||
return SubnetScanResult(devices=devices)
|
||||
|
||||
def _get_local_device(self, subnet: str) -> Optional[Device]:
|
||||
|
|
@ -473,7 +482,7 @@ class SubnetService:
|
|||
hostname = self.device_info.get_hostname(ip)
|
||||
open_ports = self.port_scanner.scan(ip)
|
||||
logger.debug(
|
||||
f"Локальное устройство: IP={ip}, MAC={mac}, Ports={open_ports}, Hostname={hostname}"
|
||||
f"Local device: IP={ip}, MAC={mac}, Ports={open_ports}, Hostname={hostname}"
|
||||
)
|
||||
return Device(
|
||||
ip=ip,
|
||||
|
|
@ -486,11 +495,11 @@ class SubnetService:
|
|||
note=None,
|
||||
)
|
||||
logger.warning(
|
||||
f"Локальное устройство не найдено в подсети {subnet}")
|
||||
f"Local device not found in subnet {subnet}")
|
||||
return None
|
||||
except Exception as e:
|
||||
logger.warning(
|
||||
f"Не удалось получить локальное устройство: {str(e)}")
|
||||
f"Failed to get local device: {str(e)}")
|
||||
return None
|
||||
|
||||
def add_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]:
|
||||
|
|
@ -511,21 +520,21 @@ class SubnetService:
|
|||
try:
|
||||
saved_devices = self.storage.load_devices()
|
||||
if mac not in saved_devices:
|
||||
logger.warning(f"Устройство с MAC {mac} не найдено")
|
||||
return False, f"Устройство с MAC {mac} не найдено"
|
||||
logger.warning(f"Device with MAC {mac} not found")
|
||||
return False, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac)
|
||||
if saved_devices[mac].note is not None:
|
||||
logger.error(
|
||||
f"Заметка для устройства с MAC {mac} уже существует")
|
||||
return False, f"Заметка для устройства с MAC {mac} уже существует"
|
||||
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()))
|
||||
logger.info(
|
||||
f"Добавлена заметка '{note}' для устройства с MAC {mac}")
|
||||
f"Added note '{note}' for device with MAC {mac}")
|
||||
return True, None
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Ошибка при добавлении заметки для MAC {mac}: {str(e)}")
|
||||
return False, f"Внутренняя ошибка сервера: {str(e)}"
|
||||
f"Error adding note for MAC {mac}: {str(e)}")
|
||||
return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e))
|
||||
|
||||
def update_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]:
|
||||
"""Обновляет заметку устройства по MAC-адресу.
|
||||
|
|
@ -545,21 +554,21 @@ class SubnetService:
|
|||
try:
|
||||
saved_devices = self.storage.load_devices()
|
||||
if mac not in saved_devices:
|
||||
logger.warning(f"Устройство с MAC {mac} не найдено")
|
||||
return False, f"Устройство с MAC {mac} не найдено"
|
||||
logger.warning(f"Device with MAC {mac} not found")
|
||||
return False, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac)
|
||||
if saved_devices[mac].note is None:
|
||||
logger.error(
|
||||
f"Заметка для устройства с MAC {mac} не существует")
|
||||
return False, f"Заметка для устройства с MAC {mac} не существует"
|
||||
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()))
|
||||
logger.info(
|
||||
f"Обновлена заметка '{note}' для устройства с MAC {mac}")
|
||||
f"Updated note '{note}' for device with MAC {mac}")
|
||||
return True, None
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Ошибка при обновлении заметки для MAC {mac}: {str(e)}")
|
||||
return False, f"Внутренняя ошибка сервера: {str(e)}"
|
||||
f"Error updating note for MAC {mac}: {str(e)}")
|
||||
return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e))
|
||||
|
||||
def delete_note(self, mac: str) -> Tuple[bool, Optional[str]]:
|
||||
"""Удаляет заметку устройства по MAC-адресу.
|
||||
|
|
@ -578,17 +587,17 @@ class SubnetService:
|
|||
try:
|
||||
saved_devices = self.storage.load_devices()
|
||||
if mac not in saved_devices:
|
||||
logger.warning(f"Устройство с MAC {mac} не найдено")
|
||||
return False, f"Устройство с MAC {mac} не найдено"
|
||||
logger.warning(f"Device with MAC {mac} not found")
|
||||
return False, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac)
|
||||
if saved_devices[mac].note is None:
|
||||
logger.error(
|
||||
f"Заметка для устройства с MAC {mac} не существует")
|
||||
return False, f"Заметка для устройства с MAC {mac} не существует"
|
||||
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()))
|
||||
logger.info(f"Удалена заметка для устройства с MAC {mac}")
|
||||
logger.info(f"Deleted note for device with MAC {mac}")
|
||||
return True, None
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Ошибка при удалении заметки для MAC {mac}: {str(e)}")
|
||||
return False, f"Внутренняя ошибка сервера: {str(e)}"
|
||||
f"Error deleting note for MAC {mac}: {str(e)}")
|
||||
return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e))
|
||||
Loading…
Add table
Add a link
Reference in a new issue