ettore.dreucci/lagomareGates
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add Telegram notifications

Browse Source
This commit is contained in:
Ettore
2026-05-10 11:58:56 +02:00
parent 54f1ebb62d
commit a5470544a1
7 changed files with 245 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
src/core/database.py
View File

@@ -73,6 +73,15 @@ class AdminUser(Base):
role: Mapped[str] = mapped_column(String, nullable=False, default="admin") # 'admin' | 'manager'
class TelegramConfig(Base):
__tablename__ = "telegram_config"
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
bot_token_enc: Mapped[str] = mapped_column(String, nullable=False) # Fernet-encrypted
chat_id: Mapped[str] = mapped_column(String, nullable=False)
enabled: Mapped[bool] = mapped_column(Boolean, default=True)
def get_db():
db = SessionLocal()
try:

2
src/main.py
View File

@@ -54,6 +54,7 @@ from routers.gates import router as gates_router
from routers.credentials import router as credentials_router
from routers.admins import router as admins_router
from routers.stats import router as stats_router
from routers.telegram import router as telegram_router
# ── App ───────────────────────────────────────────────────────────────────────
@asynccontextmanager
@@ -98,6 +99,7 @@ app.include_router(gates_router)
app.include_router(credentials_router)
app.include_router(admins_router)
app.include_router(stats_router)
app.include_router(telegram_router)
# ── Static / frontend ─────────────────────────────────────────────────────────
@app.get("/favicon.ico", include_in_schema=False)

25
src/routers/gates.py
View File

@@ -1,5 +1,6 @@
import json
import logging
import threading
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Request
@@ -7,15 +8,33 @@ from sqlalchemy.orm import Session
from core.auth import decrypt_secret
from core.config import utcnow
from core.database import ApiCredential, GateAccessLog, GateDB, Keypass, get_db
from core.database import ApiCredential, GateAccessLog, GateDB, Keypass, TelegramConfig, get_db
from core.dependencies import require_admin, require_manager, require_keypass
from core.schemas import GateCreate, GatePublicResponse, GateResponse
from services.gates import call_open_gate
from services.telegram import send_gate_notification
router = APIRouter(tags=["gates"])
logger = logging.getLogger(__name__)
def _notify(db: Session, gate_name: str, opened_by: str, ip: str | None) -> None:
"""Fire a Telegram notification in a background thread if configured and enabled."""
try:
cfg: Optional[TelegramConfig] = db.query(TelegramConfig).first()
if not cfg or not cfg.enabled:
return
token = decrypt_secret(cfg.bot_token_enc)
chat_id = cfg.chat_id
except Exception:
return
threading.Thread(
target=send_gate_notification,
args=(token, chat_id, gate_name, opened_by, ip),
daemon=True,
).start()
# ── Admin: gate CRUD ──────────────────────────────────────────────────────────
@router.get("/api/admin/gates", response_model=list[GateResponse])
@@ -120,6 +139,7 @@ async def admin_open_gate(
raise HTTPException(502, error_msg or "Gate operation failed")
logger.info("Gate opened by admin: gate_id=%d gate=%r caller=%r ip=%r", gate_db.id, gate_db.name, caller["sub"], ip)
_notify(db, gate_db.name, f"[{caller['sub']}]", ip)
return {"success": True, "gate": gate_db.name}
@@ -185,5 +205,6 @@ async def open_gate(
logger.error("Gate open failed: gate_id=%d keypass_id=%d error=%r", gate_db.id, _kp.id, error_msg)
raise HTTPException(502, error_msg or "Gate operation failed")
logger.info("Gate opened by keypass: gate_id=%d gate=%r keypass_id=%d ip=%r", gate_db.id, gate_db.name, _kp.id, ip)
logger.info("Gate opened by keypass: gate_id=%d gate=%r keypass=%r ip=%r", gate_db.id, gate_db.name, f"{_kp.description} ({_kp.code})", ip)
_notify(db, gate_db.name, f"{_kp.description} ({_kp.code})", ip)
return {"success": True, "gate": gate_db.name}

75
src/routers/telegram.py Normal file
View File

@@ -0,0 +1,75 @@
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException
from pydantic import BaseModel
from sqlalchemy.orm import Session
from core.auth import decrypt_secret, encrypt_secret
from core.database import TelegramConfig, get_db
from core.dependencies import require_admin
from services.telegram import send_test_message
router = APIRouter(prefix="/api/admin/telegram", tags=["admin-telegram"])
class TelegramConfigRead(BaseModel):
enabled: bool
chat_id: str
configured: bool # True when a bot token has been saved
class TelegramConfigUpsert(BaseModel):
bot_token: Optional[str] = None # None = keep existing token
chat_id: str
enabled: bool = True
@router.get("", response_model=TelegramConfigRead)
async def get_telegram_config(
db: Session = Depends(get_db),
_: dict = Depends(require_admin),
):
cfg: Optional[TelegramConfig] = db.query(TelegramConfig).first()
if not cfg:
return TelegramConfigRead(enabled=False, chat_id="", configured=False)
return TelegramConfigRead(enabled=cfg.enabled, chat_id=cfg.chat_id, configured=True)
@router.put("", response_model=TelegramConfigRead)
async def upsert_telegram_config(
req: TelegramConfigUpsert,
db: Session = Depends(get_db),
_: dict = Depends(require_admin),
):
cfg: Optional[TelegramConfig] = db.query(TelegramConfig).first()
if cfg:
if req.bot_token:
cfg.bot_token_enc = encrypt_secret(req.bot_token)
cfg.chat_id = req.chat_id
cfg.enabled = req.enabled
else:
if not req.bot_token:
raise HTTPException(422, "bot_token is required for initial setup")
cfg = TelegramConfig(
bot_token_enc=encrypt_secret(req.bot_token),
chat_id=req.chat_id,
enabled=req.enabled,
)
db.add(cfg)
db.commit()
db.refresh(cfg)
return TelegramConfigRead(enabled=cfg.enabled, chat_id=cfg.chat_id, configured=True)
@router.post("/test")
async def test_telegram(
db: Session = Depends(get_db),
_: dict = Depends(require_admin),
):
cfg: Optional[TelegramConfig] = db.query(TelegramConfig).first()
if not cfg:
raise HTTPException(422, "Telegram is not configured yet")
ok, err = send_test_message(decrypt_secret(cfg.bot_token_enc), cfg.chat_id)
if not ok:
raise HTTPException(502, err or "Telegram API error")
return {"success": True}

52
src/services/telegram.py Normal file
View File

@@ -0,0 +1,52 @@
import logging
from datetime import datetime, timezone
import requests
logger = logging.getLogger(__name__)
_TELEGRAM_API = "https://api.telegram.org/bot{token}/sendMessage"
def send_gate_notification(
bot_token: str,
chat_id: str,
gate_name: str,
opened_by: str,
ip: str | None,
) -> None:
"""Send a Telegram message. Runs synchronously; call from a background thread or task."""
ts = datetime.now(timezone.utc).strftime("%d/%m/%Y %H:%M UTC")
text = (
f"🔓 *Gate opened*\n"
f"*Gate:* {gate_name}\n"
f"*By:* `{opened_by}`\n"
f"*Time:* {ts}"
)
if ip:
text += f"\n*IP:* `{ip}`"
try:
resp = requests.post(
_TELEGRAM_API.format(token=bot_token),
json={"chat_id": chat_id, "text": text, "parse_mode": "Markdown"},
timeout=5,
)
if not resp.ok:
logger.warning("Telegram notification failed: %s", resp.text)
except Exception as exc:
logger.warning("Telegram notification error: %s", exc)
def send_test_message(bot_token: str, chat_id: str) -> tuple[bool, str]:
"""Send a test message. Returns (success, error_or_empty)."""
try:
resp = requests.post(
_TELEGRAM_API.format(token=bot_token),
json={"chat_id": chat_id, "text": "✅ Lagomare Gates — Telegram notifications are working!"},
timeout=5,
)
if resp.ok:
return True, ""
return False, resp.json().get("description", resp.text)
except Exception as exc:
return False, str(exc)

37
src/static/admin.html
View File

@@ -119,6 +119,7 @@
<button class="tab-btn admin-only" data-tab="credentials">AVConnect Credentials</button>
<button class="tab-btn" data-tab="stats">Statistics</button>
<button class="tab-btn admin-only" data-tab="admins">Admins</button>
<button class="tab-btn admin-only" data-tab="telegram">Notifications</button>
</nav>
<div class="tab-content">
@@ -286,6 +287,42 @@
</div>
</div>
<!-- ── Telegram / Notifications pane ────────────────────────────────── -->
<div id="tab-telegram" class="tab-pane">
<h3 style="margin-bottom:1rem">Telegram Notifications</h3>
<div class="card" style="max-width:480px">
<p style="color:var(--text-muted);font-size:.9rem;margin-bottom:1.25rem">
Send a message to a Telegram group or chat every time a gate is opened.
Create a bot via <a href="https://t.me/BotFather" target="_blank" rel="noopener" style="color:var(--primary)">@BotFather</a>,
add it to your group, and paste its token and the chat ID below.
</p>
<form id="telegram-form">
<div class="field">
<label for="tg-token">Bot Token</label>
<input id="tg-token" type="password" autocomplete="off"
placeholder="Leave empty to keep current" />
</div>
<div class="field">
<label for="tg-chat-id">Chat / Group ID</label>
<input id="tg-chat-id" type="text" autocomplete="off"
placeholder="e.g. -1001234567890" required />
</div>
<div class="field">
<label style="display:flex;align-items:center;gap:.75rem;cursor:pointer;margin:0">
<input type="checkbox" id="tg-enabled" checked style="width:1.1rem;height:1.1rem;flex-shrink:0;cursor:pointer" />
<span style="font-weight:600">Enable notifications</span>
</label>
</div>
<p id="tg-status" style="font-size:.85rem;color:var(--text-muted);margin-bottom:.75rem"></p>
<p id="tg-error" class="error-msg hidden"></p>
<div style="display:flex;gap:.75rem;flex-wrap:wrap">
<button type="submit" class="btn btn-primary">Save</button>
<button type="button" id="btn-tg-test" class="btn btn-ghost">Send test message</button>
</div>
</form>
</div>
</div>
</div><!-- /.tab-content -->
</div><!-- /#admin-view -->

47
src/static/admin.js
View File

@@ -725,6 +725,52 @@ document.getElementById("chpw-form").addEventListener("submit", async e => {
}
});
// ── Telegram / Notifications ──────────────────────────────────────────────────
async function loadTelegram() {
try {
const cfg = await api("GET", "/api/admin/telegram");
document.getElementById("tg-chat-id").value = cfg.chat_id || "";
document.getElementById("tg-enabled").checked = cfg.enabled;
document.getElementById("tg-status").textContent = cfg.configured
? "Bot token is saved. Leave the token field empty to keep it."
: "Not configured yet. Enter a bot token to get started.";
} catch { /* non-admin: tab hidden anyway */ }
}
document.getElementById("telegram-form").addEventListener("submit", async e => {
e.preventDefault();
const token = document.getElementById("tg-token").value.trim() || null;
const chat_id = document.getElementById("tg-chat-id").value.trim();
const enabled = document.getElementById("tg-enabled").checked;
const errEl = document.getElementById("tg-error");
errEl.classList.add("hidden");
try {
await api("PUT", "/api/admin/telegram", { bot_token: token, chat_id, enabled });
document.getElementById("tg-token").value = "";
showToast("Telegram settings saved");
loadTelegram();
} catch (e) {
errEl.textContent = e.message;
errEl.classList.remove("hidden");
}
});
document.getElementById("btn-tg-test").addEventListener("click", async () => {
const btn = document.getElementById("btn-tg-test");
const errEl = document.getElementById("tg-error");
btn.disabled = true;
errEl.classList.add("hidden");
try {
await api("POST", "/api/admin/telegram/test");
showToast("Test message sent ✓");
} catch (e) {
errEl.textContent = e.message;
errEl.classList.remove("hidden");
} finally {
btn.disabled = false;
}
});
// ── Load all data ─────────────────────────────────────────────────────────────
function loadAllData() {
const isAdmin = _tokenPayload().scope === "admin";
@@ -734,6 +780,7 @@ function loadAllData() {
if (isAdmin) {
loadCredentials();
loadAdmins();
loadTelegram();
}
}