Add RBAC: admin user management, audit log, runtime settings, delete PC

server/notifications.py

@@ -6,11 +6,19 @@ from sqlalchemy.orm import Session
 
 from config import get_settings
 from models import Computer, Notification
+from settings_store import get_setting
 
 settings = get_settings()
 logger = logging.getLogger(__name__)
 
 
+def _int_or_default(value, default):
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
 def create_notification(db: Session, computer: Computer, ntype: str, title: str, message: str) -> Notification:
     n = Notification(
         computer_id=computer.id,
@@ -21,13 +29,17 @@ def create_notification(db: Session, computer: Computer, ntype: str, title: str,
     db.add(n)
     db.commit()
     db.refresh(n)
-    send_external_notifications(n, computer)
+    send_external_notifications(db, n, computer)
     return n
 
 
 def check_offline_computers(db: Session):
     """Mark online computers that missed the threshold as offline and create notifications."""
-    threshold = datetime.now(timezone.utc) - timedelta(minutes=settings.OFFLINE_THRESHOLD_MINUTES)
+    threshold_minutes = _int_or_default(
+        get_setting(db, "OFFLINE_THRESHOLD_MINUTES"),
+        settings.OFFLINE_THRESHOLD_MINUTES,
+    )
+    threshold = datetime.now(timezone.utc) - timedelta(minutes=threshold_minutes)
     stale = (
         db.query(Computer)
         .filter(Computer.status == "online", Computer.last_seen < threshold)
@@ -49,7 +61,7 @@ def check_offline_computers(db: Session):
         if not existing:
             msg = (
                 f"ПК {computer.hostname} не присылал данные более "
-                f"{settings.OFFLINE_THRESHOLD_MINUTES} минут. "
+                f"{threshold_minutes} минут. "
                 f"Последний пользователь: {computer.current_user or '-'}, "
                 f"IP: {computer.current_ip or '-'}"
             )
@@ -57,11 +69,15 @@ def check_offline_computers(db: Session):
     db.commit()
 
 
-def send_external_notifications(notification: Notification, computer: Computer):
-    if settings.NOTIFICATION_WEBHOOK_URL:
+def send_external_notifications(db: Session, notification: Notification, computer: Computer):
+    webhook_url = get_setting(db, "NOTIFICATION_WEBHOOK_URL") or settings.NOTIFICATION_WEBHOOK_URL
+    telegram_token = get_setting(db, "NOTIFICATION_TELEGRAM_BOT_TOKEN") or settings.NOTIFICATION_TELEGRAM_BOT_TOKEN
+    telegram_chat = get_setting(db, "NOTIFICATION_TELEGRAM_CHAT_ID") or settings.NOTIFICATION_TELEGRAM_CHAT_ID
+
+    if webhook_url:
         try:
             requests.post(
-                settings.NOTIFICATION_WEBHOOK_URL,
+                webhook_url,
                 json={
                     "type": notification.type,
                     "title": notification.title,
@@ -75,13 +91,13 @@ def send_external_notifications(notification: Notification, computer: Computer):
         except Exception as e:
             logger.error("Webhook notification failed: %s", e)
 
-    if settings.NOTIFICATION_TELEGRAM_BOT_TOKEN and settings.NOTIFICATION_TELEGRAM_CHAT_ID:
+    if telegram_token and telegram_chat:
         try:
             text = f"*{notification.title}*\n\n{notification.message}"
             requests.post(
-                f"https://api.telegram.org/bot{settings.NOTIFICATION_TELEGRAM_BOT_TOKEN}/sendMessage",
+                f"https://api.telegram.org/bot{telegram_token}/sendMessage",
                 json={
-                    "chat_id": settings.NOTIFICATION_TELEGRAM_CHAT_ID,
+                    "chat_id": telegram_chat,
                     "text": text,
                     "parse_mode": "Markdown",
                 },