71 lines
2.2 KiB
Python
71 lines
2.2 KiB
Python
import logging
|
|
|
|
import requests
|
|
|
|
from infrasynth.shared.enums import ChannelType
|
|
from infrasynth.shared.results import Result
|
|
|
|
from .base import BaseChannel
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class TelegramChannel(BaseChannel):
|
|
"""Chat channel that sends via the Telegram Bot API.
|
|
|
|
Configuration keys (read from ``ChannelConfig.config``):
|
|
- ``bot_token`` (required)
|
|
- ``chat_id`` (optional default chat, overridden by recipient)
|
|
- ``timeout``
|
|
"""
|
|
|
|
channel_type = ChannelType.TELEGRAM
|
|
API_BASE = "https://api.telegram.org/bot{token}/{method}"
|
|
|
|
def __init__(self, bot_token=None, chat_id=None, timeout=30):
|
|
self.bot_token = bot_token
|
|
self.chat_id = chat_id
|
|
self.timeout = timeout
|
|
|
|
@classmethod
|
|
def from_config(cls, config: dict) -> "TelegramChannel":
|
|
config = {str(key).lower(): value for key, value in (config or {}).items()}
|
|
return cls(**config)
|
|
|
|
def send(
|
|
self,
|
|
recipient: str,
|
|
subject: str,
|
|
body: str,
|
|
is_html: bool = True,
|
|
attachments=None,
|
|
) -> Result[bool, str]:
|
|
if not self.bot_token:
|
|
return Result.err("Telegram bot token not configured")
|
|
chat_id = recipient or self.chat_id
|
|
if not chat_id:
|
|
return Result.err("No chat_id provided")
|
|
try:
|
|
payload = {
|
|
"chat_id": chat_id,
|
|
"text": body or subject,
|
|
}
|
|
if is_html:
|
|
payload["parse_mode"] = "HTML"
|
|
response = requests.post(
|
|
self.API_BASE.format(token=self.bot_token, method="sendMessage"),
|
|
json=payload,
|
|
timeout=self.timeout,
|
|
)
|
|
except requests.RequestException as exc:
|
|
logger.exception("Telegram send failed for chat %s", chat_id)
|
|
return Result.err(str(exc))
|
|
|
|
data = response.json()
|
|
if not data.get("ok"):
|
|
description = data.get("description") or f"HTTP {response.status_code}"
|
|
return Result.err(f"Telegram error: {description}")
|
|
return Result.ok(True)
|
|
|
|
def health_check(self) -> bool:
|
|
return bool(self.bot_token)
|