From 3eff71bc975184ad84923ddfd20622ecc6c971d8 Mon Sep 17 00:00:00 2001 From: "s.galanskiy" Date: Tue, 17 Jun 2025 11:40:20 +0500 Subject: [PATCH] =?UTF-8?q?=D0=A0=D0=B5=D0=B0=D0=BB=D0=B8=D0=B7=D0=B0?= =?UTF-8?q?=D1=86=D0=B8=D1=8F=20=D0=B4=D0=BE=D0=B1=D0=B0=D0=B2=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=B8=D1=8F/=D1=83=D0=B4=D0=B0=D0=BB=D0=B5=D0=BD=D0=B8?= =?UTF-8?q?=D1=8F/=D0=B8=D0=B7=D0=BC=D0=B5=D0=BD=D0=B5=D0=BD=D0=B8=D1=8F?= =?UTF-8?q?=20=D0=B7=D0=B0=D0=BC=D0=B5=D1=82=D0=BE=D0=BA=20=D0=B4=D0=BB?= =?UTF-8?q?=D1=8F=20=D1=82=D0=BE=D1=87=D0=B5=D0=BA=20=D0=B4=D0=BE=D1=81?= =?UTF-8?q?=D1=82=D1=83=D0=BF=D0=B0=20Wi-Fi=20=D0=BF=D0=BE=20ssid?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 3 +- SRV/api/v1/subnet.py | 81 +++++++++++++------ SRV/models/wifi.py | 4 +- SRV/services/wifi_service.py | 149 +++++++++++++++++++++++++++++++---- 4 files changed, 198 insertions(+), 39 deletions(-) diff --git a/.gitignore b/.gitignore index 8a295a8..ea15021 100644 --- a/.gitignore +++ b/.gitignore @@ -6,4 +6,5 @@ __pycache__ .wheels .vendor .python3-deb -SRV/storage/devices/devices.json \ No newline at end of file +SRV/storage/devices/devices.json +SRV/storage/wifi/wifi.json \ No newline at end of file diff --git a/SRV/api/v1/subnet.py b/SRV/api/v1/subnet.py index bad8885..101d3e6 100644 --- a/SRV/api/v1/subnet.py +++ b/SRV/api/v1/subnet.py @@ -8,7 +8,6 @@ from services.wifi_service import WifiService router = APIRouter() - @router.get("/scan", response_model=SubnetScanResult) def scan_subnet(subnet: str = "192.168.1.0/24", ports: List[int] = None): """Сканирует локальную подсеть и возвращает данные об устройствах""" @@ -17,9 +16,7 @@ def scan_subnet(subnet: str = "192.168.1.0/24", ports: List[int] = None): result = subnet_service.scan_subnet(subnet, ports) return result except Exception as e: - raise HTTPException(status_code=500, - detail=f"Ошибка сканирования: {str(e)}") - + raise HTTPException(status_code=500, detail=f"Ошибка сканирования: {str(e)}") @router.get("/wifi", response_model=WifiScanResult) def scan_wifi(timeout: int = 30): @@ -29,9 +26,7 @@ def scan_wifi(timeout: int = 30): result = wifi_service.scan_wifi(timeout=timeout) return result except Exception as e: - raise HTTPException(status_code=500, - detail=f"Ошибка сканирования Wi-Fi: {str(e)}") - + raise HTTPException(status_code=500, detail=f"Ошибка сканирования Wi-Fi: {str(e)}") @router.post("/devices/{mac}/note") def add_note(mac: str, request: NoteRequest): @@ -44,16 +39,11 @@ def add_note(mac: str, request: NoteRequest): raise HTTPException(status_code=404, detail=error_message) elif "уже существует" in error_message: raise HTTPException(status_code=400, detail=error_message) - return { - "message": - f"Заметка '{request.note}' добавлена для устройства {mac}" - } + return {"message": f"Заметка '{request.note}' добавлена для устройства {mac}"} except HTTPException as e: raise e except Exception as e: - raise HTTPException(status_code=500, - detail=f"Ошибка добавления заметки: {str(e)}") - + raise HTTPException(status_code=500, detail=f"Ошибка добавления заметки: {str(e)}") @router.put("/devices/{mac}/note") def update_note(mac: str, request: NoteRequest): @@ -66,16 +56,11 @@ def update_note(mac: str, request: NoteRequest): raise HTTPException(status_code=404, detail=error_message) elif "не существует" in error_message: raise HTTPException(status_code=400, detail=error_message) - return { - "message": - f"Заметка '{request.note}' обновлена для устройства {mac}" - } + return {"message": f"Заметка '{request.note}' обновлена для устройства {mac}"} except HTTPException as e: raise e except Exception as e: - raise HTTPException(status_code=500, - detail=f"Ошибка обновления заметки: {str(e)}") - + raise HTTPException(status_code=500, detail=f"Ошибка обновления заметки: {str(e)}") @router.delete("/devices/{mac}/note") def delete_note(mac: str): @@ -92,5 +77,55 @@ def delete_note(mac: str): except HTTPException as e: raise e except Exception as e: - raise HTTPException(status_code=500, - detail=f"Ошибка удаления заметки: {str(e)}") + raise HTTPException(status_code=500, detail=f"Ошибка удаления заметки: {str(e)}") + +@router.post("/wifi/{ssid}/note") +def add_wifi_note(ssid: str, request: NoteRequest): + """Добавляет заметку для сети Wi-Fi по SSID""" + try: + wifi_service = WifiService() + success, error_message = wifi_service.add_note(ssid, request.note) + if not success: + if "не найдена" in error_message: + raise HTTPException(status_code=404, detail=error_message) + elif "уже существует" in error_message: + raise HTTPException(status_code=400, detail=error_message) + return {"message": f"Заметка '{request.note}' добавлена для сети с SSID {ssid}"} + except HTTPException as e: + raise e + except Exception as e: + raise HTTPException(status_code=500, detail=f"Ошибка добавления заметки: {str(e)}") + +@router.put("/wifi/{ssid}/note") +def update_wifi_note(ssid: str, request: NoteRequest): + """Обновляет заметку для сети Wi-Fi по SSID""" + try: + wifi_service = WifiService() + success, error_message = wifi_service.update_note(ssid, request.note) + if not success: + if "не найдена" in error_message: + raise HTTPException(status_code=404, detail=error_message) + elif "не существует" in error_message: + raise HTTPException(status_code=400, detail=error_message) + return {"message": f"Заметка '{request.note}' обновлена для сети с SSID {ssid}"} + except HTTPException as e: + raise e + except Exception as e: + raise HTTPException(status_code=500, detail=f"Ошибка обновления заметки: {str(e)}") + +@router.delete("/wifi/{ssid}/note") +def delete_wifi_note(ssid: str): + """Удаляет заметку для сети Wi-Fi по SSID""" + try: + wifi_service = WifiService() + success, error_message = wifi_service.delete_note(ssid) + if not success: + if "не найдена" in error_message: + raise HTTPException(status_code=404, detail=error_message) + elif "не существует" in error_message: + raise HTTPException(status_code=400, detail=error_message) + return {"message": f"Заметка удалена для сети с SSID {ssid}"} + except HTTPException as e: + raise e + except Exception as e: + raise HTTPException(status_code=500, detail=f"Ошибка удаления заметки: {str(e)}") \ No newline at end of file diff --git a/SRV/models/wifi.py b/SRV/models/wifi.py index bfbb731..e870026 100644 --- a/SRV/models/wifi.py +++ b/SRV/models/wifi.py @@ -15,6 +15,7 @@ class WifiAccessPoint(BaseModel): frequency (Optional[str]): Частота вещания (2.4GHz/5GHz) vendor (Optional[str]): Производитель точки доступа (определяется по MAC-адресу) hidden (bool): Флаг скрытия названия точки доступа (ssid) + note (Optional[str]): Заметка для точки доступа """ ssid: Optional[str] = None @@ -25,6 +26,7 @@ class WifiAccessPoint(BaseModel): frequency: Optional[str] = None vendor: Optional[str] = None hidden: bool = False + note: Optional[str] = None class WifiScanResult(BaseModel): @@ -35,4 +37,4 @@ class WifiScanResult(BaseModel): access_points (List[WifiAccessPoint]): Список найденных Wi-Fi точек доступа """ - access_points: List[WifiAccessPoint] + access_points: List[WifiAccessPoint] \ No newline at end of file diff --git a/SRV/services/wifi_service.py b/SRV/services/wifi_service.py index 3e41ae0..0289cc7 100644 --- a/SRV/services/wifi_service.py +++ b/SRV/services/wifi_service.py @@ -3,17 +3,48 @@ import logging import subprocess import re from functools import lru_cache -from typing import List, Optional, Dict +from typing import List, Optional, Dict, Tuple from mac_vendor_lookup import MacLookup import pywifi from pywifi import Profile from models.wifi import WifiAccessPoint, WifiScanResult +from pathlib import Path +import json +from datetime import datetime logging.basicConfig( level=logging.DEBUG, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s") logger = logging.getLogger(__name__) +class WifiStorage: + """Класс для хранения данных о Wi-Fi сетях и их заметках по SSID.""" + + def __init__(self, storage_path: str = "storage/wifi/wifi.json"): + self.storage_file = Path(storage_path) + self.storage_file.parent.mkdir(parents=True, exist_ok=True) + + def load_notes(self) -> Dict[str, Optional[str]]: + """Загружает заметки по SSID из файла хранения.""" + try: + if self.storage_file.exists(): + with open(self.storage_file, "r", encoding="utf-8") as f: + data = json.load(f) + return {item["ssid"]: item.get("note") for item in data.get("networks", [])} + return {} + except Exception as e: + logger.error(f"Ошибка при загрузке заметок: {str(e)}") + return {} + + def save_notes(self, notes: Dict[str, Optional[str]]) -> None: + """Сохраняет заметки по SSID в файл.""" + try: + data = {"networks": [{"ssid": ssid, "note": note} for ssid, note in notes.items()]} + with open(self.storage_file, "w", encoding="utf-8") as f: + json.dump(data, f, indent=2, ensure_ascii=False) + logger.info(f"Сохранено {len(notes)} заметок в {self.storage_file}") + except Exception as e: + logger.error(f"Ошибка при сохранении заметок: {str(e)}") class WifiService: """Сервис для сканирования текущих Wi-Fi точек доступа""" @@ -22,6 +53,7 @@ class WifiService: self.mac_lookup = MacLookup() self.wifi = pywifi.PyWiFi() self.interface = self._get_wifi_interface() + self.storage = WifiStorage() def _get_wifi_interface(self) -> Optional[object]: """Автоматически определяет активный Wi-Fi-интерфейс""" @@ -37,9 +69,7 @@ class WifiService: for iface in interfaces: iface_name = iface.name() - if ( - "p2p" in iface_name.lower() - ): # Исключаю сетевые интерфейсы типа p2p из выбора, возможно сделать решение получше нежели исключать интерфейсы + if "p2p" in iface_name.lower(): logger.debug(f"Пропущен интерфейс {iface_name} (P2P)") continue try: @@ -208,21 +238,16 @@ class WifiService: f"Начало сканирования Wi-Fi на интерфейсе {self.interface.name()}" ) self.interface.scan() - time.sleep( - 10 - ) # Нативное значение для wifi интерфейса предназначенное на сканирование и сбор результата. Оптимальное решение от 5 до 15 секунд + time.sleep(10) 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") - } + iw_by_bssid = {ap["bssid"]: ap for ap in iw_results if ap.get("bssid")} + + best_aps_by_ssid: Dict[str, WifiAccessPoint] = {} + saved_notes = self.storage.load_notes() - best_aps_by_ssid: Dict[str, WifiAccessPoint] = ( - {} - ) # Сделал словарь, чтобы агрегировать точки доступа у сетей для выявления лучшего for ap in py_results: bssid = ap.bssid.lower() if ap.bssid else None if not bssid: @@ -257,9 +282,14 @@ class WifiService: 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): + # При обновлении или добавлении точки сохраняем заметку из saved_notes + current_ap.note = saved_notes.get(ssid) best_aps_by_ssid[ssid] = current_ap aps = list(best_aps_by_ssid.values()) + # Сохраняем заметки для всех найденных SSID + notes_to_save = {ap.ssid: ap.note for ap in aps if ap.ssid} + self.storage.save_notes(notes_to_save) logger.info( f"Сканирование Wi-Fi завершено за {time.time() - start_time:.2f} секунд, " @@ -269,3 +299,94 @@ class WifiService: except Exception as e: logger.error(f"Ошибка при сканировании Wi-Fi: {str(e)}") return WifiScanResult(access_points=aps) + + def add_note(self, ssid: str, note: str) -> Tuple[bool, Optional[str]]: + """Добавляет заметку к сети Wi-Fi по SSID. + + Args: + ssid (str): SSID сети Wi-Fi. + note (str): Текст заметки. + + Returns: + Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке). + - Успех: (True, None). + - Ошибка: (False, сообщение об ошибке). + + Raises: + Exception: Если произошла ошибка при сохранении. + """ + try: + saved_notes = self.storage.load_notes() + if ssid not in saved_notes or saved_notes[ssid] is None: + logger.info(f"Добавлена новая заметка для сети с SSID {ssid}") + elif saved_notes[ssid] is not None: + logger.error(f"Заметка для сети с SSID {ssid} уже существует") + return False, f"Заметка для сети с SSID {ssid} уже существует" + saved_notes[ssid] = note + self.storage.save_notes(saved_notes) + logger.info(f"Добавлена заметка '{note}' для сети с SSID {ssid}") + return True, None + except Exception as e: + logger.error(f"Ошибка при добавлении заметки для SSID {ssid}: {str(e)}") + return False, f"Внутренняя ошибка сервера: {str(e)}" + + def update_note(self, ssid: str, note: str) -> Tuple[bool, Optional[str]]: + """Обновляет заметку сети Wi-Fi по SSID. + + Args: + ssid (str): SSID сети Wi-Fi. + note (str): Новый текст заметки. + + Returns: + Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке). + - Успех: (True, None). + - Ошибка: (False, сообщение об ошибке). + + Raises: + Exception: Если произошла ошибка при сохранении. + """ + try: + saved_notes = self.storage.load_notes() + if ssid not in saved_notes: + logger.warning(f"Сеть с SSID {ssid} не найдена") + return False, f"Сеть с SSID {ssid} не найдена" + if saved_notes[ssid] is None: + logger.error(f"Заметка для сети с SSID {ssid} не существует") + return False, f"Заметка для сети с SSID {ssid} не существует" + saved_notes[ssid] = note + self.storage.save_notes(saved_notes) + logger.info(f"Обновлена заметка '{note}' для сети с SSID {ssid}") + return True, None + except Exception as e: + logger.error(f"Ошибка при обновлении заметки для SSID {ssid}: {str(e)}") + return False, f"Внутренняя ошибка сервера: {str(e)}" + + def delete_note(self, ssid: str) -> Tuple[bool, Optional[str]]: + """Удаляет заметку сети Wi-Fi по SSID. + + Args: + ssid (str): SSID сети Wi-Fi. + + Returns: + Tuple[bool, Optional[str]]: Кортеж (успех, сообщение об ошибке). + - Успех: (True, None). + - Ошибка: (False, сообщение об ошибке). + + Raises: + Exception: Если произошла ошибка при сохранении. + """ + try: + saved_notes = self.storage.load_notes() + if ssid not in saved_notes: + logger.warning(f"Сеть с SSID {ssid} не найдена") + return False, f"Сеть с SSID {ssid} не найдена" + if saved_notes[ssid] is None: + logger.error(f"Заметка для сети с SSID {ssid} не существует") + return False, f"Заметка для сети с SSID {ssid} не существует" + saved_notes[ssid] = None + self.storage.save_notes(saved_notes) + logger.info(f"Удалена заметка для сети с SSID {ssid}") + return True, None + except Exception as e: + logger.error(f"Ошибка при удалении заметки для SSID {ssid}: {str(e)}") + return False, f"Внутренняя ошибка сервера: {str(e)}" \ No newline at end of file