Bot de support en Python : code source

1"""Telegram front end for the support assistant."""23from __future__ import annotations45import asyncio6import contextlib7import html8import logging9import re10from collections import defaultdict1112from telegram import Update13from telegram.constants import ChatAction, MessageLimit14from telegram.error import TelegramError15from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters1617from .assistant import StoreAssistant18from .config import Settings19from .ratelimit import RateLimiter2021logger = logging.getLogger(__name__)2223WELCOME = (24    "مرحبا بيك في {store} 👋\n"25    "نقدر نعاونك في: متابعة الكوموند، الأسعار والمقاسات، ثمن التوصيل، وسياسة الاسترجاع.\n\n"26    "Bienvenue chez {store} ! Je peux vous aider pour le suivi de commande, les prix et tailles, "27    "les frais de livraison et les retours.\n\n"28    "/reset — بداية محادثة جديدة · nouvelle conversation"29)30HELP = (31    "أمثلة · Exemples :\n"32    "• وين راهي الكوموند NS-10241؟\n"33    "• Kayen sneakers taille 42 ?\n"34    "• Combien coûte la livraison à Oran ?\n"35    "• نحب نهدر مع واحد من الفريق"36)37RESET_DONE = "تم ✅ بدينا محادثة جديدة. · C'est fait, nouvelle conversation."38TEXT_ONLY = "نقرا غير الرسائل المكتوبة. · Je ne lis que les messages texte."39TOO_LONG = "الرسالة طويلة بزاف، اختصرها من فضلك. · Message trop long, merci de le raccourcir."40SLOW_DOWN = "شوية بشوية 🙏 عاود بعد {seconds} ثانية. · Merci de patienter {seconds} s."4142TYPING_REFRESH_S = 4.0434445_HEADING = re.compile(r"^#{1,6}\s+", re.MULTILINE)464748def to_plain_text(text: str) -> str:49    """Remove the Markdown that models sometimes add anyway; Telegram would show it as raw symbols."""50    return _HEADING.sub("", text.replace("**", ""))515253def split_message(text: str, limit: int = MessageLimit.MAX_TEXT_LENGTH) -> list[str]:54    """Split text into Telegram-sized chunks, preferring paragraph, line and word boundaries."""55    text = text.strip()56    chunks: list[str] = []57    while len(text) > limit:58        cut = -159        for sep in ("\n\n", "\n", " "):60            cut = text.rfind(sep, 0, limit + 1)61            if cut > 0:62                break63        if cut <= 0:64            cut = limit65        chunks.append(text[:cut].rstrip())66        text = text[cut:].lstrip()67    if text:68        chunks.append(text)69    return chunks707172class SupportBot:73    def __init__(self, settings: Settings, assistant: StoreAssistant, store_name: str) -> None:74        self._settings = settings75        self._assistant = assistant76        self._store_name = store_name77        self._limiter = RateLimiter(settings.rate_limit_per_minute)78        # One lock per chat so a customer's messages are answered in order.79        self._chat_locks: defaultdict[int, asyncio.Lock] = defaultdict(asyncio.Lock)8081    def build_application(self) -> Application:82        if not self._settings.telegram_bot_token:83            raise RuntimeError("TELEGRAM_BOT_TOKEN is required to run the bot")84        app = Application.builder().token(self._settings.telegram_bot_token).concurrent_updates(True).build()85        private = filters.ChatType.PRIVATE86        app.add_handler(CommandHandler("start", self.on_start, filters=private))87        app.add_handler(CommandHandler("help", self.on_help, filters=private))88        app.add_handler(CommandHandler("reset", self.on_reset, filters=private))89        app.add_handler(CommandHandler("id", self.on_id, filters=private))90        app.add_handler(MessageHandler(private & filters.TEXT & ~filters.COMMAND, self.on_text))91        app.add_handler(MessageHandler(private & ~filters.TEXT & ~filters.COMMAND, self.on_other))92        app.add_error_handler(self.on_error)93        return app9495    async def on_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:96        if update.effective_message:97            await update.effective_message.reply_text(WELCOME.format(store=self._store_name))9899    async def on_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:100        if update.effective_message:101            await update.effective_message.reply_text(HELP)102103    async def on_id(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:104        """Show the chat id, so the store owner can set ADMIN_CHAT_ID."""105        if update.effective_chat and update.effective_message:106            await update.effective_message.reply_text(f"Chat ID: {update.effective_chat.id}")107108    async def on_reset(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:109        if update.effective_chat and update.effective_message:110            self._assistant.reset(update.effective_chat.id)111            await update.effective_message.reply_text(RESET_DONE)112113    async def on_other(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:114        if update.effective_message:115            await update.effective_message.reply_text(TEXT_ONLY)116117    async def on_text(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:118        message, chat, user = update.effective_message, update.effective_chat, update.effective_user119        if message is None or chat is None or not message.text:120            return121122        text = message.text.strip()123        if len(text) > self._settings.max_input_chars:124            await message.reply_text(TOO_LONG)125            return126        wait = self._limiter.check(user.id if user else chat.id)127        if wait > 0:128            await message.reply_text(SLOW_DOWN.format(seconds=max(1, round(wait))))129            return130131        username = user.username if user else None132        async with self._chat_locks[chat.id]:133            typing = asyncio.create_task(self._keep_typing(context, chat.id))134            try:135                reply = await self._assistant.reply(chat.id, text, username)136            finally:137                typing.cancel()138                with contextlib.suppress(asyncio.CancelledError):139                    await typing140141        for chunk in split_message(to_plain_text(reply.text)):142            await message.reply_text(chunk)143144        for ticket_id in reply.ticket_ids:145            await self._notify_admin(context, ticket_id, chat.id, username, text)146147    async def _keep_typing(self, context: ContextTypes.DEFAULT_TYPE, chat_id: int) -> None:148        while True:149            with contextlib.suppress(TelegramError):150                await context.bot.send_chat_action(chat_id=chat_id, action=ChatAction.TYPING)151            await asyncio.sleep(TYPING_REFRESH_S)152153    async def _notify_admin(154        self, context: ContextTypes.DEFAULT_TYPE, ticket_id: int, chat_id: int, username: str | None, last_message: str155    ) -> None:156        admin = self._settings.admin_chat_id157        if admin is None:158            logger.info("ticket %s opened for chat %s (no ADMIN_CHAT_ID set)", ticket_id, chat_id)159            return160        who = f"@{html.escape(username)}" if username else f"chat {chat_id}"161        body = f"🎫 <b>Ticket #{ticket_id}</b>\nCustomer: {who}\nLast message: {html.escape(last_message[:500])}"162        try:163            await context.bot.send_message(chat_id=admin, text=body, parse_mode="HTML")164        except TelegramError:165            logger.exception("could not notify admin about ticket %s", ticket_id)166167    async def on_error(self, update: object, context: ContextTypes.DEFAULT_TYPE) -> None:168        logger.error("unhandled error while processing an update", exc_info=context.error)