HybrydMain/gonets
3
2
Fork 0
Code Issues 5 Pull requests Projects 1 Releases 3 Packages Wiki Activity Actions

Изменение сканирования подсети и точек доступа wi-fi в связи с последними изменениями на сервере

Browse source
This commit is contained in:
Галанский Сергей Александрович 2025-07-15 13:07:14 +05:00
parent 990a7594fc
commit f0d5481dfe
4 changed files with 188 additions and 501 deletions
Download patch file Download diff file Expand all files Collapse all files

17
SRV/api/v1/subnet.py
View file

@ -7,18 +7,22 @@ from services.wifi_service import WifiService
from services.connect_service import WifiConnectionService from services.connect_service import WifiConnectionService
from utils.i18n import get_translations, _ from utils.i18n import get_translations, _
from babel.support import Translations from babel.support import Translations
import logging
router = APIRouter() router = APIRouter()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
@router.post("/devices", response_model=SubnetScanResult) @router.post("/devices", response_model=SubnetScanResult)
def scan_subnet(request: SubnetScanRequest = Body(...), translations: Translations = Depends(get_translations("api_v1"))): def scan_subnet(request: SubnetScanRequest = Body(...), translations: Translations = Depends(get_translations("api_v1"))):
"""Сканирует локальную подсеть и возвращает данные об устройствах. """Сканирует локальную подсеть и возвращает данные об устройствах.
Выполняет сканирование указанной подсети для обнаружения активных устройств Выполняет сканирование указанной подсети для обнаружения активных устройств.
с возможностью проверки указанных портов.
Args: Args:
request (SubnetScanRequest): Объект с полями subnet и ports. request (SubnetScanRequest): Объект с полем subnet.
translations: Объект переводов для текущего языка. translations: Объект переводов для текущего языка.
Returns: Returns:
@ -29,10 +33,15 @@ def scan_subnet(request: SubnetScanRequest = Body(...), translations: Translatio
""" """
try: try:
subnet_service = SubnetService() subnet_service = SubnetService()
subnet = request.subnet or "192.168.1.0/24" # Если subnet пустая строка, определяем автоматически
subnet = request.subnet.strip() if request.subnet else None
logger.info(f"Requested subnet scan for: {subnet or 'auto-detected'}")
result = subnet_service.scan_subnet(subnet) result = subnet_service.scan_subnet(subnet)
if not result.devices:
logger.warning(f"No devices found in subnet {subnet or 'auto-detected'}")
return result return result
except Exception as e: except Exception as e:
logger.error(f"Error in scan_subnet: {str(e)}")
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка сканирования: {error}").format(error=str(e))) raise HTTPException(status_code=500, detail=translations.gettext("Ошибка сканирования: {error}").format(error=str(e)))
@router.get("/wifi", response_model=WifiScanResult) @router.get("/wifi", response_model=WifiScanResult)

5
SRV/models/subnet.py
View file

@ -37,11 +37,12 @@ class SubnetScanResult(BaseModel):
devices: List[Device] = [] devices: List[Device] = []
class SubnetScanRequest(BaseModel): class SubnetScanRequest(BaseModel):
""" """
Параметры для сканирования подсети Параметры для сканирования подсети
Атрибуты: Атрибуты:
subnet (str): Подсеть для сканирования (например 192.168.1.0/24) subnet (Optional[str]): Подсеть для сканирования (например 192.168.1.0/24), опционально
""" """
subnet: Optional[str] = "192.168.1.0/24" subnet: Optional[str] = None

367
SRV/services/subnet_service.py
View file

@ -12,172 +12,69 @@ import json
from pathlib import Path from pathlib import Path
from datetime import datetime, timedelta from datetime import datetime, timedelta
import subprocess import subprocess
from utils.i18n import get_translations, _
from scapy.layers.l2 import ARP, Ether
from scapy.sendrecv import srp
from utils.i18n import get_translations
from babel.support import Translations from babel.support import Translations
from fastapi import Depends from fastapi import Depends
from models.subnet import SubnetScanResult, Device from models.subnet import SubnetScanResult, Device
from config.ports import port_config
from scapy.layers.inet import IP, TCP
from scapy.layers.l2 import ARP, Ether
from scapy.sendrecv import srp, sr1
logging.basicConfig( logging.basicConfig(
level=logging.INFO, level=logging.INFO,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class NetworkScanner: class NetworkScanner:
"""Класс для сканирования подсети с использованием ARP-запросов. """Класс для сканирования подсети с использованием ARP-запросов."""
Выполняет обнаружение устройств в указанной подсети через ARP-пакеты.
"""
def __init__(self, interface: str, translations: Translations = Depends(get_translations("services"))): def __init__(self, interface: str, translations: Translations = Depends(get_translations("services"))):
"""Инициализирует сканер сети с указанным сетевым интерфейсом.
Args:
interface (str): Название сетевого интерфейса (например, "eth0").
"""
self.interface = interface self.interface = interface
conf.iface = interface conf.iface = interface
self.translations = translations self.translations = translations
def scan(self, subnet: str) -> List[Tuple[str, str]]: def scan(self, subnet: str) -> List[Tuple[str, str]]:
"""Сканирует подсеть с использованием ARP-запросов.
Args:
subnet (str): Подсеть в формате CIDR (например, "192.168.1.0/24").
Returns:
List[Tuple[str, str]]: Список кортежей (IP-адрес, MAC-адрес) обнаруженных
устройств.
Raises:
Exception: Если произошла ошибка при отправке ARP-пакетов.
"""
try: try:
arp = ARP(pdst=subnet) arp = ARP(pdst=subnet)
ether = Ether(dst="ff:ff:ff:ff:ff:ff") ether = Ether(dst="ff:ff:ff:ff:ff:ff")
packet = ether / arp packet = ether / arp
result = srp(packet, result = srp(packet, timeout=2, verbose=False, iface=self.interface)[0]
timeout=2, return [(received.psrc, received.hwsrc) for sent, received in result]
verbose=False,
iface=self.interface)[0]
return [(received.psrc, received.hwsrc)
for sent, received in result]
except Exception as e: except Exception as e:
logger.error(f"Error scanning subnet {subnet}: {str(e)}") logger.error(f"Error scanning subnet {subnet}: {str(e)}")
return [] return []
class PortScanner: class PortScanner:
"""Класс для сканирования открытых портов на устройствах. """Класс для сканирования открытых портов на устройствах."""
Выполняет сканирование портов с использованием TCP SYN-запросов.
"""
def __init__(self, interface: str, translations: Translations = Depends(get_translations("services"))): def __init__(self, interface: str, translations: Translations = Depends(get_translations("services"))):
"""Инициализирует сканер портов с указанным сетевым интерфейсом.
Args:
interface (str): Название сетевого интерфейса (например, "eth0").
"""
self.interface = interface self.interface = interface
self.translations = translations self.translations = translations
def scan(self, ip: str) -> List[str]: def scan(self, ip: str) -> List[str]:
"""Сканирует открытые порты на указанном IP-адресе.
Args:
ip (str): IP-адрес устройства для сканирования.
Returns:
List[str]: Список имён открытых портов (например, "HTTP", "SSH").
Raises:
Exception: Если произошла ошибка при сканировании портов.
"""
open_ports = []
def check_port(port: int) -> Optional[str]:
"""Проверяет доступность конкретного порта.
Args:
port (int): Номер порта для проверки.
Returns:
Optional[str]: Имя порта, если открыт, иначе None.
"""
try:
pkt = sr1(
IP(dst=ip) / TCP(dport=port, flags="S"),
timeout=2,
verbose=False,
)
if pkt and pkt.haslayer(TCP) and pkt[TCP].flags == 0x12:
return port_config.get_port_name(port)
return None
except Exception as e:
logger.debug(
f"Error checking port {port} for IP {ip}: {str(e)}")
return None
try:
max_workers = min(len(port_config.get_ports()), 15)
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"Open ports for IP {ip}: {open_ports}")
return open_ports
except Exception as e:
logger.warning(
f"Error scanning ports for IP {ip}: {str(e)}")
return [] return []
class DeviceInfoCollector: class DeviceInfoCollector:
"""Класс для сбора информации об устройствах, включая вендоров и хостнеймы. """Класс для сбора информации об устройствах, включая вендоров и хостнеймы."""
Управляет кэшированием данных NetBIOS и DNS.
"""
def __init__(self, translations: Translations = Depends(get_translations("services"))): def __init__(self, translations: Translations = Depends(get_translations("services"))):
"""Инициализирует коллектор информации об устройствах.
Создаёт объект MacLookup для определения вендоров и инициализирует кэши.
"""
self.mac_lookup = MacLookup() self.mac_lookup = MacLookup()
self.nbtscan_cache = {} self.nbtscan_cache = {}
self.nbtscan_results = {} self.nbtscan_results = {}
if not hasattr(translations, 'gettext'):
logger.warning("Translations object does not support gettext, using fallback")
self.translations = Translations.load("utils/locales/en/LC_MESSAGES/services.mo")
else:
self.translations = translations self.translations = translations
@lru_cache(maxsize=1000) @lru_cache(maxsize=1000)
def get_vendor(self, mac: str) -> str: def get_vendor(self, mac: str) -> str:
"""Определяет вендора по MAC-адресу с использованием кэширования.
Args:
mac (str): MAC-адрес устройства в формате XX:XX:XX:XX:XX:XX.
Returns:
str: Название вендора или "Unknown", если вендор не определён.
Raises:
Exception: Если произошла ошибка при запросе к базе вендоров.
"""
try: try:
return self.mac_lookup.lookup(mac) return self.mac_lookup.lookup(mac)
except Exception as e: except Exception as e:
logger.warning( logger.warning(f"Failed to determine vendor for MAC {mac}: {str(e)}")
f"Failed to determine vendor for MAC {mac}: {str(e)}")
return self.translations.gettext("Неизвестный") return self.translations.gettext("Неизвестный")
def scan_nbtscan(self, subnet: str) -> None: def scan_nbtscan(self, subnet: str) -> None:
"""Выполняет сканирование NetBIOS для указанной подсети.
Args:
subnet (str): Подсеть в формате CIDR (например, "192.168.1.0/24").
Сохраняет результаты в кэш, игнорируя хостнеймы "<unknown>".
"""
if subnet not in self.nbtscan_results: if subnet not in self.nbtscan_results:
try: try:
logger.debug(f"Starting nbtscan for subnet {subnet}") logger.debug(f"Starting nbtscan for subnet {subnet}")
@ -204,16 +101,6 @@ class DeviceInfoCollector:
logger.error(f"NetBIOS: Error for subnet {subnet}: {str(e)}") logger.error(f"NetBIOS: Error for subnet {subnet}: {str(e)}")
def get_hostname(self, ip: str) -> Optional[str]: def get_hostname(self, ip: str) -> Optional[str]:
"""Определяет хостнейм для указанного IP-адреса.
Args:
ip (str): IP-адрес устройства.
Returns:
Optional[str]: Хостнейм, если найден, иначе None.
Сначала пытается использовать DNS, затем NetBIOS через nbtscan.
"""
try: try:
socket.setdefaulttimeout(2) socket.setdefaulttimeout(2)
return socket.gethostbyaddr(ip)[0] return socket.gethostbyaddr(ip)[0]
@ -231,38 +118,25 @@ class DeviceInfoCollector:
return None return None
class DeviceStorage: class DeviceStorage:
"""Класс для хранения и управления данными об устройствах. """Класс для хранения и управления данными об устройствах."""
Хранит устройства в JSON-файле и удаляет устаревшие записи.
"""
def __init__(self, storage_path: str = "storage/devices/devices.json", translations: Translations = Depends(get_translations("services"))): def __init__(self, storage_path: str = "storage/devices/devices.json", translations: Translations = Depends(get_translations("services"))):
"""Инициализирует хранилище устройств с указанным путём.
Args:
storage_path (str, optional): Путь к файлу хранения.
По умолчанию "storage/devices/devices.json".
Создаёт родительские директории, если они не существуют.
"""
self.storage_file = Path(storage_path) self.storage_file = Path(storage_path)
self.storage_file.parent.mkdir(parents=True, exist_ok=True) self.storage_file.parent.mkdir(parents=True, exist_ok=True)
self.offline_timeout = timedelta(days=7) self.offline_timeout = timedelta(days=7)
if not hasattr(translations, 'gettext'):
logger.warning("Translations object does not support gettext, using fallback")
self.translations = Translations.load("utils/locales/en/LC_MESSAGES/services.mo")
else:
self.translations = translations self.translations = translations
def load_devices(self) -> Dict[str, Device]: def load_devices(self) -> Dict[str, Device]:
"""Загружает устройства из файла хранения.
Returns:
Dict[str, Device]: Словарь устройств, где ключ — MAC-адрес.
Удаляет устройства с неверным форматом last_seen и возвращает пустой словарь
при ошибке.
"""
try: try:
if self.storage_file.exists(): if self.storage_file.exists():
with open(self.storage_file, "r", encoding="utf-8") as f: with open(self.storage_file, "r", encoding="utf-8") as f:
data = json.load(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}")
return {}
devices = {} devices = {}
for device_data in data.get("devices", []): for device_data in data.get("devices", []):
device = Device(**device_data) device = Device(**device_data)
@ -270,20 +144,14 @@ class DeviceStorage:
devices[device.mac] = device devices[device.mac] = device
return self._clean_old_devices(devices) return self._clean_old_devices(devices)
return {} return {}
except json.JSONDecodeError as e:
logger.error(f"JSON decode error in {self.storage_file}: {str(e)}")
return {}
except Exception as e: except Exception as e:
logger.error(f"Error loading devices: {str(e)}") logger.error(f"Error loading devices: {str(e)}")
return {} return {}
def _clean_old_devices(self, devices: Dict[str, Device]) -> Dict[str, Device]: def _clean_old_devices(self, devices: Dict[str, Device]) -> Dict[str, Device]:
"""Очищает устаревшие устройства из словаря.
Args:
devices (Dict[str, Device]): Словарь устройств с MAC-адресами как ключами.
Returns:
Dict[str, Device]: Отфильтрованный словарь, исключающий устройства
offline более 7 дней.
"""
now = datetime.now() now = datetime.now()
filtered = {} filtered = {}
for mac, device in devices.items(): for mac, device in devices.items():
@ -301,14 +169,6 @@ class DeviceStorage:
return filtered return filtered
def save_devices(self, devices: List[Device]) -> None: def save_devices(self, devices: List[Device]) -> None:
"""Сохраняет список устройств в файл.
Args:
devices (List[Device]): Список объектов Device для сохранения.
Raises:
Exception: Если произошла ошибка при записи в файл.
"""
try: try:
data = SubnetScanResult(devices=devices).model_dump() data = SubnetScanResult(devices=devices).model_dump()
with open(self.storage_file, "w", encoding="utf-8") as f: with open(self.storage_file, "w", encoding="utf-8") as f:
@ -319,64 +179,87 @@ class DeviceStorage:
raise Exception(self.translations.gettext("Ошибка при сохранении устройств: {error}").format(error=str(e))) raise Exception(self.translations.gettext("Ошибка при сохранении устройств: {error}").format(error=str(e)))
class SubnetService: class SubnetService:
"""Сервис для сканирования подсетей и управления устройствами. """Сервис для сканирования подсетей и управления устройствами."""
Объединяет функционал сканирования сети, портов и хранения данных об устройствах.
"""
def __init__(self, translations: Translations = Depends(get_translations("services"))): def __init__(self, translations: Translations = Depends(get_translations("services"))):
"""Инициализирует сервис подсети с настройкой всех компонентов.
Автоматически определяет интерфейс и инициализирует сканеры и хранилище.
"""
self.interface = self._get_default_interface() self.interface = self._get_default_interface()
self.network_scanner = NetworkScanner(self.interface) self.network_scanner = NetworkScanner(self.interface)
self.port_scanner = PortScanner(self.interface) self.port_scanner = PortScanner(self.interface)
self.device_info = DeviceInfoCollector() self.device_info = DeviceInfoCollector()
self.storage = DeviceStorage() self.storage = DeviceStorage()
if not hasattr(translations, 'gettext'):
logger.warning("Translations object does not support gettext, using fallback")
self.translations = Translations.load("utils/locales/en/LC_MESSAGES/services.mo")
else:
self.translations = translations self.translations = translations
logger.info(f"Initialized SubnetService with interface: {self.interface}")
def _get_default_interface(self) -> str: def _get_default_interface(self) -> str:
"""Определяет имя сетевого интерфейса по умолчанию.
Returns:
str: Название интерфейса (например, "eth0").
Использует информацию о шлюзе или первый доступный интерфейс, отличный от "lo".
"""
try: try:
gateways = netifaces.gateways() gateways = netifaces.gateways()
default_gateway = gateways.get("default", {}).get(netifaces.AF_INET) default_gateway = gateways.get("default", {}).get(netifaces.AF_INET)
if default_gateway: if default_gateway:
return default_gateway[1] iface = default_gateway[1]
logger.info(f"Default interface from gateway: {iface}")
return iface
interfaces = netifaces.interfaces() interfaces = netifaces.interfaces()
for iface in interfaces: for iface in interfaces:
if iface != "lo": if iface != "lo":
addrs = netifaces.ifaddresses(iface)
if netifaces.AF_INET in addrs:
logger.info(f"Selected interface: {iface}")
return iface return iface
logger.warning("No suitable interface found, defaulting to eth0")
return "eth0" return "eth0"
except Exception as e: except Exception as e:
logger.warning(f"Failed to determine interface: {str(e)}") logger.warning(f"Failed to determine interface: {str(e)}, defaulting to eth0")
return "eth0" return "eth0"
def scan_subnet(self, subnet: str, ports: List[int] = None) -> SubnetScanResult: def _get_default_subnet(self) -> str:
"""Автоматически определяет подсеть локального устройства."""
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"
def scan_subnet(self, subnet: Optional[str] = None) -> SubnetScanResult:
"""Сканирует указанную подсеть и возвращает результаты. """Сканирует указанную подсеть и возвращает результаты.
Args: Args:
subnet (str): Подсеть в формате CIDR (например, "192.168.1.0/24"). subnet (Optional[str]): Подсеть в формате CIDR (например, "10.1.1.0/24").
ports (List[int], optional): Список портов для сканирования. Если None или пустая строка, определяется автоматически.
По умолчанию None (используются порты из config.ports).
Returns: Returns:
SubnetScanResult: Объект с результатами сканирования, содержащий список SubnetScanResult: Объект с результатами сканирования, содержащий список
устройств. устройств.
Сохраняет данные в хранилище и обновляет статус устройств.
""" """
start_time = time.time() start_time = time.time()
devices = [] devices = []
now = datetime.now().astimezone() now = datetime.now().astimezone()
formatted_now = now.strftime("%d-%m-%Y, %H:%M") formatted_now = now.strftime("%d-%m-%Y, %H:%M")
try: 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)
saved_devices = self.storage.load_devices() saved_devices = self.storage.load_devices()
local_device = self._get_local_device(subnet) local_device = self._get_local_device(subnet)
@ -386,19 +269,14 @@ class SubnetService:
devices.append(local_device) devices.append(local_device)
if local_device.mac: if local_device.mac:
saved_devices[local_device.mac] = local_device 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) discovered_devices = self.network_scanner.scan(subnet)
logger.info(f"Discovered devices: {len(discovered_devices)}")
self.device_info.scan_nbtscan(subnet) self.device_info.scan_nbtscan(subnet)
def process_device(ip_mac: Tuple[str, str]) -> Optional[Device]: def process_device(ip_mac: Tuple[str, str]) -> Optional[Device]:
"""Обрабатывает устройство из пары IP и MAC.
Args:
ip_mac (Tuple[str, str]): Кортеж с IP-адресом и MAC-адресом.
Returns:
Optional[Device]: Объект Device или None, если устройство локальное.
"""
ip, mac = ip_mac ip, mac = ip_mac
if local_device and ip == local_device.ip: if local_device and ip == local_device.ip:
return None return None
@ -410,20 +288,19 @@ class SubnetService:
mac=mac, mac=mac,
hostname=self.device_info.get_hostname(ip) or existing_device.hostname, hostname=self.device_info.get_hostname(ip) or existing_device.hostname,
vendor=self.device_info.get_vendor(mac), vendor=self.device_info.get_vendor(mac),
open_ports=self.port_scanner.scan(ip) or existing_device.open_ports, open_ports=existing_device.open_ports,
status="online", status="online",
last_seen=formatted_now, last_seen=formatted_now,
note=existing_device.note, note=existing_device.note,
) )
vendor = self.device_info.get_vendor(mac) vendor = self.device_info.get_vendor(mac)
hostname = self.device_info.get_hostname(ip) hostname = self.device_info.get_hostname(ip)
open_ports = self.port_scanner.scan(ip)
return Device( return Device(
ip=ip, ip=ip,
mac=mac, mac=mac,
hostname=hostname, hostname=hostname,
vendor=vendor, vendor=vendor,
open_ports=open_ports, open_ports=[],
status="online", status="online",
last_seen=formatted_now, last_seen=formatted_now,
note=None, note=None,
@ -457,14 +334,6 @@ class SubnetService:
return SubnetScanResult(devices=devices) return SubnetScanResult(devices=devices)
def _get_local_device(self, subnet: str) -> Optional[Device]: def _get_local_device(self, subnet: str) -> Optional[Device]:
"""Определяет локальное устройство в указанной подсети.
Args:
subnet (str): Подсеть в формате CIDR (например, "192.168.1.0/24").
Returns:
Optional[Device]: Объект локального устройства или None, если не найдено.
"""
now = datetime.now().astimezone() now = datetime.now().astimezone()
formatted_now = now.strftime("%d-%m-%Y, %H:%M") formatted_now = now.strftime("%d-%m-%Y, %H:%M")
try: try:
@ -480,124 +349,70 @@ class SubnetService:
mac = addrs[netifaces.AF_LINK][0]["addr"] mac = addrs[netifaces.AF_LINK][0]["addr"]
vendor = self.device_info.get_vendor(mac) vendor = self.device_info.get_vendor(mac)
hostname = self.device_info.get_hostname(ip) hostname = self.device_info.get_hostname(ip)
open_ports = self.port_scanner.scan(ip) logger.info(f"Local device detected: IP={ip}, MAC={mac}, Interface={iface}")
logger.debug(
f"Local device: IP={ip}, MAC={mac}, Ports={open_ports}, Hostname={hostname}"
)
return Device( return Device(
ip=ip, ip=ip,
mac=mac, mac=mac,
hostname=hostname, hostname=hostname,
vendor=vendor, vendor=vendor,
open_ports=open_ports, open_ports=[],
status="online", status="online",
last_seen=formatted_now, last_seen=formatted_now,
note=None, note=None,
) )
logger.warning( logger.warning(f"Local device not found in subnet {subnet}")
f"Local device not found in subnet {subnet}")
return None return None
except Exception as e: except Exception as e:
logger.warning( logger.warning(f"Failed to get local device: {str(e)}")
f"Failed to get local device: {str(e)}")
return None return None
def add_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]: def add_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]:
"""Добавляет заметку к устройству по MAC-адресу.
Args:
mac (str): MAC-адрес устройства.
note (str): Текст заметки.
Returns:
Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке).
- Успех: (True, None).
- Ошибка: (False, сообщение об ошибке).
Raises:
Exception: Если произошла ошибка при сохранении.
"""
try: try:
saved_devices = self.storage.load_devices() saved_devices = self.storage.load_devices()
if mac not in saved_devices: if mac not in saved_devices:
logger.warning(f"Device with MAC {mac} not found") logger.warning(f"Device with MAC {mac} not found")
return False, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac) return False, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac)
if saved_devices[mac].note is not None: if saved_devices[mac].note is not None:
logger.error( logger.error(f"Note for device with MAC {mac} already exists")
f"Note for device with MAC {mac} already exists")
return False, self.translations.gettext("Заметка для устройства с MAC {mac} уже существует").format(mac=mac) return False, self.translations.gettext("Заметка для устройства с MAC {mac} уже существует").format(mac=mac)
saved_devices[mac].note = note saved_devices[mac].note = note
self.storage.save_devices(list(saved_devices.values())) self.storage.save_devices(list(saved_devices.values()))
logger.info( logger.info(f"Added note '{note}' for device with MAC {mac}")
f"Added note '{note}' for device with MAC {mac}")
return True, None return True, None
except Exception as e: except Exception as e:
logger.error( logger.error(f"Error adding note for MAC {mac}: {str(e)}")
f"Error adding note for MAC {mac}: {str(e)}")
return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e)) return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e))
def update_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]: def update_note(self, mac: str, note: str) -> Tuple[bool, Optional[str]]:
"""Обновляет заметку устройства по MAC-адресу.
Args:
mac (str): MAC-адрес устройства.
note (str): Новый текст заметки.
Returns:
Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке).
- Успех: (True, None).
- Ошибка: (False, сообщение об ошибке).
Raises:
Exception: Если произошла ошибка при сохранении.
"""
try: try:
saved_devices = self.storage.load_devices() saved_devices = self.storage.load_devices()
if mac not in saved_devices: if mac not in saved_devices:
logger.warning(f"Device with MAC {mac} not found") logger.warning(f"Device with MAC {mac} not found")
return False, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac) return False, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac)
if saved_devices[mac].note is None: if saved_devices[mac].note is None:
logger.error( logger.error(f"Note for device with MAC {mac} does not exist")
f"Note for device with MAC {mac} does not exist")
return False, self.translations.gettext("Заметка для устройства с MAC {mac} не существует").format(mac=mac) return False, self.translations.gettext("Заметка для устройства с MAC {mac} не существует").format(mac=mac)
saved_devices[mac].note = note saved_devices[mac].note = note
self.storage.save_devices(list(saved_devices.values())) self.storage.save_devices(list(saved_devices.values()))
logger.info( logger.info(f"Updated note '{note}' for device with MAC {mac}")
f"Updated note '{note}' for device with MAC {mac}")
return True, None return True, None
except Exception as e: except Exception as e:
logger.error( logger.error(f"Error updating note for MAC {mac}: {str(e)}")
f"Error updating note for MAC {mac}: {str(e)}")
return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e)) return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e))
def delete_note(self, mac: str) -> Tuple[bool, Optional[str]]: def delete_note(self, mac: str) -> Tuple[bool, Optional[str]]:
"""Удаляет заметку устройства по MAC-адресу.
Args:
mac (str): MAC-адрес устройства.
Returns:
Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке).
- Успех: (True, None).
- Ошибка: (False, сообщение об ошибке).
Raises:
Exception: Если произошла ошибка при сохранении.
"""
try: try:
saved_devices = self.storage.load_devices() saved_devices = self.storage.load_devices()
if mac not in saved_devices: if mac not in saved_devices:
logger.warning(f"Device with MAC {mac} not found") logger.warning(f"Device with MAC {mac} not found")
return False, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac) return False, self.translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac)
if saved_devices[mac].note is None: if saved_devices[mac].note is None:
logger.error( logger.error(f"Note for device with MAC {mac} does not exist")
f"Note for device with MAC {mac} does not exist")
return False, self.translations.gettext("Заметка для устройства с MAC {mac} не существует").format(mac=mac) return False, self.translations.gettext("Заметка для устройства с MAC {mac} не существует").format(mac=mac)
saved_devices[mac].note = None saved_devices[mac].note = None
self.storage.save_devices(list(saved_devices.values())) self.storage.save_devices(list(saved_devices.values()))
logger.info(f"Deleted note for device with MAC {mac}") logger.info(f"Deleted note for device with MAC {mac}")
return True, None return True, None
except Exception as e: except Exception as e:
logger.error( logger.error(f"Error deleting note for MAC {mac}: {str(e)}")
f"Error deleting note for MAC {mac}: {str(e)}")
return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e)) return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e))

284
SRV/services/wifi_service.py
View file

@ -9,112 +9,79 @@ import pywifi
from models.wifi import WifiAccessPoint, WifiScanResult from models.wifi import WifiAccessPoint, WifiScanResult
from pathlib import Path from pathlib import Path
import json import json
from utils.i18n import get_translations, _ from utils.i18n import get_translations
from babel.support import Translations from babel.support import Translations
from fastapi import Depends from fastapi import Depends
# Настройка логирования
logging.basicConfig( logging.basicConfig(
level=logging.DEBUG, level=logging.DEBUG,
format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
)
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
class WifiStorage: class WifiStorage:
"""Класс для хранения данных о Wi-Fi сетях и их заметках по SSID. """Класс для хранения данных о Wi-Fi сетях и их заметках по SSID."""
Предоставляет методы для загрузки и сохранения заметок, связанных с SSID сетей,
в JSON-файле на диске.
"""
def __init__(self, storage_path: str = "storage/wifi/wifi.json", translations: Translations = Depends(get_translations("services"))): def __init__(self, storage_path: str = "storage/wifi/wifi.json", translations: Translations = Depends(get_translations("services"))):
"""Инициализирует объект хранилища Wi-Fi с указанным путём к файлу.
Args:
storage_path (str, optional): Путь к файлу для хранения заметок.
По умолчанию "storage/wifi/wifi.json".
Создаёт родительские директории, если они не существуют.
"""
self.storage_file = Path(storage_path) self.storage_file = Path(storage_path)
self.storage_file.parent.mkdir(parents=True, exist_ok=True) self.storage_file.parent.mkdir(parents=True, exist_ok=True)
self.translations = translations self.translations = translations
logger.info(f"Wi-Fi storage initialized at: {self.storage_file.absolute()}")
def load_notes(self) -> Dict[str, Optional[str]]: def load_notes(self) -> Dict[str, Optional[str]]:
"""Загружает заметки по SSID из файла хранения. """Загружает заметки по SSID из файла хранения."""
Returns:
Dict[str, Optional[str]]: Словарь, где ключ — SSID, значение — заметка
или None, если заметки нет.
Raises:
Exception: Если произошла ошибка при чтении или парсинге файла.
"""
try: try:
if self.storage_file.exists(): if self.storage_file.exists():
with open(self.storage_file, "r", encoding="utf-8") as f: with open(self.storage_file, "r", encoding="utf-8") as f:
data = json.load(f) data = json.load(f)
if not isinstance(data, dict) or "networks" not in data:
logger.warning(f"Invalid JSON structure in {self.storage_file}, expected {'networks': [...]}, got {data}")
return {}
return {item["ssid"]: item.get("note") for item in data.get("networks", [])} return {item["ssid"]: item.get("note") for item in data.get("networks", [])}
return {} return {}
except json.JSONDecodeError as e:
logger.error(f"JSON decode error in {self.storage_file}: {str(e)}")
return {}
except Exception as e: except Exception as e:
logger.error(f"Error loading notes: {str(e)}") logger.error(f"Error loading notes from {self.storage_file}: {str(e)}")
return {} return {}
def save_notes(self, notes: Dict[str, Optional[str]]) -> None: def save_notes(self, notes: Dict[str, Optional[str]]) -> None:
"""Сохраняет заметки по SSID в файл. """Сохраняет заметки по SSID в файл."""
Args:
notes (Dict[str, Optional[str]]): Словарь с SSID как ключами и заметками
как значениями.
Raises:
Exception: Если произошла ошибка при записи в файл.
"""
try: try:
data = {"networks": [{"ssid": ssid, "note": note} for ssid, note in notes.items()]} data = {"networks": [{"ssid": ssid, "note": note} for ssid, note in notes.items()]}
with open(self.storage_file, "w", encoding="utf-8") as f: with open(self.storage_file, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2, ensure_ascii=False) json.dump(data, f, indent=2, ensure_ascii=False)
logger.info(f"Saved {len(notes)} notes to {self.storage_file}") logger.info(f"Saved {len(notes)} notes to {self.storage_file.absolute()}")
except PermissionError as e:
logger.error(f"Permission denied writing to {self.storage_file}: {str(e)}")
raise
except Exception as e: except Exception as e:
logger.error(f"Error saving notes: {str(e)}") logger.error(f"Error saving notes to {self.storage_file}: {str(e)}")
raise Exception(self.translations.gettext("Ошибка при сохранении заметок: {error}").format(error=str(e))) raise Exception(self.translations.gettext("Ошибка при сохранении заметок: {error}").format(error=str(e)))
class WifiService: class WifiService:
"""Сервис для сканирования текущих Wi-Fi точек доступа. """Сервис для сканирования текущих Wi-Fi точек доступа."""
Предоставляет функционал для обнаружения Wi-Fi сетей с использованием pywifi и iw,
определения характеристик точек доступа (сигнал, шифрование, канал и т.д.),
а также управления заметками.
"""
def __init__(self, translations: Translations = Depends(get_translations("services"))): def __init__(self, translations: Translations = Depends(get_translations("services"))):
"""Инициализирует сервис Wi-Fi с настройкой интерфейса и хранилища.
Инициализирует объект MacLookup для определения вендоров по MAC-адресам,
PyWiFi для работы с Wi-Fi-интерфейсами и объект хранилища заметок.
"""
self.mac_lookup = MacLookup() self.mac_lookup = MacLookup()
self.wifi = pywifi.PyWiFi() self.wifi = pywifi.PyWiFi()
self.interface = self._get_wifi_interface() self.interface = self._get_wifi_interface()
self.storage = WifiStorage() if not hasattr(translations, 'gettext'):
logger.warning("Translations object does not support gettext, using fallback")
self.translations = Translations.load("utils/locales/en/LC_MESSAGES/services.mo")
else:
self.translations = translations self.translations = translations
self.storage = WifiStorage()
def _get_wifi_interface(self) -> Optional[object]: def _get_wifi_interface(self) -> Optional[object]:
"""Автоматически определяет активный Wi-Fi-интерфейс. """Автоматически определяет активный Wi-Fi-интерфейс."""
Returns:
Optional[object]: Объект интерфейса PyWiFi или None, если интерфейс не найден.
Проверяет доступные интерфейсы, выполняет тестовое сканирование и возвращает
первый активный интерфейс, исключая P2P-интерфейсы.
"""
try: try:
interfaces = self.wifi.interfaces() interfaces = self.wifi.interfaces()
if not interfaces: if not interfaces:
logger.error(self.translations.gettext("Не найдено сетевых интерфейсов")) logger.error(self.translations.gettext("Не найдено сетевых интерфейсов"))
return None return None
logger.debug( logger.debug(f"Available interfaces: {[iface.name() for iface in interfaces]}")
f"Available interfaces: {[iface.name() for iface in interfaces]}"
)
for iface in interfaces: for iface in interfaces:
iface_name = iface.name() iface_name = iface.name()
@ -128,9 +95,7 @@ class WifiService:
if results: if results:
logger.info(f"Selected Wi-Fi interface: {iface_name}") logger.info(f"Selected Wi-Fi interface: {iface_name}")
return iface return iface
logger.debug( logger.debug(f"Interface {iface_name} returned no scan results")
f"Interface {iface_name} returned no scan results"
)
except Exception as e: except Exception as e:
logger.debug(f"Interface {iface_name} inactive: {str(e)}") logger.debug(f"Interface {iface_name} inactive: {str(e)}")
@ -142,36 +107,17 @@ class WifiService:
@lru_cache(maxsize=1000) @lru_cache(maxsize=1000)
def get_vendor(self, mac: str) -> str: def get_vendor(self, mac: str) -> str:
"""Определяет вендора по MAC-адресу с использованием кэширования. """Определяет вендора по MAC-адресу с использованием кэширования."""
Args:
mac (str): MAC-адрес устройства в формате XX:XX:XX:XX:XX:XX.
Returns:
str: Название вендора или "Unknown", если вендор не определён.
Raises:
Exception: Если произошла ошибка при запросе к базе вендоров.
"""
try: try:
return self.mac_lookup.lookup(mac) return self.mac_lookup.lookup(mac)
except Exception as e: except Exception as e:
logger.warning( logger.warning(f"Failed to determine vendor for MAC {mac}: {str(e)}")
f"Failed to determine vendor for MAC {mac}: {str(e)}")
return self.translations.gettext("Неизвестный") return self.translations.gettext("Неизвестный")
def _get_encryption_from_iw(self, ap_info: Dict) -> str: def _get_encryption_from_iw(self, ap_info: Dict) -> str:
"""Определяет тип шифрования из данных, полученных через iw. """Определяет тип шифрования из данных, полученных через iw."""
Args:
ap_info (Dict): Словарь с данными точки доступа, включая поле "security".
Returns:
str: Тип шифрования ("WPA3", "WPA2", "WPA", "WEP" или "Open").
"""
security = ap_info.get("security", "") security = ap_info.get("security", "")
logger.debug( logger.debug(f"Encryption for BSSID {ap_info.get('bssid')}: {security}")
f"Encryption for BSSID {ap_info.get('bssid')}: {security}")
if "WPA3" in security: if "WPA3" in security:
return self.translations.gettext("WPA3") return self.translations.gettext("WPA3")
elif "WPA2" in security or "RSN" in security: elif "WPA2" in security or "RSN" in security:
@ -183,14 +129,7 @@ class WifiService:
return self.translations.gettext("Открытая") return self.translations.gettext("Открытая")
def _get_channel(self, frequency: Optional[int]) -> Optional[int]: def _get_channel(self, frequency: Optional[int]) -> Optional[int]:
"""Преобразует частоту в номер канала. """Преобразует частоту в номер канала."""
Args:
frequency (Optional[int]): Частота в МГц или None.
Returns:
Optional[int]: Номер канала или None, если частота вне диапазона.
"""
if frequency is None: if frequency is None:
return None return None
try: try:
@ -201,20 +140,11 @@ class WifiService:
logger.debug(f"Frequency {frequency} outside 2.4/5 GHz range") logger.debug(f"Frequency {frequency} outside 2.4/5 GHz range")
return None return None
except Exception as e: except Exception as e:
logger.debug( logger.debug(f"Error determining channel for frequency {frequency}: {str(e)}")
f"Error determining channel for frequency {frequency}: {str(e)}"
)
return None return None
def _get_frequency(self, frequency: Optional[int]) -> Optional[str]: def _get_frequency(self, frequency: Optional[int]) -> Optional[str]:
"""Определяет частотный диапазон на основе частоты. """Определяет частотный диапазон на основе частоты."""
Args:
frequency (Optional[int]): Частота в МГц или None.
Returns:
Optional[str]: Диапазон ("2.4 GHz", "5 GHz" или None).
"""
if frequency is None: if frequency is None:
return None return None
try: try:
@ -225,39 +155,23 @@ class WifiService:
logger.debug(f"Frequency {frequency} outside 2.4/5 GHz range") logger.debug(f"Frequency {frequency} outside 2.4/5 GHz range")
return None return None
except Exception as e: except Exception as e:
logger.debug( logger.debug(f"Error determining band for frequency {frequency}: {str(e)}")
f"Error determining band for frequency {frequency}: {str(e)}"
)
return None return None
def _get_iw_scan_data(self) -> List[Dict]: def _get_iw_scan_data(self) -> List[Dict]:
"""Получает данные сканирования Wi-Fi через команду iw. """Получает данные сканирования Wi-Fi через команду iw."""
Returns:
List[Dict]: Список словарей с данными точек доступа (BSSID, SSID, частота,
сигнал, шифрование).
Raises:
subprocess.CalledProcessError: Если команда iw завершилась с ошибкой.
subprocess.TimeoutExpired: Если превышен таймаут выполнения.
Exception: Если произошла ошибка при парсинге вывода.
"""
try: try:
if not self.interface: if not self.interface:
logger.error(self.translations.gettext("Интерфейс не определён")) logger.error(self.translations.gettext("Интерфейс не определён"))
return [] return []
cmd = ["sudo", "iw", "dev", self.interface.name(), "scan"] cmd = ["sudo", "iw", "dev", self.interface.name(), "scan"]
logger.debug(f"Executing command: {' '.join(cmd)}") logger.debug(f"Executing command: {' '.join(cmd)}")
result = subprocess.run(cmd, result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
capture_output=True,
text=True,
timeout=10)
if result.returncode != 0: if result.returncode != 0:
logger.error(f"iw scan error: {result.stderr}") logger.error(f"iw scan error: {result.stderr}")
return [] return []
output = result.stdout output = result.stdout
logger.debug( logger.debug(f"iw scan output (first 500 chars): {output[:500]}...")
f"iw scan output (first 500 chars): {output[:500]}...")
aps = [] aps = []
current_ap = None current_ap = None
@ -274,15 +188,10 @@ class WifiService:
if bssid_match := bssid_pattern.match(line): if bssid_match := bssid_pattern.match(line):
if current_ap: if current_ap:
aps.append(current_ap) aps.append(current_ap)
current_ap = { current_ap = {"bssid": bssid_match.group(1).lower(), "security": ""}
"bssid": bssid_match.group(1).lower(),
"security": ""
}
elif current_ap: elif current_ap:
if ssid_match := ssid_pattern.match(line): if ssid_match := ssid_pattern.match(line):
current_ap["ssid"] = (ssid_match.group(1) current_ap["ssid"] = ssid_match.group(1).strip() if ssid_match.group(1).strip() else ""
if ssid_match.group(1).strip()
else "")
elif freq_match := freq_pattern.match(line): elif freq_match := freq_pattern.match(line):
current_ap["frequency"] = int(freq_match.group(1)) current_ap["frequency"] = int(freq_match.group(1))
elif signal_match := signal_pattern.match(line): elif signal_match := signal_pattern.match(line):
@ -291,19 +200,13 @@ class WifiService:
current_ap["security"] += "RSN;" current_ap["security"] += "RSN;"
elif wpa_pattern.match(line): elif wpa_pattern.match(line):
current_ap["security"] += "WPA;" current_ap["security"] += "WPA;"
elif security_type_match := security_type_pattern.search( elif security_type_match := security_type_pattern.search(line):
line): current_ap["security"] += security_type_match.group(1) + ";"
current_ap["security"] += security_type_match.group(
1) + ";"
if current_ap: if current_ap:
aps.append(current_ap) aps.append(current_ap)
logger.debug( logger.debug(f"Found {len(aps)} access points via iw: {[ap['bssid'] for ap in aps]}")
f"Found {len(aps)} access points via iw: {[ap['bssid'] for ap in aps]}"
)
for ap in aps:
logger.debug(f"Access point {ap['bssid']}: {ap}")
return aps return aps
except subprocess.CalledProcessError as e: except subprocess.CalledProcessError as e:
logger.error(f"Error executing iw scan: {e.stderr}") logger.error(f"Error executing iw scan: {e.stderr}")
@ -316,18 +219,7 @@ class WifiService:
return [] return []
def scan_wifi(self, timeout: int = 30) -> WifiScanResult: def scan_wifi(self, timeout: int = 30) -> WifiScanResult:
"""Сканирует текущие Wi-Fi точки доступа. """Сканирует текущие Wi-Fi точки доступа."""
Args:
timeout (int, optional): Таймаут ожидания в секундах. По умолчанию 30.
Returns:
WifiScanResult: Объект с результатами сканирования, содержащий список точек
доступа.
Сканирует Wi-Fi с использованием pywifi и iw, объединяет данные, определяет
лучшие точки по сигналу и сохраняет заметки.
"""
start_time = time.time() start_time = time.time()
aps = [] aps = []
@ -336,13 +228,20 @@ class WifiService:
return WifiScanResult(access_points=aps) return WifiScanResult(access_points=aps)
try: try:
logger.info( logger.info(f"Starting Wi-Fi scan on interface {self.interface.name()}")
f"Starting Wi-Fi scan on interface {self.interface.name()}" max_attempts = 3
) for attempt in range(max_attempts):
self.interface.scan() self.interface.scan()
time.sleep(10) time.sleep(min(timeout, 5))
py_results = self.interface.scan_results() py_results = self.interface.scan_results()
logger.debug(f"Found {len(py_results)} access points via pywifi") if py_results:
logger.info(f"Found {len(py_results)} access points via pywifi after attempt {attempt + 1}")
break
else:
logger.warning(f"Scan attempt {attempt + 1} failed, trying again (FAIL-BUSY?)")
else:
logger.error("All scan attempts failed")
return WifiScanResult(access_points=aps)
iw_results = self._get_iw_scan_data() iw_results = self._get_iw_scan_data()
iw_by_bssid = {ap["bssid"]: ap for ap in iw_results if ap.get("bssid")} iw_by_bssid = {ap["bssid"]: ap for ap in iw_results if ap.get("bssid")}
@ -355,8 +254,8 @@ class WifiService:
if not bssid: if not bssid:
logger.debug("Skipped access point without BSSID") logger.debug("Skipped access point without BSSID")
continue continue
ssid = ap.ssid if ap.ssid else "" ssid = ap.ssid if ap.ssid else f"Hidden_{bssid}"
hidden = not ssid hidden = not ap.ssid
signal = ap.signal if ap.signal else None signal = ap.signal if ap.signal else None
iw_data = iw_by_bssid.get(bssid, {}) iw_data = iw_by_bssid.get(bssid, {})
@ -366,10 +265,6 @@ class WifiService:
encryption = self._get_encryption_from_iw(iw_data) encryption = self._get_encryption_from_iw(iw_data)
vendor = self.get_vendor(bssid) vendor = self.get_vendor(bssid)
logger.debug(
f"Access point {bssid}: freq={frequency}, channel={channel}, encryption={encryption}, signal={signal}"
)
current_ap = WifiAccessPoint( current_ap = WifiAccessPoint(
ssid=ssid, ssid=ssid,
bssid=bssid, bssid=bssid,
@ -379,22 +274,27 @@ class WifiService:
frequency=freq_band, frequency=freq_band,
vendor=vendor, vendor=vendor,
hidden=hidden, hidden=hidden,
note=saved_notes.get(ssid) if ap.ssid else None,
) )
logger.debug(f"Processed AP: {current_ap.model_dump()}")
if ssid not in best_aps_by_ssid or ( if ssid not in best_aps_by_ssid or (
current_ap.signal and best_aps_by_ssid[ssid].signal current_ap.signal and best_aps_by_ssid[ssid].signal
and current_ap.signal > best_aps_by_ssid[ssid].signal): and current_ap.signal > best_aps_by_ssid[ssid].signal):
current_ap.note = saved_notes.get(ssid) current_ap.note = saved_notes.get(ssid) if ap.ssid else None
best_aps_by_ssid[ssid] = current_ap best_aps_by_ssid[ssid] = current_ap
aps = list(best_aps_by_ssid.values()) aps = list(best_aps_by_ssid.values())
# Сохраняем хотя бы SSID, если заметок нет
notes_to_save = {ap.ssid: ap.note for ap in aps if ap.ssid} notes_to_save = {ap.ssid: ap.note for ap in aps if ap.ssid}
if not notes_to_save and aps:
notes_to_save = {ap.ssid: None for ap in aps if ap.ssid} # Сохраняем SSID без заметок
if notes_to_save:
self.storage.save_notes(notes_to_save) self.storage.save_notes(notes_to_save)
else:
logger.warning("No notes or SSIDs to save, skipping file write")
logger.info( logger.info(f"Wi-Fi scan completed, found {len(aps)} unique networks")
f"Wi-Fi scan completed in {time.time() - start_time:.2f} seconds, "
f"found {len(aps)} unique networks"
)
return WifiScanResult(access_points=aps) return WifiScanResult(access_points=aps)
except Exception as e: except Exception as e:
@ -402,20 +302,7 @@ class WifiService:
return WifiScanResult(access_points=aps) return WifiScanResult(access_points=aps)
def add_note(self, ssid: str, note: str) -> Tuple[bool, Optional[str]]: def add_note(self, ssid: str, note: str) -> Tuple[bool, Optional[str]]:
"""Добавляет заметку к сети Wi-Fi по SSID. """Добавляет заметку к сети Wi-Fi по SSID."""
Args:
ssid (str): SSID сети Wi-Fi.
note (str): Текст заметки.
Returns:
Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке).
- Успех: (True, None).
- Ошибка: (False, сообщение об ошибке).
Raises:
Exception: Если произошла ошибка при сохранении заметки.
"""
try: try:
saved_notes = self.storage.load_notes() saved_notes = self.storage.load_notes()
if ssid not in saved_notes or saved_notes[ssid] is None: if ssid not in saved_notes or saved_notes[ssid] is None:
@ -432,20 +319,7 @@ class WifiService:
return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e)) return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e))
def update_note(self, ssid: str, note: str) -> Tuple[bool, Optional[str]]: def update_note(self, ssid: str, note: str) -> Tuple[bool, Optional[str]]:
"""Обновляет заметку сети Wi-Fi по SSID. """Обновляет заметку сети Wi-Fi по SSID."""
Args:
ssid (str): SSID сети Wi-Fi.
note (str): Новый текст заметки.
Returns:
Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке).
- Успех: (True, None).
- Ошибка: (False, сообщение об ошибке).
Raises:
Exception: Если произошла ошибка при сохранении заметки.
"""
try: try:
saved_notes = self.storage.load_notes() saved_notes = self.storage.load_notes()
if ssid not in saved_notes: if ssid not in saved_notes:
@ -463,19 +337,7 @@ class WifiService:
return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e)) return False, self.translations.gettext("Внутренняя ошибка сервера: {error}").format(error=str(e))
def delete_note(self, ssid: str) -> Tuple[bool, Optional[str]]: def delete_note(self, ssid: str) -> Tuple[bool, Optional[str]]:
"""Удаляет заметку сети Wi-Fi по SSID. """Удаляет заметку сети Wi-Fi по SSID."""
Args:
ssid (str): SSID сети Wi-Fi.
Returns:
Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке).
- Успех: (True, None).
- Ошибка: (False, сообщение об ошибке).
Raises:
Exception: Если произошла ошибка при сохранении изменений.
"""
try: try:
saved_notes = self.storage.load_notes() saved_notes = self.storage.load_notes()
if ssid not in saved_notes: if ssid not in saved_notes: