mirror of
https://github.com/idrainformatica/PecFlow.git
synced 2026-06-16 12:45:42 +02:00
114 lines
3.5 KiB
Python
114 lines
3.5 KiB
Python
"""
|
||
Telegram Bot API – invio messaggi via sendMessage.
|
||
|
||
Usato da NotificationService.test_channel() e dalle notifiche real-time.
|
||
|
||
Formato configurazione canale:
|
||
config: { "chat_id": "-100123456789" } # pubblico
|
||
config_secret: { "bot_token": "123:AAA..." } # cifrato in config_enc
|
||
|
||
API Telegram:
|
||
POST https://api.telegram.org/bot{token}/sendMessage
|
||
Body: { "chat_id": "...", "text": "...", "parse_mode": "MarkdownV2" }
|
||
"""
|
||
|
||
import httpx
|
||
|
||
TELEGRAM_API_BASE = "https://api.telegram.org"
|
||
DEFAULT_TIMEOUT = 10.0 # secondi
|
||
|
||
|
||
class TelegramError(Exception):
|
||
"""Errore durante l'invio di un messaggio Telegram."""
|
||
|
||
def __init__(self, message: str, http_status: int | None = None, api_code: int | None = None):
|
||
super().__init__(message)
|
||
self.http_status = http_status
|
||
self.api_code = api_code
|
||
|
||
|
||
async def send_message(
|
||
bot_token: str,
|
||
chat_id: str,
|
||
text: str,
|
||
parse_mode: str = "HTML",
|
||
disable_web_page_preview: bool = True,
|
||
timeout: float = DEFAULT_TIMEOUT,
|
||
) -> dict:
|
||
"""
|
||
Invia un messaggio a un canale/gruppo/utente Telegram.
|
||
|
||
Args:
|
||
bot_token: Token del bot Telegram (es. "123456789:AAF...")
|
||
chat_id: ID della chat/canale (es. "-100123456789" o "@mychannel")
|
||
text: Testo del messaggio (supporta HTML o MarkdownV2)
|
||
parse_mode: "HTML" (default) | "MarkdownV2" | "" (plain text)
|
||
disable_web_page_preview: Disabilita anteprima link
|
||
timeout: Timeout HTTP in secondi
|
||
|
||
Returns:
|
||
dict con il risultato della API Telegram (result.message_id, ecc.)
|
||
|
||
Raises:
|
||
TelegramError: in caso di errore HTTP o risposta API non-ok
|
||
"""
|
||
url = f"{TELEGRAM_API_BASE}/bot{bot_token}/sendMessage"
|
||
payload: dict = {
|
||
"chat_id": chat_id,
|
||
"text": text,
|
||
}
|
||
if parse_mode:
|
||
payload["parse_mode"] = parse_mode
|
||
if disable_web_page_preview:
|
||
payload["link_preview_options"] = {"is_disabled": True}
|
||
|
||
async with httpx.AsyncClient(timeout=timeout) as client:
|
||
try:
|
||
response = await client.post(url, json=payload)
|
||
except httpx.TimeoutException as exc:
|
||
raise TelegramError(
|
||
f"Timeout nella connessione a Telegram ({timeout}s)"
|
||
) from exc
|
||
except httpx.RequestError as exc:
|
||
raise TelegramError(f"Errore di rete Telegram: {exc}") from exc
|
||
|
||
if response.status_code != 200:
|
||
raise TelegramError(
|
||
f"Telegram API ha risposto con HTTP {response.status_code}: {response.text[:200]}",
|
||
http_status=response.status_code,
|
||
)
|
||
|
||
data = response.json()
|
||
if not data.get("ok"):
|
||
api_code = data.get("error_code")
|
||
description = data.get("description", "Errore sconosciuto")
|
||
raise TelegramError(
|
||
f"Telegram API error {api_code}: {description}",
|
||
http_status=response.status_code,
|
||
api_code=api_code,
|
||
)
|
||
|
||
return data.get("result", {})
|
||
|
||
|
||
async def send_test_message(
|
||
bot_token: str,
|
||
chat_id: str,
|
||
channel_name: str = "PecFlow",
|
||
) -> dict:
|
||
"""
|
||
Invia un messaggio di test formattato al canale configurato.
|
||
|
||
Returns:
|
||
dict result da Telegram (con message_id)
|
||
"""
|
||
from datetime import datetime
|
||
|
||
ts = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
text = (
|
||
f"✅ <b>PecFlow – Test canale Telegram</b>\n\n"
|
||
f"Il canale <b>{channel_name}</b> è configurato correttamente.\n\n"
|
||
f"🕐 <i>{ts}</i>"
|
||
)
|
||
return await send_message(bot_token=bot_token, chat_id=chat_id, text=text, parse_mode="HTML")
|