You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
73 lines
2.8 KiB
Python
73 lines
2.8 KiB
Python
"""Telegram notification client — sends messages via Bot API."""
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import time
|
|
from typing import Optional
|
|
|
|
from loguru import logger
|
|
|
|
TELEGRAM_API_URL = "https://api.telegram.org/bot{token}/sendMessage"
|
|
MAX_TEXT_LENGTH = 4096 # Telegram message limit
|
|
_RETRY_DELAYS = (5, 10) # seconds between retries (2 retries = 3 total attempts)
|
|
|
|
|
|
class TelegramClient:
|
|
def __init__(
|
|
self,
|
|
bot_token: Optional[str] = None,
|
|
chat_id: Optional[str] = None,
|
|
):
|
|
self.bot_token = bot_token or os.environ.get("TELEGRAM_BOT_TOKEN", "")
|
|
raw_ids = chat_id or os.environ.get("TELEGRAM_CHAT_ID", "")
|
|
# Support comma-separated multiple chat IDs: "123456789,987654321"
|
|
self.chat_ids: list[str] = [cid.strip() for cid in raw_ids.split(",") if cid.strip()]
|
|
|
|
def is_configured(self) -> bool:
|
|
return bool(self.bot_token and self.chat_ids)
|
|
|
|
def send_message(self, text: str) -> bool:
|
|
"""Send a message to all configured chats. Returns True if all succeeded."""
|
|
if not self.is_configured():
|
|
logger.warning("Telegram not configured (missing BOT_TOKEN or CHAT_ID)")
|
|
return False
|
|
|
|
if len(text) > MAX_TEXT_LENGTH:
|
|
text = text[: MAX_TEXT_LENGTH - 20] + "\n\n[...truncated]"
|
|
|
|
import httpx
|
|
all_ok = True
|
|
for chat_id in self.chat_ids:
|
|
ok = self._send_to(httpx, chat_id, text)
|
|
if not ok:
|
|
all_ok = False
|
|
return all_ok
|
|
|
|
def _send_to(self, httpx, chat_id: str, text: str) -> bool:
|
|
"""Send to a single chat ID with retry. Returns True on success."""
|
|
for attempt, delay in enumerate([0] + list(_RETRY_DELAYS), start=1):
|
|
if delay:
|
|
logger.debug(f"Telegram retry {attempt} for {chat_id} in {delay}s...")
|
|
time.sleep(delay)
|
|
try:
|
|
response = httpx.post(
|
|
TELEGRAM_API_URL.format(token=self.bot_token),
|
|
json={
|
|
"chat_id": chat_id,
|
|
"text": text,
|
|
"parse_mode": "HTML",
|
|
},
|
|
timeout=30.0,
|
|
)
|
|
if response.status_code == 200 and response.json().get("ok"):
|
|
logger.info(f"Telegram message sent to {chat_id}")
|
|
return True
|
|
else:
|
|
logger.error(f"Telegram API error for {chat_id}: {response.text[:200]}")
|
|
return False # API errors won't be fixed by retrying
|
|
except Exception as e:
|
|
logger.warning(f"Telegram send attempt {attempt} to {chat_id} failed: {e}")
|
|
|
|
logger.error(f"Telegram send to {chat_id} failed after all retries")
|
|
return False
|