log_server/log_server.py - master - Shtirlitc/TickBot

log_server/log_server.py
218 строк9 KB
Shtirlyts
Initial commit: TickBot attendance bot with stale-status fixes
31 июл 2026, 23:40
31 июл 2026, 23:4030e955c
import asyncio
import aiohttp
from aiohttp import web, WSMsgType
import os
import json
import logging
from pathlib import Path
from typing import Set, Dict, List
import sys
import os
sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from config import settings

# Настройка логгирования
logger = logging.getLogger(__name__)

class LogWebSocketServer:
    def __init__(self, host: str = "localhost", port: int = 8083):
        self.host = host
        self.port = port
        self.clients: Set[web.WebSocketResponse] = set()
        self.log_file_path = Path(settings.log_file_path)
        self.server_token = os.getenv("LOG_SERVER_TOKEN", "default_token")
        self.app = web.Application()
        self.runner = None
        self.site = None
        
        # Настройка маршрутов
        self.app.add_routes([
            web.get('/ws', self.websocket_handler),
            web.get('/logger.html', self.serve_logger_html),
            web.get('/', self.serve_logger_html),
            web.static('/', '.'),
        ])
        
    async def serve_logger_html(self, request: web.Request) -> web.StreamResponse:
        """Обслуживание файла logger.html"""
        try:
            with open('logger.html', 'r', encoding='utf-8') as f:
                content = f.read()
            return web.Response(text=content, content_type='text/html')
        except FileNotFoundError:
            return web.Response(text="File not found", status=404)
    
    async def authenticate_client(self, ws: web.WebSocketResponse) -> bool:
        """Аутентификация клиента по токену"""
        try:
            # Ожидаем токен от клиента
            msg = await ws.receive(timeout=10.0)
            if msg.type == WSMsgType.TEXT:
                data = json.loads(msg.data)
                if data.get('token') == self.server_token:
                    await ws.send_str(json.dumps({'type': 'auth', 'status': 'success'}))
                    return True
                else:
                    await ws.send_str(json.dumps({'type': 'auth', 'status': 'failed', 'message': 'Invalid token'}))
                    return False
            else:
                await ws.send_str(json.dumps({'type': 'auth', 'status': 'failed', 'message': 'Invalid message type'}))
                return False
        except asyncio.TimeoutError:
            await ws.send_str(json.dumps({'type': 'auth', 'status': 'failed', 'message': 'Authentication timeout'}))
            return False
        except Exception as e:
            logger.error(f"Authentication error: {e}")
            await ws.send_str(json.dumps({'type': 'auth', 'status': 'failed', 'message': str(e)}))
            return False
    
    async def websocket_handler(self, request: web.Request) -> web.WebSocketResponse:
        """Обработчик WebSocket-соединений"""
        ws = web.WebSocketResponse()
        await ws.prepare(request)
        self.clients.add(ws)
        logger.info(f"New WebSocket connection. Total clients: {len(self.clients)}")
        
        try:
            # Аутентификация клиента
            if not await self.authenticate_client(ws):
                await ws.close()
                return ws
            
            # Отправка последних 100 строк лога
            last_lines = await self.get_last_log_lines(100)
            await ws.send_str(json.dumps({
                'type': 'logs',
                'lines': last_lines
            }))
            
            # Основной цикл обработки сообщений
            async for msg in ws:
                if msg.type == WSMsgType.TEXT:
                    try:
                        data = json.loads(msg.data)
                        if data.get('action') == 'ping':
                            await ws.send_str(json.dumps({'type': 'pong'}))
                    except json.JSONDecodeError:
                        pass
                elif msg.type == WSMsgType.ERROR:
                    logger.error(f"WebSocket error: {ws.exception()}")
                    
        finally:
            self.clients.discard(ws)
            logger.info(f"WebSocket connection closed. Total clients: {len(self.clients)}")
            
        return ws
    
    async def get_last_log_lines(self, count: int = 100) -> List[str]:
        """Получение последних строк из файла логов"""
        try:
            if not self.log_file_path.exists():
                return []
            
            # Читаем последние N строк
            with open(self.log_file_path, 'r', encoding='utf-8') as f:
                lines = f.readlines()
                return [line.strip() for line in lines[-count:] if line.strip()]
        except Exception as e:
            logger.error(f"Error reading log file: {e}")
            return []
    
    async def broadcast_log_line(self, line: str):
        """Отправка новой строки лога всем подключенным клиентам"""
        if not self.clients:
            return
            
        message = json.dumps({
            'type': 'log',
            'line': line
        })
        
        # Отправляем всем клиентам
        for ws in list(self.clients):  # Копируем список, чтобы избежать изменений во время итерации
            try:
                await ws.send_str(message)
            except Exception as e:
                logger.error(f"Error sending to client: {e}")
                self.clients.discard(ws)
    
    async def monitor_log_file(self):
        """Мониторинг файла логов и отправка новых строк клиентам"""
        last_position = 0
        
        while True:
            try:
                if not self.log_file_path.exists():
                    await asyncio.sleep(1)
                    continue
                
                # Получаем текущий размер файла
                current_size = self.log_file_path.stat().st_size
                
                # Если файл уменьшился (пересоздан), начинаем с начала
                if current_size < last_position:
                    last_position = 0
                
                # Если есть новые данные
                if current_size > last_position:
                    with open(self.log_file_path, 'r', encoding='utf-8') as f:
                        f.seek(last_position)
                        new_data = f.read()
                        last_position = f.tell()
                    
                    # Разбиваем на строки и отправляем каждую
                    lines = new_data.splitlines()
                    for line in lines:
                        if line.strip():  # Пропускаем пустые строки
                            await self.broadcast_log_line(line.strip())
                
                await asyncio.sleep(0.5)  # Проверяем каждые 500 мс
                
            except Exception as e:
                logger.error(f"Error monitoring log file: {e}")
                await asyncio.sleep(1)
    
    async def start(self):
        """Запуск сервера"""
        try:
            self.runner = web.AppRunner(self.app)
            await self.runner.setup()
            self.site = web.TCPSite(self.runner, self.host, self.port)
            await self.site.start()
            logger.info(f"Log server started at http://{self.host}:{self.port}")
            
            # Запускаем мониторинг файла логов
            asyncio.create_task(self.monitor_log_file())
            
        except Exception as e:
            logger.error(f"Error starting server: {e}")
            raise
    
    async def stop(self):
        """Остановка сервера"""
        try:
            # Закрываем все WebSocket соединения
            for ws in list(self.clients):
                await ws.close()
            
            # Останавливаем сервер
            if self.site:
                await self.site.stop()
            if self.runner:
                await self.runner.cleanup()
                
            logger.info("Log server stopped")
        except Exception as e:
            logger.error(f"Error stopping server: {e}")

# Функция для запуска сервера
async def start_log_server():
    """Функция для запуска сервера логов"""
    server = LogWebSocketServer()
    await server.start()
    return server

if __name__ == "__main__":
    # Для тестового запуска
    logging.basicConfig(level=logging.INFO)
    asyncio.run(start_log_server())