dinlo
f624fee5e0
- SSH/SFTP управление без агентов - Пакеты (apt), Docker, файлы, systemd, ufw, cron - Плейбуки, групповые операции, планировщик - Бэкапы, отчёты, Portainer, Proxmox VE, Gitea Deploy - Терминал, Telegram-уведомления, тёмная/светлая тема
92 lines
3.0 KiB
Python
92 lines
3.0 KiB
Python
"""Telegram-уведомления с антиспамом."""
|
|
from __future__ import annotations
|
|
|
|
import time
|
|
from typing import Optional
|
|
|
|
import httpx
|
|
|
|
from app.db.database import get_session
|
|
from app.db.models import Setting
|
|
|
|
_antispam: dict[str, float] = {}
|
|
THROTTLE_S = 600 # не чаще 1 раза в 10 мин на ключ
|
|
|
|
|
|
def _get(key: str) -> str:
|
|
with get_session() as s:
|
|
row = s.get(Setting, key)
|
|
return row.value if row else ""
|
|
|
|
|
|
def _set(key: str, value: str) -> None:
|
|
with get_session() as s:
|
|
row = s.get(Setting, key) or Setting(key=key)
|
|
row.value = value
|
|
s.add(row)
|
|
s.commit()
|
|
|
|
|
|
def get_config() -> dict:
|
|
return {
|
|
"token": _get("tg_token"),
|
|
"chat_id": _get("tg_chat_id"),
|
|
"enabled": _get("tg_enabled") == "1",
|
|
"on_alert": _get("tg_on_alert") == "1",
|
|
"on_task_done": _get("tg_on_task_done") == "1",
|
|
}
|
|
|
|
|
|
def save_config(token: str, chat_id: str, enabled: bool,
|
|
on_alert: bool, on_task_done: bool) -> None:
|
|
_set("tg_token", token)
|
|
_set("tg_chat_id", chat_id)
|
|
_set("tg_enabled", "1" if enabled else "0")
|
|
_set("tg_on_alert", "1" if on_alert else "0")
|
|
_set("tg_on_task_done", "1" if on_task_done else "0")
|
|
|
|
|
|
async def send(text: str, key: Optional[str] = None, force: bool = False) -> bool:
|
|
"""Отправить сообщение. key — для антиспама (например 'disk:server1')."""
|
|
cfg = get_config()
|
|
if not cfg["enabled"] or not cfg["token"] or not cfg["chat_id"]:
|
|
return False
|
|
if key and not force:
|
|
last = _antispam.get(key, 0)
|
|
if time.time() - last < THROTTLE_S:
|
|
return False
|
|
_antispam[key] = time.time()
|
|
url = f"https://api.telegram.org/bot{cfg['token']}/sendMessage"
|
|
try:
|
|
async with httpx.AsyncClient(timeout=15) as cl:
|
|
r = await cl.post(url, json={
|
|
"chat_id": cfg["chat_id"],
|
|
"text": text,
|
|
"parse_mode": "HTML",
|
|
"disable_web_page_preview": True,
|
|
})
|
|
return r.status_code == 200
|
|
except Exception:
|
|
return False
|
|
|
|
|
|
async def test() -> tuple[bool, str]:
|
|
cfg = get_config()
|
|
if not cfg["token"] or not cfg["chat_id"]:
|
|
return False, "заполните token и chat_id"
|
|
ok = await send("✅ Debian Manager: тестовое уведомление", force=True)
|
|
return ok, "отправлено" if ok else "ошибка отправки (проверьте token/chat_id)"
|
|
|
|
|
|
async def alert(server_name: str, text: str) -> None:
|
|
cfg = get_config()
|
|
if cfg["enabled"] and cfg["on_alert"]:
|
|
await send(f"⚠️ <b>{server_name}</b>: {text}", key=f"alert:{server_name}:{text[:30]}")
|
|
|
|
|
|
async def task_done(server_name: str, title: str, status: str) -> None:
|
|
cfg = get_config()
|
|
if cfg["enabled"] and cfg["on_task_done"]:
|
|
icon = "✅" if status == "success" else "❌"
|
|
await send(f"{icon} <b>{server_name}</b>: {title} — {status}", force=True)
|