"""
DOOBAT - Service Telegram
Envoi de messages, documents, gestion des commandes Telegram.
Copyright 2026 Vincent Gallet - Vincesoft
"""
import logging
import os
from configparser import ConfigParser
from typing import Optional
import aiohttp
logger = logging.getLogger("doobat.telegram")
TELEGRAM_API_BASE = "https://api.telegram.org/bot"
class TelegramService:
"""
Service d'envoi de messages et documents Telegram via l'API Bot.
Toutes les méthodes sont asynchrones (aiohttp).
"""
def __init__(self, config: ConfigParser):
self._token = config.get("TELEGRAM", "bot_token", fallback="")
self._base_url = f"{TELEGRAM_API_BASE}{self._token}"
self._timeout = aiohttp.ClientTimeout(total=30)
logger.info("TelegramService initialisé.")
# ── Méthodes d'envoi ───────────────────────────────────────────────────────
async def send_message(
self,
chat_id: int,
text: str,
parse_mode: str = "HTML",
reply_markup: Optional[dict] = None,
) -> bool:
"""
Envoie un message texte à un chat Telegram.
Retourne True si succès.
"""
if not text:
return False
# Telegram limite à 4096 caractères par message
if len(text) > 4096:
text = text[:4090] + "\n..."
payload: dict = {
"chat_id": chat_id,
"text": text,
"parse_mode": parse_mode,
}
if reply_markup:
payload["reply_markup"] = reply_markup
return await self._api_call("sendMessage", payload)
async def send_document(
self,
chat_id: int,
file_path: str,
caption: str = "",
) -> bool:
"""
Envoie un document (PDF) à un chat Telegram.
Retourne True si succès.
"""
if not os.path.exists(file_path):
logger.error(f"Fichier introuvable pour envoi Telegram : {file_path}")
return False
file_size = os.path.getsize(file_path)
if file_size > 50 * 1024 * 1024: # 50 MB limite Telegram
logger.error(f"Fichier trop lourd pour Telegram ({file_size} bytes) : {file_path}")
await self.send_message(
chat_id,
"Le fichier est trop volumineux pour être envoyé via Telegram (> 50 MB).",
)
return False
try:
async with aiohttp.ClientSession(timeout=self._timeout) as session:
with open(file_path, "rb") as f:
form = aiohttp.FormData()
form.add_field("chat_id", str(chat_id))
form.add_field(
"document",
f,
filename=os.path.basename(file_path),
content_type="application/pdf",
)
if caption:
form.add_field("caption", caption[:1024])
form.add_field("parse_mode", "HTML")
url = f"{self._base_url}/sendDocument"
async with session.post(url, data=form) as resp:
response_json = await resp.json()
if resp.status == 200 and response_json.get("ok"):
logger.info(f"Document envoyé à chat_id={chat_id} : {os.path.basename(file_path)}")
return True
else:
logger.error(
f"Erreur Telegram sendDocument chat_id={chat_id} : "
f"{response_json.get('description', 'Erreur inconnue')}"
)
return False
except aiohttp.ClientError as e:
logger.error(f"Erreur réseau envoi document Telegram : {e}")
return False
except OSError as e:
logger.error(f"Erreur lecture fichier PDF : {e}")
return False
async def send_pdf_to_telegram(
self, chat_id: int, pdf_path: str, caption: str = ""
) -> bool:
"""Alias explicite pour l'envoi de PDF."""
return await self.send_document(chat_id, pdf_path, caption)
async def send_typing_action(self, chat_id: int) -> None:
"""Envoie l'indicateur 'en train d'écrire...'."""
await self._api_call("sendChatAction", {"chat_id": chat_id, "action": "typing"})
async def send_upload_action(self, chat_id: int) -> None:
"""Envoie l'indicateur 'en train d'envoyer un document...'."""
await self._api_call(
"sendChatAction",
{"chat_id": chat_id, "action": "upload_document"},
)
# ── Messages prédéfinis ────────────────────────────────────────────────────
async def send_not_authorized(self, chat_id: int) -> None:
"""Message de refus d'accès standard."""
await self.send_message(
chat_id,
"❌ Ce service ne vous est pas disponible.\n\n"
"Votre compte Telegram n'est pas associé à un contact dans notre système.\n"
"Contactez votre administrateur.",
)
async def send_client_not_assigned(self, chat_id: int, commercial_name: str = "") -> None:
"""Message lorsqu'un client n'est pas affecté au commercial."""
msg = "⚠️ Ce client ne vous est pas affecté.\n\n"
if commercial_name:
msg += f"Ce client est géré par : {commercial_name}"
await self.send_message(chat_id, msg)
async def send_client_restricted(self, chat_id: int, commercial_name: str = "") -> None:
"""Message pour un client qui tente d'accéder à une fonction réservée aux commerciaux."""
msg = "ℹ️ Accès restreint.\n\n"
if commercial_name:
msg += f"Pour cette demande, contactez votre commercial : {commercial_name}"
else:
msg += "Pour cette demande, contactez votre commercial."
await self.send_message(chat_id, msg)
async def send_help(self, chat_id: int, is_commercial: bool = False) -> None:
"""Envoie le message d'aide selon le profil."""
if is_commercial:
text = (
"🤖 DOOBAT - Aide Commerciale\n\n"
"Commandes disponibles :\n"
"/start — Démarrer le bot\n"
"/help — Afficher cette aide\n"
"/tarif — Générer le catalogue tarifaire PDF\n"
"/link <numéro> — Associer un contact Odoo\n\n"
"Demandes en langage naturel :\n"
"• \"envoie le tarif\" → PDF catalogue\n"
"• \"stock du produit REF001\" → PDF stock\n"
"• \"stock de tous les produits\" → PDF stock global\n"
"• \"stock de la catégorie Peinture\" → PDF stock catégorie\n"
"• \"factures du client DUPONT\" → PDF documents\n"
"• \"commandes de Martin SAS\" → PDF documents\n"
"• \"devis client TOTO\" → PDF documents\n\n"
"• \"factures fournisseurs\" → PDF fournisseurs global\n"
"• \"factures fournisseur ACME\" → PDF d'un fournisseur\n\n"
"Copyright 2026 Vincent Gallet - Vincesoft"
)
else:
text = (
"🤖 DOOBAT - Aide Client\n\n"
"Commandes disponibles :\n"
"/start — Démarrer le bot\n"
"/help — Afficher cette aide\n"
"/tarif — Recevoir le catalogue tarifaire\n\n"
"Pour toute autre demande, contactez votre commercial.\n\n"
"Copyright 2026 Vincent Gallet - Vincesoft"
)
await self.send_message(chat_id, text)
async def send_start(self, chat_id: int, user_name: str = "") -> None:
"""Message de bienvenue /start."""
greeting = f"Bonjour {user_name} !" if user_name else "Bonjour !"
text = (
f"🤖 Bienvenue sur DOOBAT\n\n"
f"{greeting}\n\n"
"Ce bot vous permet d'accéder à vos informations Odoo directement depuis Telegram.\n\n"
"Pour activer votre acces, partagez votre contact avec le bouton ci-dessous.\n\n"
"Utilisez /help pour voir les commandes disponibles.\n\n"
"Copyright 2026 Vincent Gallet - Vincesoft"
)
reply_markup = {
"keyboard": [[{"text": "Partager mon contact", "request_contact": True}]],
"resize_keyboard": True,
"one_time_keyboard": True,
"input_field_placeholder": "Partagez votre contact pour activer le bot",
}
await self.send_message(chat_id, text, reply_markup=reply_markup)
async def send_error(self, chat_id: int, details: str = "") -> None:
"""Message d'erreur générique."""
msg = "⚠️ Une erreur est survenue.\n\nVeuillez réessayer dans quelques instants."
if details and len(details) < 200:
msg += f"\n\nDétail : {details}"
await self.send_message(chat_id, msg)
async def send_generating_pdf(self, chat_id: int, pdf_type: str = "document") -> None:
"""Indique que le PDF est en cours de génération."""
msgs = {
"tarif": "⏳ Génération du catalogue tarifaire en cours...",
"stock": "⏳ Récupération des informations de stock...",
"client_documents": "⏳ Récupération des documents client...",
"supplier_documents": "⏳ Récupération des documents fournisseur...",
}
msg = msgs.get(pdf_type, "⏳ Génération du document en cours...")
await self.send_message(chat_id, msg)
# ── Méthode API interne ────────────────────────────────────────────────────
async def _api_call(self, method: str, payload: dict) -> bool:
"""Effectue un appel API Telegram. Retourne True si succès."""
url = f"{self._base_url}/{method}"
try:
async with aiohttp.ClientSession(timeout=self._timeout) as session:
async with session.post(url, json=payload) as resp:
response_json = await resp.json()
if resp.status == 200 and response_json.get("ok"):
return True
else:
logger.error(
f"Erreur Telegram API {method} : "
f"{response_json.get('description', 'Erreur inconnue')} "
f"(HTTP {resp.status})"
)
return False
except aiohttp.ClientError as e:
logger.error(f"Erreur réseau Telegram API {method} : {e}")
return False
except Exception as e:
logger.error(f"Erreur inattendue Telegram API {method} : {e}")
return False
# ── Configuration webhook ──────────────────────────────────────────────────
async def set_webhook(self, webhook_url: str, secret_token: str = "") -> bool:
"""Configure le webhook Telegram."""
payload: dict = {"url": webhook_url}
if secret_token:
payload["secret_token"] = secret_token
payload["allowed_updates"] = ["message", "callback_query"]
payload["drop_pending_updates"] = True
result = await self._api_call("setWebhook", payload)
if result:
logger.info(f"Webhook configuré : {webhook_url}")
return result
async def delete_webhook(self) -> bool:
"""Supprime le webhook Telegram."""
return await self._api_call("deleteWebhook", {"drop_pending_updates": True})
async def get_webhook_info(self) -> Optional[dict]:
"""Retourne les informations du webhook actuel."""
url = f"{self._base_url}/getWebhookInfo"
try:
async with aiohttp.ClientSession(timeout=self._timeout) as session:
async with session.get(url) as resp:
data = await resp.json()
return data.get("result")
except Exception as e:
logger.error(f"Erreur getWebhookInfo : {e}")
return None