Spaces:
Runtime error
Runtime error
File size: 3,408 Bytes
8f22f55 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 | """Система уведомлений"""
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()
|