gonets/SRV/services/wifi_service.py

import time
import logging
import subprocess
import re
from functools import lru_cache
from typing import List, Optional, Dict
from mac_vendor_lookup import MacLookup
import pywifi
from models.wifi import WifiAccessPoint, WifiScanResult

logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)

class WifiService:
    """Сервис для сканирования текущих Wi-Fi точек доступа"""

    def __init__(self):
        self.mac_lookup = MacLookup()
        self.wifi = pywifi.PyWiFi()
        self.interface = self._get_wifi_interface()

    def _get_wifi_interface(self) -> Optional[object]:
        """Автоматически определяет активный Wi-Fi-интерфейс"""
        try:
            interfaces = self.wifi.interfaces()
            if not interfaces:
                logger.error("Не найдено сетевых интерфейсов")
                return None

            logger.debug(f"Доступные интерфейсы: {[iface.name() for iface in interfaces]}")

            for iface in interfaces:
                iface_name = iface.name()
                if 'p2p' in iface_name.lower(): # Исключаю сетевые интерфейсы типа p2p из выбора, возможно сделать решение получше нежели исключать интерфейсы
                    logger.debug(f"Пропущен интерфейс {iface_name} (P2P)")
                    continue
                try:
                    iface.scan()
                    time.sleep(1)
                    results = iface.scan_results()
                    if results:
                        logger.info(f"Выбран Wi-Fi-интерфейс: {iface_name}")
                        return iface
                    logger.debug(f"Интерфейс {iface_name} не вернул результатов сканирования")
                except Exception as e:
                    logger.debug(f"Интерфейс {iface_name} неактивен: {str(e)}")

            logger.error("Не найден активный Wi-Fi-интерфейс")
            return None
        except Exception as e:
            logger.error(f"Ошибка при определении Wi-Fi-интерфейса: {str(e)}")
            return None

    @lru_cache(maxsize=1000)
    def get_vendor(self, mac: str) -> str:
        try:
            return self.mac_lookup.lookup(mac)
        except Exception as e:
            logger.warning(f"Не удалось определить производителя для MAC {mac}: {str(e)}")
            return "Unknown"

    def _get_encryption_from_iw(self, ap_info: Dict) -> str:
        """Определяет шифрование из данных iw"""
        security = ap_info.get("security", "")
        logger.debug(f"Шифрование для BSSID {ap_info.get('bssid')}: {security}")
        if "WPA3" in security:
            return "WPA3"
        elif "WPA2" in security or "RSN" in security:
            return "WPA2"
        elif "WPA" in security:
            return "WPA"
        elif "WEP" in security:
            return "WEP"
        return "Open"

    def _get_channel(self, frequency: Optional[int]) -> Optional[int]:
        """Преобразует частоту в канал."""
        if frequency is None:
            return None
        try:
            if 2412 <= frequency <= 2472:
                return (frequency - 2412) // 5 + 1
            elif 5180 <= frequency <= 5825:
                return (frequency - 5180) // 5 + 36
            logger.debug(f"Частота {frequency} вне диапазона 2.4/5 GHz")
            return None
        except Exception as e:
            logger.debug(f"Ошибка при определении канала для частоты {frequency}: {str(e)}")
            return None

    def _get_frequency(self, frequency: Optional[int]) -> Optional[str]:
        """Определяет частотный диапазон."""
        if frequency is None:
            return None
        try:
            if 2412 <= frequency <= 2472:
                return "2.4 GHz"
            elif 5180 <= frequency <= 5825:
                return "5 GHz"
            logger.debug(f"Частота {frequency} вне диапазона 2.4/5 GHz")
            return None
        except Exception as e:
            logger.debug(f"Ошибка при определении диапазона для частоты {frequency}: {str(e)}")
            return None

    def _get_iw_scan_data(self) -> List[Dict]:
        """Получает данные сканирования через iw."""
        try:
            if not self.interface:
                logger.error("Интерфейс не определён")
                return []
            cmd = ["sudo", "iw", "dev", self.interface.name(), "scan"]
            logger.debug(f"Выполняем команду: {' '.join(cmd)}")
            result = subprocess.run(cmd, capture_output=True, text=True, timeout=10)
            if result.returncode != 0:
                logger.error(f"Ошибка iw scan: {result.stderr}")
                return []
            output = result.stdout
            logger.debug(f"Вывод iw scan (первые 500 символов): {output[:500]}...")

            aps = []
            current_ap = None
            bssid_pattern = re.compile(r"BSS ([0-9a-f:]{17})")
            ssid_pattern = re.compile(r"SSID: ?(.*)")
            freq_pattern = re.compile(r"freq: (\d+)")
            signal_pattern = re.compile(r"signal: ([-\d.]+) dBm")
            rsn_pattern = re.compile(r"RSN:")
            wpa_pattern = re.compile(r"WPA:")
            security_type_pattern = re.compile(r"\* (WPA3|WPA2|WPA|WEP)")

            for line in output.splitlines():
                line = line.strip()
                if bssid_match := bssid_pattern.match(line):
                    if current_ap:
                        aps.append(current_ap)
                    current_ap = {"bssid": bssid_match.group(1).lower(), "security": ""}
                elif current_ap:
                    if ssid_match := ssid_pattern.match(line):
                        current_ap["ssid"] = ssid_match.group(1) if ssid_match.group(1).strip() else ""
                    elif freq_match := freq_pattern.match(line):
                        current_ap["frequency"] = int(freq_match.group(1))
                    elif signal_match := signal_pattern.match(line):
                        current_ap["signal"] = float(signal_match.group(1))
                    elif rsn_pattern.match(line):
                        current_ap["security"] += "RSN;"
                    elif wpa_pattern.match(line):
                        current_ap["security"] += "WPA;"
                    elif security_type_match := security_type_pattern.search(line):
                        current_ap["security"] += security_type_match.group(1) + ";"

            if current_ap:
                aps.append(current_ap)

            logger.debug(f"Найдено {len(aps)} точек через iw: {[ap['bssid'] for ap in aps]}")
            for ap in aps:
                logger.debug(f"Данные точки {ap['bssid']}: {ap}")
            return aps
        except subprocess.CalledProcessError as e:
            logger.error(f"Ошибка при выполнении iw scan: {e.stderr}")
            return []
        except subprocess.TimeoutExpired:
            logger.error("Таймаут при выполнении iw scan")
            return []
        except Exception as e:
            logger.error(f"Ошибка при парсинге iw scan: {str(e)}")
            return []

    def scan_wifi(self, timeout: int = 30) -> WifiScanResult:
        """Сканирует текущие Wi-Fi точки доступа"""
        start_time = time.time()
        aps = []

        if not self.interface:
            logger.error("Wi-Fi-интерфейс не доступен")
            return WifiScanResult(access_points=aps)

        try:
            logger.info(f"Начало сканирования Wi-Fi на интерфейсе {self.interface.name()}")
            self.interface.scan()
            time.sleep(10) # Нативное значение для wifi интерфейса предназначенное на сканирование и сбор результата. Оптимальное решение от 5 до 15 секунд
            py_results = self.interface.scan_results()
            logger.debug(f"Найдено {len(py_results)} точек через pywifi")

            iw_results = self._get_iw_scan_data()
            iw_by_bssid = {ap['bssid']: ap for ap in iw_results if ap.get('bssid')}

            best_aps_by_ssid: Dict[str, WifiAccessPoint] = {} # Сделал словарь чтобы агрегировать точки доступа у сетей для выявления лучшего
            for ap in py_results:
                bssid = ap.bssid.lower() if ap.bssid else None
                if not bssid:
                    logger.debug("Пропущена точка без BSSID")
                    continue
                ssid = ap.ssid if ap.ssid else ""
                hidden = not ssid
                signal = ap.signal if ap.signal else None
                iw_data = iw_by_bssid.get(bssid, {})

                frequency = iw_data.get("frequency")
                channel = self._get_channel(frequency)
                freq_band = self._get_frequency(frequency)
                encryption = self._get_encryption_from_iw(iw_data)
                vendor = self.get_vendor(bssid)

                logger.debug(
                    f"Точка {bssid}: freq={frequency}, channel={channel}, encryption={encryption}, signal={signal}")

                current_ap = WifiAccessPoint(
                    ssid=ssid,
                    bssid=bssid,
                    channel=channel,
                    signal=signal if signal else iw_data.get("signal"),
                    encryption=encryption,
                    frequency=freq_band,
                    vendor=vendor,
                    hidden=hidden
                )

                if ssid not in best_aps_by_ssid or (current_ap.signal and best_aps_by_ssid[ssid].signal and current_ap.signal > best_aps_by_ssid[ssid].signal):
                    best_aps_by_ssid[ssid] = current_ap

            aps = list(best_aps_by_ssid.values())

            logger.info(
                f"Сканирование Wi-Fi завершено за {time.time() - start_time:.2f} секунд, "
                f"найдено {len(aps)} уникальных сетей"
            )
            return WifiScanResult(access_points=aps)

        except Exception as e:
            logger.error(f"Ошибка при сканировании Wi-Fi: {str(e)}")
            return WifiScanResult(access_points=aps)