209 lines
9.9 KiB
Python
209 lines
9.9 KiB
Python
import os
|
||
import logging
|
||
import subprocess
|
||
from typing import Optional
|
||
from fastapi import APIRouter, HTTPException, Depends, Response, Request, Body
|
||
from starlette.responses import FileResponse
|
||
from services.settings_service import SettingsService
|
||
from utils.i18n import get_translations
|
||
from babel.support import Translations
|
||
from models.subnet import SubnetScanRequest # Предполагается, что модель определена
|
||
|
||
logging.basicConfig(level=logging.DEBUG)
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter()
|
||
settings_service = SettingsService()
|
||
|
||
@router.post("/clear_subnet")
|
||
def clear_subnet(subnet: Optional[str] = None, translations: Translations = Depends(get_translations)):
|
||
"""Очищает данные узлов подсети, сохранённые в devices.json.
|
||
|
||
Удаляет все записи об узлах подсети, оставляя файл пустым.
|
||
|
||
Args:
|
||
subnet (Optional[str]): Подсеть в формате CIDR (например, "10.1.1.0/24").
|
||
Если None, очищает все подсети.
|
||
translations: Объект переводов для текущего языка.
|
||
|
||
Returns:
|
||
dict: Словарь с сообщением об успешной очистке.
|
||
|
||
Raises:
|
||
HTTPException: Если произошла ошибка очистки (status_code=500).
|
||
"""
|
||
try:
|
||
success = settings_service.clear_subnet(subnet)
|
||
if success:
|
||
if subnet:
|
||
return {"message": translations.gettext("Данные узлов подсети {subnet} успешно очищены").format(subnet=subnet)}
|
||
else:
|
||
return {"message": translations.gettext("Данные всех подсетей успешно очищены")}
|
||
else:
|
||
raise HTTPException(status_code=500, detail=translations.gettext("Не удалось очистить данные узлов подсети"))
|
||
except Exception as e:
|
||
logger.error(f"Error clearing subnet data: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка очистки: {error}").format(error=str(e)))
|
||
|
||
@router.post("/clear_wifi")
|
||
def clear_wifi(translations: Translations = Depends(get_translations)):
|
||
"""Очищает данные точек доступа Wi-Fi, сохранённые в wifi.json.
|
||
|
||
Удаляет все записи о точках доступа, оставляя файл пустым.
|
||
|
||
Args:
|
||
translations: Объект переводов для текущего языка.
|
||
|
||
Returns:
|
||
dict: Словарь с сообщением об успешной очистке.
|
||
|
||
Raises:
|
||
HTTPException: Если произошла ошибка очистки (status_code=500).
|
||
"""
|
||
try:
|
||
success = settings_service.clear_wifi()
|
||
if success:
|
||
return {"message": translations.gettext("Данные точек доступа Wi-Fi успешно очищены")}
|
||
else:
|
||
raise HTTPException(status_code=500, detail=translations.gettext("Не удалось очистить данные точек доступа Wi-Fi"))
|
||
except Exception as e:
|
||
logger.error(f"Error clearing wifi data: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка очистки: {error}").format(error=str(e)))
|
||
|
||
@router.post("/clear_all_data")
|
||
def clear_all_data(translations: Translations = Depends(get_translations)):
|
||
"""Очищает данные узлов подсети и точек доступа Wi-Fi в обоих файлах.
|
||
|
||
Удаляет все записи из devices.json и wifi.json.
|
||
|
||
Args:
|
||
translations: Объект переводов для текущего языка.
|
||
|
||
Returns:
|
||
dict: Словарь с сообщением об успешной очистке.
|
||
|
||
Raises:
|
||
HTTPException: Если произошла ошибка очистки (status_code=500).
|
||
"""
|
||
try:
|
||
success = settings_service.clear_all_data()
|
||
if success:
|
||
return {"message": translations.gettext("Все сетевые данные успешно очищены")}
|
||
else:
|
||
raise HTTPException(status_code=500, detail=translations.gettext("Не удалось очистить все сетевые данные"))
|
||
except Exception as e:
|
||
logger.error(f"Error clearing all data: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка очистки: {error}").format(error=str(e)))
|
||
|
||
WIFI_PATH = "storage/wifi/wifi.json"
|
||
|
||
@router.post("/export_devices", response_class=FileResponse)
|
||
async def export_devices(request: SubnetScanRequest = Body(...), translations: Translations = Depends(get_translations)):
|
||
"""
|
||
Экспортирует файл с данными устройств в формате JSON для указанной подсети.
|
||
|
||
Args:
|
||
request (SubnetScanRequest): Объект с полем subnet в формате CIDR.
|
||
translations: Объект переводов для текущего языка.
|
||
|
||
Returns:
|
||
FileResponse: Файл <subnet>.json для скачивания.
|
||
|
||
Raises:
|
||
HTTPException: Если файл не найден (status_code=404).
|
||
"""
|
||
subnet = request.subnet
|
||
file_path = f"storage/subnets/{subnet.replace('/', '_')}.json"
|
||
if not os.path.exists(file_path):
|
||
logger.warning(f"Export file not found: {file_path}")
|
||
raise HTTPException(status_code=404, detail=translations.gettext("Файл для подсети {subnet} не найден").format(subnet=subnet))
|
||
logger.info(f"Exporting file: {file_path}")
|
||
return FileResponse(
|
||
path=file_path,
|
||
filename=f"{subnet.replace('/', '_')}.json",
|
||
media_type="application/json"
|
||
)
|
||
|
||
@router.get("/export_wifi", response_class=FileResponse)
|
||
async def export_wifi(translations: Translations = Depends(get_translations)):
|
||
"""
|
||
Экспортирует файл с данными устройств в формате JSON.
|
||
|
||
Args:
|
||
translations: Объект переводов для текущего языка.
|
||
|
||
Returns:
|
||
FileResponse: Файл wifi.json для скачивания.
|
||
|
||
Raises:
|
||
HTTPException: Если файл не найден (status_code=404).
|
||
"""
|
||
if not os.path.exists(WIFI_PATH):
|
||
logger.warning(f"Export file not found: {WIFI_PATH}")
|
||
raise HTTPException(status_code=404, detail=translations.gettext("Файл не найден"))
|
||
logger.info(f"Exporting file: {WIFI_PATH}")
|
||
return FileResponse(
|
||
path=WIFI_PATH,
|
||
filename="wifi.json",
|
||
media_type="application/json"
|
||
)
|
||
|
||
@router.post("/set_language")
|
||
def set_language(locale: str, response: Response, translations: Translations = Depends(get_translations)):
|
||
"""
|
||
Устанавливает куку locale для смены языка пользователя.
|
||
|
||
Args:
|
||
locale (str): Используемый язык.
|
||
response (Response): Запрос.
|
||
translations: Объект переводов текущего языка.
|
||
|
||
Returns:
|
||
dict: Сообщение об успешной установке языка.
|
||
"""
|
||
lc_messages_path = os.path.join(os.path.dirname(__file__), '..', '..', 'locales', locale, 'LC_MESSAGES')
|
||
if not os.path.isdir(lc_messages_path):
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=translations.gettext("Язык '{locale}' не поддерживается.").format(locale=locale)
|
||
)
|
||
has_mo_or_po = any(f.endswith('.mo') or f.endswith('.po') for f in os.listdir(lc_messages_path))
|
||
if not has_mo_or_po:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail=translations.gettext("Язык '{locale}' не поддерживается (нет файлов переводов).").format(locale=locale)
|
||
)
|
||
response.set_cookie(key="locale", value=locale, path="/", httponly=True)
|
||
return {"message": translations.gettext("Язык успешно установлен: {locale}").format(locale=locale)}
|
||
|
||
@router.get("/get_system_language")
|
||
async def get_system_language(request: Request, translations: Translations = Depends(get_translations)):
|
||
"""Выводит текущий язык, установленный в системе на основе locale.
|
||
|
||
Выполняет команду 'locale' в системе и извлекает параметры LANG
|
||
|
||
Args:
|
||
request (Request): Объект запроса для доступа к параметрам.
|
||
translations: Объект переводов текущего языка.
|
||
|
||
Returns:
|
||
dict: Словарь с текущим языком системы.
|
||
|
||
Raises:
|
||
HTTPException: Если не удалось извлечь системный язык (status_code=500).
|
||
"""
|
||
try:
|
||
result = subprocess.run(['locale'], capture_output=True, text=True, check=True)
|
||
output = result.stdout
|
||
lang = next((line.split('=')[1].strip() for line in output.splitlines() if line.startswith('LANG=')), None)
|
||
if lang:
|
||
return {"language": lang}
|
||
else:
|
||
logger.warning(f"No locale found: {output}")
|
||
return {"language": 'unknown'}
|
||
except subprocess.CalledProcessError as e:
|
||
logger.error(f"Error executing locale command: {e}")
|
||
raise HTTPException(status_code=500, detail=translations.gettext("Не удалось определить язык в системе"))
|
||
except Exception as e:
|
||
logger.error(f"Unexpected error getting system language: {str(e)}")
|
||
raise HTTPException(status_code=500, detail=translations.gettext("Ошибка при определении языка в системе {error}").format(error=str(e)))
|