server/notification_system.py

"""Система уведомлений"""

import re
from datetime import datetime
from typing import Dict, Any

class NotificationSystem:
    CRITICAL_ACTIONS = [
        'delete', 'remove', 'kill', 'stop', 'shutdown',
        'format', 'clear', 'reset', 'purge',
        'upload', 'publish', 'deploy', 'push',
        'change_password', 'add_user', 'remove_user',
        'grant_access', 'revoke_access',
        'install', 'uninstall', 'update',
        'execute', 'run', 'start', 'stop',
        'create_file', 'delete_file', 'modify_file',
    ]

    HIGH_RISK_PATTERNS = [
        r'rm\s+-rf', r'del\s+/f', r'format\s+', r'mkfs',
        r'drop\s+database', r'truncate\s+', r'delete\s+from',
        r'ALTER\s+TABLE', r'DROP\s+TABLE',
        r'chmod\s+777', r'chown\s+root',
        r'sudo\s+', r'admin\s+',
    ]

    def __init__(self, chat_id: str = None):
        self.chat_id = chat_id
        self.notification_history = []
        self.enabled = True

    def set_chat_id(self, chat_id: str):
        self.chat_id = chat_id

    def set_enabled(self, enabled: bool):
        self.enabled = enabled

    def check_action(self, action: str, context: Dict[str, Any] = None) -> bool:
        action_lower = action.lower()
        context = context or {}
        for critical in self.CRITICAL_ACTIONS:
            if critical in action_lower:
                return True
        for pattern in self.HIGH_RISK_PATTERNS:
            if re.search(pattern, action_lower, re.IGNORECASE):
                return True
        if context.get('important', False):
            return True
        if context.get('files_changed', 0) > 3:
            return True
        return False

    def notify(self, action: str, details: str = "", severity: str = "info") -> str:
        if not self.enabled:
            return "🔇 Уведомления отключены"
        if not self.chat_id:
            return "⚠️ Chat ID не установлен"

        emoji_map = {'critical': '🚨', 'warning': '⚠️', 'info': 'ℹ️', 'success': '✅', 'error': '❌'}
        emoji = emoji_map.get(severity, 'ℹ️')

        message = f"{emoji} *УВЕДОМЛЕНИЕ*\n\n🔹 *Действие:* `{action}`\n"
        if details:
            message += f"📝 *Детали:*\n```\n{details[:500]}\n```\n"
        message += f"🕐 *Время:* {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"

        self.notification_history.append({
            'action': action, 'details': details,
            'severity': severity, 'timestamp': datetime.now().isoformat()
        })

        # Ленивый импорт — чтобы избежать циклических зависимостей при старте
        from .telegram_utils import send_tg
        send_tg(self.chat_id, message)
        return f"✅ Уведомление отправлено: {action}"

    def get_history(self, limit: int = 10) -> str:
        if not self.notification_history:
            return "📋 Нет уведомлений"
        result = "📋 *История уведомлений:*\n\n"
        for entry in self.notification_history[-limit:]:
            emoji = {'critical': '🚨', 'warning': '⚠️', 'info': 'ℹ️'}.get(entry['severity'], 'ℹ️')
            result += f"{emoji} `{entry['action']}` — {entry['timestamp']}\n"
        return result

NOTIFICATIONS = NotificationSystem()