from fastapi import APIRouter, HTTPException, Body, Depends
from models.subnet import SubnetScanResult, SubnetScanRequest
from models.wifi import WifiScanResult
from models.request import NoteRequest, WifiConnectRequest, WifiDisconnectRequest
from services.subnet_service import SubnetService, get_subnet_service
from services.wifi_service import WifiService
from services.connect_service import WifiConnectionService
from utils.i18n import get_translations, _
from babel.support import Translations
import logging
router = APIRouter()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
[docs]
@router.post("/devices", response_model=SubnetScanResult)
def scan_subnet(request: SubnetScanRequest = Body(...), subnet_service: SubnetService = Depends(get_subnet_service), translations: Translations = Depends(get_translations())):
"""Сканирует локальную подсеть и возвращает данные об устройствах.
Выполняет сканирование указанной подсети для обнаружения активных устройств.
Args:
request (SubnetScanRequest): Объект с полем subnet.
subnet_service: Сервис сканирования подсети.
translations: Объект переводов для текущего языка.
Returns:
SubnetScanResult: Объект с результатами сканирования (устройства и их данные).
Raises:
HTTPException: Если произошла ошибка сканирования (status_code=500).
"""
try:
# Если 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)
if not result.devices and result.error:
logger.warning(f"No devices found in subnet {subnet or 'auto-detected'}")
return result
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)))
[docs]
@router.get("/wifi", response_model=WifiScanResult)
def scan_wifi(timeout: int = 30, translations: Translations = Depends(get_translations())):
"""Сканирует Wi-Fi точки доступа и возвращает данные о них.
Выполняет сканирование доступных Wi-Fi сетей с указанным таймаутом.
Args:
timeout (int, optional): Таймаут сканирования в секундах. По умолчанию 30.
translations: Объект переводов для текущего языка.
Returns:
WifiScanResult: Объект с результатами сканирования (список Wi-Fi сетей).
Raises:
HTTPException: Если произошла ошибка сканирования Wi-Fi (status_code=500).
"""
try:
wifi_service = WifiService()
result = wifi_service.scan_wifi(timeout=timeout)
return result
except Exception as e:
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка сканирования Wi-Fi: {error}").format(error=str(e)))
[docs]
@router.post("/devices/{mac}/note")
def add_note(mac: str, request: NoteRequest, subnet_service: SubnetService = Depends(get_subnet_service), translations: Translations = Depends(get_translations())):
"""Добавляет заметку для устройства по MAC-адресу.
Привязывает заметку к устройству, идентифицируемому по MAC-адресу.
Args:
mac (str): MAC-адрес устройства.
request (NoteRequest): Объект с полем "note" для текста заметки.
subnet_service: Сервис сканирования подсети.
translations: Объект переводов для текущего языка.
Returns:
dict: Словарь с сообщением об успешном добавлении заметки.
Raises:
HTTPException: Если устройство не найдено (status_code=404), заметка уже
существует (status_code=400), или произошла внутренняя ошибка (status_code=500).
"""
try:
success, result = subnet_service.add_note(mac, request.note, translations)
if not success:
if result[0] == "not_found":
raise HTTPException(status_code=404, detail=translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac))
elif result[0] == "note_exists":
raise HTTPException(status_code=400, detail=translations.gettext("Заметка для устройства с MAC {mac} уже существует").format(mac=mac))
else:
raise HTTPException(status_code=500, detail=translations.gettext("Внутренняя ошибка сервера: {error}").format(error=result[1]["error"]))
return {"message": translations.gettext("Заметка '{note}' добавлена для устройства {mac}").format(note=request.note, mac=mac)}
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка добавления заметки: {error}").format(error=str(e)))
[docs]
@router.put("/devices/{mac}/note")
def update_note(mac: str, request: NoteRequest, subnet_service: SubnetService = Depends(get_subnet_service), translations: Translations = Depends(get_translations())):
"""Обновляет заметку для устройства по MAC-адресу.
Изменяет существующую заметку, привязанную к устройству.
Args:
mac (str): MAC-адрес устройства.
request (NoteRequest): Объект с полем "note" для нового текста заметки.
subnet_service: Сервис сканирования подсети.
translations: Объект переводов для текущего языка.
Returns:
dict: Словарь с сообщением об успешном обновлении заметки.
Raises:
HTTPException: Если устройство не найдено (status_code=404), заметка не
существует (status_code=400), или произошла внутренняя ошибка (status_code=500).
"""
try:
success, result = subnet_service.update_note(mac, request.note, translations)
if not success:
if result[0] == "not_found":
raise HTTPException(status_code=404, detail=translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac))
elif result[0] == "note_not_exists":
raise HTTPException(status_code=400, detail=translations.gettext("Заметка для устройства с MAC {mac} не существует").format(mac=mac))
else:
raise HTTPException(status_code=500, detail=translations.gettext("Внутренняя ошибка сервера: {error}").format(error=result[1]["error"]))
return {"message": translations.gettext("Заметка '{note}' обновлена для устройства {mac}").format(note=request.note, mac=mac)}
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка обновления заметки: {error}").format(error=str(e)))
[docs]
@router.delete("/devices/{mac}/note")
def delete_note(mac: str, subnet_service: SubnetService = Depends(get_subnet_service), translations: Translations = Depends(get_translations())):
"""Удаляет заметку для устройства по MAC-адресу.
Удаляет привязанную заметку для указанного устройства.
Args:
mac (str): MAC-адрес устройства.
subnet_service: Сервис сканирования подсети.
translations: Объект переводов для текущего языка.
Returns:
dict: Словарь с сообщением об успешном удалении заметки.
Raises:
HTTPException: Если устройство не найдено (status_code=404), заметка не
существует (status_code=400), или произошла внутренняя ошибка (status_code=500).
"""
try:
success, result = subnet_service.delete_note(mac, translations)
if not success:
if result[0] == "not_found":
raise HTTPException(status_code=404, detail=translations.gettext("Устройство с MAC {mac} не найдено").format(mac=mac))
elif result[0] == "note_not_exists":
raise HTTPException(status_code=400, detail=translations.gettext("Заметка для устройства с MAC {mac} не существует").format(mac=mac))
else:
raise HTTPException(status_code=500, detail=translations.gettext("Внутренняя ошибка сервера: {error}").format(error=result[1]["error"]))
return {"message": translations.gettext("Заметка удалена для устройства {mac}").format(mac=mac)}
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка удаления заметки: {error}").format(error=str(e)))
[docs]
@router.post("/wifi/{ssid}/note")
def add_wifi_note(ssid: str, request: NoteRequest, translations: Translations = Depends(get_translations())):
"""Добавляет заметку для сети Wi-Fi по SSID.
Привязывает заметку к Wi-Fi сети, идентифицируемой по SSID.
Args:
ssid (str): Идентификатор Wi-Fi сети.
request (NoteRequest): Объект с полем "note" для текста заметки.
translations: Объект переводов для текущего языка.
Returns:
dict: Словарь с сообщением об успешном добавлении заметки.
Raises:
HTTPException: Если сеть не найдена (status_code=404), заметка уже
существует (status_code=400), или произошла внутренняя ошибка (status_code=500).
"""
try:
wifi_service = WifiService()
success, result = wifi_service.add_note(ssid, request.note)
if not success:
if result[0] == "not_found":
raise HTTPException(status_code=404, detail=translations.gettext("Сеть с SSID {ssid} не найдена").format(ssid=ssid))
elif result[0] == "note_exists":
raise HTTPException(status_code=400, detail=translations.gettext("Заметка для сети с SSID {ssid} уже существует").format(ssid=ssid))
else:
raise HTTPException(status_code=500, detail=translations.gettext("Внутренняя ошибка сервера: {error}").format(error=result[1]["error"]))
return {"message": translations.gettext("Заметка '{note}' добавлена для сети с SSID {ssid}").format(note=request.note, ssid=ssid)}
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка добавления заметки: {error}").format(error=str(e)))
[docs]
@router.put("/wifi/{ssid}/note")
def update_wifi_note(ssid: str, request: NoteRequest, translations: Translations = Depends(get_translations())):
"""Обновляет заметку для сети Wi-Fi по SSID.
Изменяет существующую заметку, привязанную к Wi-Fi сети.
Args:
ssid (str): Идентификатор Wi-Fi сети.
request (NoteRequest): Объект с полем "note" для нового текста заметки.
translations: Объект переводов для текущего языка.
Returns:
dict: Словарь с сообщением об успешном обновлении заметки.
Raises:
HTTPException: Если сеть не найдена (status_code=404), заметка не
существует (status_code=400), или произошла внутренняя ошибка (status_code=500).
"""
try:
wifi_service = WifiService()
success, result = wifi_service.update_note(ssid, request.note)
if not success:
if result[0] == "not_found":
raise HTTPException(status_code=404, detail=translations.gettext("Сеть с SSID {ssid} не найдена").format(ssid=ssid))
elif result[0] == "note_not_exists":
raise HTTPException(status_code=400, detail=translations.gettext("Заметка для сети с SSID {ssid} не существует").format(ssid=ssid))
else:
raise HTTPException(status_code=500, detail=translations.gettext("Внутренняя ошибка сервера: {error}").format(error=result[1]["error"]))
return {"message": translations.gettext("Заметка '{note}' обновлена для сети с SSID {ssid}").format(note=request.note, ssid=ssid)}
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка обновления заметки: {error}").format(error=str(e)))
[docs]
@router.delete("/wifi/{ssid}/note")
def delete_wifi_note(ssid: str, translations: Translations = Depends(get_translations())):
"""Удаляет заметку для сети Wi-Fi по SSID.
Удаляет привязанную заметку для указанной Wi-Fi сети.
Args:
ssid (str): Идентификатор Wi-Fi сети.
translations: Объект переводов для текущего языка.
Returns:
dict: Словарь с сообщением об успешном удалении заметки.
Raises:
HTTPException: Если сеть не найдена (status_code=404), заметка не
существует (status_code=400), или произошла внутренняя ошибка (status_code=500).
"""
try:
wifi_service = WifiService()
success, result = wifi_service.delete_note(ssid)
if not success:
if result[0] == "not_found":
raise HTTPException(status_code=404, detail=translations.gettext("Сеть с SSID {ssid} не найдена").format(ssid=ssid))
elif result[0] == "note_not_exists":
raise HTTPException(status_code=400, detail=translations.gettext("Заметка для сети с SSID {ssid} не существует").format(ssid=ssid))
else:
raise HTTPException(status_code=500, detail=translations.gettext("Внутренняя ошибка сервера: {error}").format(error=result[1]["error"]))
return {"message": translations.gettext("Заметка удалена для сети с SSID {ssid}").format(ssid=ssid)}
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка удаления заметки: {error}").format(error=str(e)))
[docs]
@router.post("/wifi/connect")
def connect_to_wifi(request: WifiConnectRequest, translations: Translations = Depends(get_translations())):
"""Подключается к Wi-Fi сети по SSID и паролю.
Выполняет подключение к указанной Wi-Fi сети с использованием предоставленных
учетных данных.
Args:
request (WifiConnectRequest): Объект с полями "ssid" и "password".
translations: Объект переводов для текущего языка.
Returns:
dict: Словарь с сообщением об успешном подключении.
Raises:
HTTPException: Если подключение не удалось (status_code=400).
"""
try:
wifi_service = WifiConnectionService(translations)
success, message = wifi_service.connect(request.ssid, request.password)
if not success:
raise HTTPException(status_code=400, detail=translations.gettext(message))
return {"message": translations.gettext("Успешно подключено к {ssid}").format(ssid=request.ssid)}
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка подключения к Wi-Fi: {error}").format(error=str(e)))
[docs]
@router.post("/wifi/disconnect")
def disconnect_from_wifi(request: WifiDisconnectRequest, translations: Translations = Depends(get_translations())):
"""Отключается от Wi-Fi сети по указанному SSID.
Выполняет отключение от указанной Wi-Fi сети.
Args:
request (WifiDisconnectRequest): Объект с полем "ssid".
translations: Объект переводов для текущего языка.
Returns:
dict: Словарь с сообщением об успешном отключении.
Raises:
HTTPException: Если отключение не удалось (status_code=400).
"""
try:
wifi_service = WifiConnectionService(translations)
success, message = wifi_service.disconnect(request.ssid)
if not success:
raise HTTPException(status_code=400, detail=translations.gettext(message))
return {"message": translations.gettext(message or "Успешно отключено от {ssid}").format(ssid=request.ssid)}
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка отключения от Wi-Fi: {error}").format(error=str(e)))
[docs]
@router.get("/wifi/status")
def get_wifi_status(translations: Translations = Depends(get_translations())):
"""Проверяет статус текущего подключения к Wi-Fi.
Возвращает информацию о текущем состоянии подключения и SSID, если подключение активно.
Args:
translations: Объект переводов для текущего языка.
Returns:
dict: Словарь с ключами:
- "status" (str): Состояние ("connected" или другое).
- "ssid" (str, optional): SSID текущей сети, если подключено.
Raises:
HTTPException: Если проверка статуса не удалась (status_code=400).
"""
try:
wifi_service = WifiConnectionService(translations)
success, message = wifi_service.get_status()
if not success:
raise HTTPException(status_code=400, detail=translations.gettext(message))
return {"status": translations.gettext("connected"), "ssid": message}
except HTTPException as e:
raise e
except Exception as e:
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка проверки статуса Wi-Fi: {error}").format(error=str(e)))