ComConfigCopy/src/network/servers/tftp_server.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import os
import socket
import threading
from typing import Optional, Dict, Any, Callable
import tftpy

from core.config import AppConfig
from core.exceptions import TFTPError
from .base_server import BaseServer

class TFTPServer(BaseServer):
    """TFTP сервер для передачи файлов."""
    
    def __init__(self):
        super().__init__()
        self._server: Optional[tftpy.TftpServer] = None
        self._root_dir = AppConfig.CONFIGS_DIR
        self._host = "0.0.0.0"
        self._port = AppConfig.TFTP_PORT
        self._timeout = AppConfig.TFTP_TIMEOUT
        
        # Создаем директорию, если её нет
        os.makedirs(self._root_dir, exist_ok=True)
    
    def configure(self, root_dir: Optional[str] = None, host: Optional[str] = None,
                 port: Optional[int] = None, timeout: Optional[int] = None) -> None:
        """
        Конфигурация TFTP сервера.
        
        Args:
            root_dir: Корневая директория для файлов
            host: IP-адрес для прослушивания
            port: Порт сервера
            timeout: Таймаут операций
        """
        if root_dir is not None:
            self._root_dir = root_dir
            os.makedirs(self._root_dir, exist_ok=True)
        if host is not None:
            self._host = host
        if port is not None:
            self._port = port
        if timeout is not None:
            self._timeout = timeout
    
    def start(self) -> None:
        """
        Запуск TFTP сервера.
        
        Raises:
            TFTPError: При ошибке запуска сервера
        """
        if self.is_running:
            self._logger.warning("TFTP сервер уже запущен")
            return
            
        try:
            # Создаем серверный объект
            self._server = tftpy.TftpServer(self._root_dir)
            
            # Запускаем сервер в отдельном потоке
            self._start_in_thread(self._serve)
            
            self._notify_started({
                "host": self._host,
                "port": self._port,
                "root_dir": self._root_dir
            })
            
            self._logger.info(f"TFTP сервер запущен на {self._host}:{self._port}")
            
        except Exception as e:
            self._notify_error(e)
            raise TFTPError(f"Ошибка запуска TFTP сервера: {e}")
    
    def stop(self) -> None:
        """Остановка TFTP сервера."""
        if not self.is_running:
            return
            
        try:
            if self._server:
                self._server.stop()
                self._server = None
            
            self._stop_thread()
            self._notify_stopped()
            
            self._logger.info("TFTP сервер остановлен")
            
        except Exception as e:
            self._logger.error(f"Ошибка при остановке TFTP сервера: {e}")
    
    def _serve(self) -> None:
        """Основной цикл сервера."""
        try:
            self._server.listen(self._host, self._port, timeout=self._timeout)
        except Exception as e:
            self._notify_error(e)
            self._logger.error(f"Ошибка в работе TFTP сервера: {e}")
            self.stop()
    
    def get_server_info(self) -> Dict[str, Any]:
        """
        Получение информации о сервере.
        
        Returns:
            Dict[str, Any]: Информация о сервере
        """
        return {
            "running": self.is_running,
            "host": self._host,
            "port": self._port,
            "root_dir": self._root_dir,
            "timeout": self._timeout
        }
    
    def list_files(self) -> list[str]:
        """
        Получение списка файлов в корневой директории.
        
        Returns:
            list[str]: Список файлов
        """
        try:
            return os.listdir(self._root_dir)
        except Exception as e:
            self._logger.error(f"Ошибка при получении списка файлов: {e}")
            return []
    
    def get_file_info(self, filename: str) -> Optional[Dict[str, Any]]:
        """
        Получение информации о файле.
        
        Args:
            filename: Имя файла
            
        Returns:
            Optional[Dict[str, Any]]: Информация о файле или None
        """
        file_path = os.path.join(self._root_dir, filename)
        if not os.path.exists(file_path):
            return None
            
        try:
            stat = os.stat(file_path)
            return {
                "name": filename,
                "size": stat.st_size,
                "modified": stat.st_mtime,
                "created": stat.st_ctime
            }
        except Exception as e:
            self._logger.error(f"Ошибка при получении информации о файле: {e}")
            return None
    
    def delete_file(self, filename: str) -> bool:
        """
        Удаление файла из корневой директории.
        
        Args:
            filename: Имя файла
            
        Returns:
            bool: True если файл удален успешно
        """
        file_path = os.path.join(self._root_dir, filename)
        if not os.path.exists(file_path):
            return False
            
        try:
            os.remove(file_path)
            self._logger.info(f"Файл удален: {filename}")
            return True
        except Exception as e:
            self._logger.error(f"Ошибка при удалении файла: {e}")
            return False