Реализация добавления/удаления/изменения заметок для точек доступа Wi-Fi по ssid
This commit is contained in:
parent
58fa7d1ee6
commit
3eff71bc97
4 changed files with 198 additions and 39 deletions
|
|
@ -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)}"
|
||||
Loading…
Add table
Add a link
Reference in a new issue