add typing simulation

assistant_bot/telegram_utils.py

@@ -1,10 +1,13 @@
+import asyncio
 import logging
 import re
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager, suppress
 from uuid import uuid4
 from zoneinfo import ZoneInfo
 
 from telegram import Bot, InlineQueryResultArticle, InputTextMessageContent, Message, Update
-from telegram.constants import ParseMode
+from telegram.constants import ChatAction, ParseMode
 from telegram.error import BadRequest
 from telegram.ext import ContextTypes
 from telegramify_markdown import (
@@ -24,6 +27,7 @@ from .storage import AssistantStorage
 logger = logging.getLogger(__name__)
 
 FORMATTING_FALLBACK_NOTICE = "⚠️ Не удалось применить форматирование."
+TYPING_REFRESH_INTERVAL_SECONDS = 4.0
 _MARKDOWN_SPECIAL_CHARS = re.compile(r"([\\`*_[\]{}()#+\-.!|>~])")
 _RENDER_CONFIG = RenderConfig()
 _RENDER_CONFIG.markdown_symbol.heading_level_1 = ""
@@ -32,6 +36,51 @@ _RENDER_CONFIG.markdown_symbol.heading_level_3 = ""
 _RENDER_CONFIG.markdown_symbol.heading_level_4 = ""
 
 
+async def _refresh_typing(
+    bot: Bot,
+    chat_id: int,
+    interval: float,
+) -> None:
+    while True:
+        await asyncio.sleep(interval)
+        try:
+            await bot.send_chat_action(
+                chat_id=chat_id,
+                action=ChatAction.TYPING,
+            )
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            logger.warning("Failed to refresh Telegram typing action", exc_info=True)
+
+
+@asynccontextmanager
+async def typing_action(
+    bot: Bot,
+    chat_id: int,
+    interval: float = TYPING_REFRESH_INTERVAL_SECONDS,
+) -> AsyncIterator[None]:
+    """Keep Telegram's typing indicator active while work is in progress."""
+    try:
+        await bot.send_chat_action(
+            chat_id=chat_id,
+            action=ChatAction.TYPING,
+        )
+    except Exception:
+        logger.warning("Failed to send Telegram typing action", exc_info=True)
+
+    refresh_task = asyncio.create_task(
+        _refresh_typing(bot, chat_id, interval),
+        name=f"telegram-typing-{chat_id}",
+    )
+    try:
+        yield
+    finally:
+        refresh_task.cancel()
+        with suppress(asyncio.CancelledError):
+            await refresh_task
+
+
 def escape_markdown_text(value: object) -> str:
     """Escape dynamic text before embedding it into ordinary Markdown."""
     return _MARKDOWN_SPECIAL_CHARS.sub(r"\\\1", str(value))