49 lines
1.4 KiB
Python
49 lines
1.4 KiB
Python
"""Telegram notification sender. Uses httpx directly — no bot framework needed
|
|
for outbound messages, so we can call it from the scheduler thread without
|
|
needing an event loop."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
|
|
import httpx
|
|
|
|
from app.config import settings
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
TG_API = "https://api.telegram.org"
|
|
|
|
|
|
def send_message(chat_id: str, text: str, parse_mode: str = "HTML") -> bool:
|
|
if not settings.tg_bot_token:
|
|
logger.warning("TG_BOT_TOKEN not set — skipping notification to %s", chat_id)
|
|
return False
|
|
if not chat_id:
|
|
logger.warning("Empty chat_id — skipping notification")
|
|
return False
|
|
url = f"{TG_API}/bot{settings.tg_bot_token}/sendMessage"
|
|
try:
|
|
with httpx.Client(timeout=15.0) as client:
|
|
r = client.post(
|
|
url,
|
|
json={
|
|
"chat_id": chat_id,
|
|
"text": text,
|
|
"parse_mode": parse_mode,
|
|
"disable_web_page_preview": False,
|
|
},
|
|
)
|
|
if r.status_code != 200:
|
|
logger.error("TG send failed: %s %s", r.status_code, r.text)
|
|
return False
|
|
return True
|
|
except httpx.HTTPError as e:
|
|
logger.error("TG send exception: %s", e)
|
|
return False
|
|
|
|
|
|
def notify_admin(text: str) -> None:
|
|
if settings.admin_chat_id:
|
|
send_message(settings.admin_chat_id, text)
|