"""Telegram front end for the support assistant.""" from __future__ import annotations import asyncio import contextlib import html import logging import re from collections import defaultdict from telegram import Update from telegram.constants import ChatAction, MessageLimit from telegram.error import TelegramError from telegram.ext import Application, CommandHandler, ContextTypes, MessageHandler, filters from .assistant import StoreAssistant from .config import Settings from .ratelimit import RateLimiter logger = logging.getLogger(__name__) WELCOME = ( "مرحبا بيك في {store} 👋\n" "نقدر نعاونك في: متابعة الكوموند، الأسعار والمقاسات، ثمن التوصيل، وسياسة الاسترجاع.\n\n" "Bienvenue chez {store} ! Je peux vous aider pour le suivi de commande, les prix et tailles, " "les frais de livraison et les retours.\n\n" "/reset — بداية محادثة جديدة · nouvelle conversation" ) HELP = ( "أمثلة · Exemples :\n" "• وين راهي الكوموند NS-10241؟\n" "• Kayen sneakers taille 42 ?\n" "• Combien coûte la livraison à Oran ?\n" "• نحب نهدر مع واحد من الفريق" ) RESET_DONE = "تم ✅ بدينا محادثة جديدة. · C'est fait, nouvelle conversation." TEXT_ONLY = "نقرا غير الرسائل المكتوبة. · Je ne lis que les messages texte." TOO_LONG = "الرسالة طويلة بزاف، اختصرها من فضلك. · Message trop long, merci de le raccourcir." SLOW_DOWN = "شوية بشوية 🙏 عاود بعد {seconds} ثانية. · Merci de patienter {seconds} s." TYPING_REFRESH_S = 4.0 _HEADING = re.compile(r"^#{1,6}\s+", re.MULTILINE) def to_plain_text(text: str) -> str: """Remove the Markdown that models sometimes add anyway; Telegram would show it as raw symbols.""" return _HEADING.sub("", text.replace("**", "")) def split_message(text: str, limit: int = MessageLimit.MAX_TEXT_LENGTH) -> list[str]: """Split text into Telegram-sized chunks, preferring paragraph, line and word boundaries.""" text = text.strip() chunks: list[str] = [] while len(text) > limit: cut = -1 for sep in ("\n\n", "\n", " "): cut = text.rfind(sep, 0, limit + 1) if cut > 0: break if cut <= 0: cut = limit chunks.append(text[:cut].rstrip()) text = text[cut:].lstrip() if text: chunks.append(text) return chunks class SupportBot: def __init__(self, settings: Settings, assistant: StoreAssistant, store_name: str) -> None: self._settings = settings self._assistant = assistant self._store_name = store_name self._limiter = RateLimiter(settings.rate_limit_per_minute) # One lock per chat so a customer's messages are answered in order. self._chat_locks: defaultdict[int, asyncio.Lock] = defaultdict(asyncio.Lock) def build_application(self) -> Application: if not self._settings.telegram_bot_token: raise RuntimeError("TELEGRAM_BOT_TOKEN is required to run the bot") app = Application.builder().token(self._settings.telegram_bot_token).concurrent_updates(True).build() private = filters.ChatType.PRIVATE app.add_handler(CommandHandler("start", self.on_start, filters=private)) app.add_handler(CommandHandler("help", self.on_help, filters=private)) app.add_handler(CommandHandler("reset", self.on_reset, filters=private)) app.add_handler(CommandHandler("id", self.on_id, filters=private)) app.add_handler(MessageHandler(private & filters.TEXT & ~filters.COMMAND, self.on_text)) app.add_handler(MessageHandler(private & ~filters.TEXT & ~filters.COMMAND, self.on_other)) app.add_error_handler(self.on_error) return app async def on_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if update.effective_message: await update.effective_message.reply_text(WELCOME.format(store=self._store_name)) async def on_help(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if update.effective_message: await update.effective_message.reply_text(HELP) async def on_id(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """Show the chat id, so the store owner can set ADMIN_CHAT_ID.""" if update.effective_chat and update.effective_message: await update.effective_message.reply_text(f"Chat ID: {update.effective_chat.id}") async def on_reset(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if update.effective_chat and update.effective_message: self._assistant.reset(update.effective_chat.id) await update.effective_message.reply_text(RESET_DONE) async def on_other(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if update.effective_message: await update.effective_message.reply_text(TEXT_ONLY) async def on_text(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: message, chat, user = update.effective_message, update.effective_chat, update.effective_user if message is None or chat is None or not message.text: return text = message.text.strip() if len(text) > self._settings.max_input_chars: await message.reply_text(TOO_LONG) return wait = self._limiter.check(user.id if user else chat.id) if wait > 0: await message.reply_text(SLOW_DOWN.format(seconds=max(1, round(wait)))) return username = user.username if user else None async with self._chat_locks[chat.id]: typing = asyncio.create_task(self._keep_typing(context, chat.id)) try: reply = await self._assistant.reply(chat.id, text, username) finally: typing.cancel() with contextlib.suppress(asyncio.CancelledError): await typing for chunk in split_message(to_plain_text(reply.text)): await message.reply_text(chunk) for ticket_id in reply.ticket_ids: await self._notify_admin(context, ticket_id, chat.id, username, text) async def _keep_typing(self, context: ContextTypes.DEFAULT_TYPE, chat_id: int) -> None: while True: with contextlib.suppress(TelegramError): await context.bot.send_chat_action(chat_id=chat_id, action=ChatAction.TYPING) await asyncio.sleep(TYPING_REFRESH_S) async def _notify_admin( self, context: ContextTypes.DEFAULT_TYPE, ticket_id: int, chat_id: int, username: str | None, last_message: str ) -> None: admin = self._settings.admin_chat_id if admin is None: logger.info("ticket %s opened for chat %s (no ADMIN_CHAT_ID set)", ticket_id, chat_id) return who = f"@{html.escape(username)}" if username else f"chat {chat_id}" body = f"🎫 Ticket #{ticket_id}\nCustomer: {who}\nLast message: {html.escape(last_message[:500])}" try: await context.bot.send_message(chat_id=admin, text=body, parse_mode="HTML") except TelegramError: logger.exception("could not notify admin about ticket %s", ticket_id) async def on_error(self, update: object, context: ContextTypes.DEFAULT_TYPE) -> None: logger.error("unhandled error while processing an update", exc_info=context.error)