122 lines
4.0 KiB
Python
122 lines
4.0 KiB
Python
import json
|
|
import logging
|
|
import requests
|
|
from datetime import datetime, timezone, timedelta
|
|
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,
|
|
type=ntype,
|
|
title=title,
|
|
message=message,
|
|
)
|
|
db.add(n)
|
|
db.commit()
|
|
db.refresh(n)
|
|
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_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)
|
|
.all()
|
|
)
|
|
for computer in stale:
|
|
computer.status = "offline"
|
|
db.add(computer)
|
|
# avoid duplicate notifications for the same offline event
|
|
existing = (
|
|
db.query(Notification)
|
|
.filter(
|
|
Notification.computer_id == computer.id,
|
|
Notification.type == "offline",
|
|
Notification.is_read == False,
|
|
)
|
|
.first()
|
|
)
|
|
if not existing:
|
|
msg = (
|
|
f"ПК {computer.hostname} не присылал данные более "
|
|
f"{threshold_minutes} минут. "
|
|
f"Последний пользователь: {computer.current_user or '-'}, "
|
|
f"IP: {computer.current_ip or '-'}"
|
|
)
|
|
create_notification(db, computer, "offline", f"ПК {computer.hostname} офлайн", msg)
|
|
db.commit()
|
|
|
|
|
|
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(
|
|
webhook_url,
|
|
json={
|
|
"type": notification.type,
|
|
"title": notification.title,
|
|
"message": notification.message,
|
|
"hostname": computer.hostname,
|
|
"user": computer.current_user,
|
|
"ip": computer.current_ip,
|
|
},
|
|
timeout=10,
|
|
)
|
|
except Exception as e:
|
|
logger.error("Webhook notification failed: %s", e)
|
|
|
|
if telegram_token and telegram_chat:
|
|
try:
|
|
text = f"*{notification.title}*\n\n{notification.message}"
|
|
requests.post(
|
|
f"https://api.telegram.org/bot{telegram_token}/sendMessage",
|
|
json={
|
|
"chat_id": telegram_chat,
|
|
"text": text,
|
|
"parse_mode": "Markdown",
|
|
},
|
|
timeout=10,
|
|
)
|
|
except Exception as e:
|
|
logger.error("Telegram notification failed: %s", e)
|
|
|
|
|
|
def mark_notification_read(db: Session, notification_id: int) -> bool:
|
|
n = db.query(Notification).filter(Notification.id == notification_id).first()
|
|
if not n:
|
|
return False
|
|
n.is_read = True
|
|
n.read_at = datetime.now(timezone.utc)
|
|
db.commit()
|
|
return True
|
|
|
|
|
|
def get_unread_count(db: Session) -> int:
|
|
return db.query(Notification).filter(Notification.is_read == False).count()
|