Spaces:
Running
Running
| """Telegram inbound event normalization.""" | |
| from loguru import logger | |
| from telegram import Update | |
| from telegram.ext import ContextTypes | |
| from ..models import IncomingMessage | |
| from ..rendering.telegram_markdown import format_status | |
| from .voice_flow import VoiceNoteRequest, audio_suffix_from_metadata | |
| def telegram_text_message_from_update( | |
| update: Update, | |
| *, | |
| allowed_user_id: str | None, | |
| log_raw_messaging_content: bool, | |
| ) -> IncomingMessage | None: | |
| """Normalize a Telegram text update into an incoming message.""" | |
| if ( | |
| not update.message | |
| or not update.message.text | |
| or not update.effective_user | |
| or not update.effective_chat | |
| ): | |
| return None | |
| user_id = str(update.effective_user.id) | |
| chat_id = str(update.effective_chat.id) | |
| if allowed_user_id and user_id != str(allowed_user_id).strip(): | |
| logger.warning("Unauthorized access attempt from {}", user_id) | |
| return None | |
| message = update.message | |
| message_id = str(message.message_id) | |
| reply_to = ( | |
| str(message.reply_to_message.message_id) if message.reply_to_message else None | |
| ) | |
| thread_id = ( | |
| str(message.message_thread_id) | |
| if getattr(message, "message_thread_id", None) is not None | |
| else None | |
| ) | |
| raw_text = message.text or "" | |
| if log_raw_messaging_content: | |
| text_preview = raw_text[:80] | |
| if len(raw_text) > 80: | |
| text_preview += "..." | |
| logger.info( | |
| "TELEGRAM_MSG: chat_id={} message_id={} reply_to={} text_preview={!r}", | |
| chat_id, | |
| message_id, | |
| reply_to, | |
| text_preview, | |
| ) | |
| else: | |
| logger.info( | |
| "TELEGRAM_MSG: chat_id={} message_id={} reply_to={} text_len={}", | |
| chat_id, | |
| message_id, | |
| reply_to, | |
| len(raw_text), | |
| ) | |
| return IncomingMessage( | |
| text=raw_text, | |
| chat_id=chat_id, | |
| user_id=user_id, | |
| message_id=message_id, | |
| platform="telegram", | |
| reply_to_message_id=reply_to, | |
| message_thread_id=thread_id, | |
| raw_event=update, | |
| ) | |
| def telegram_voice_request_from_update( | |
| update: Update, | |
| context: ContextTypes.DEFAULT_TYPE, | |
| *, | |
| allowed_user_id: str | None, | |
| ) -> VoiceNoteRequest | None: | |
| """Normalize a Telegram voice update into a voice-note request.""" | |
| message = update.message | |
| effective_user = update.effective_user | |
| effective_chat = update.effective_chat | |
| if ( | |
| message is None | |
| or message.voice is None | |
| or effective_user is None | |
| or effective_chat is None | |
| ): | |
| return None | |
| user_id = str(effective_user.id) | |
| if allowed_user_id and user_id != str(allowed_user_id).strip(): | |
| logger.warning("Unauthorized voice access attempt from {}", user_id) | |
| return None | |
| voice = message.voice | |
| chat_id = str(effective_chat.id) | |
| message_id = str(message.message_id) | |
| thread_id = ( | |
| str(message.message_thread_id) | |
| if getattr(message, "message_thread_id", None) is not None | |
| else None | |
| ) | |
| reply_to = ( | |
| str(message.reply_to_message.message_id) if message.reply_to_message else None | |
| ) | |
| async def _download_to(tmp_path) -> None: | |
| tg_file = await context.bot.get_file(voice.file_id) | |
| await tg_file.download_to_drive(custom_path=str(tmp_path)) | |
| async def _reply_text(text: str) -> None: | |
| await message.reply_text(text) | |
| return VoiceNoteRequest( | |
| platform="telegram", | |
| chat_id=chat_id, | |
| user_id=user_id, | |
| message_id=message_id, | |
| raw_event=update, | |
| content_type=voice.mime_type or "audio/ogg", | |
| temp_suffix=audio_suffix_from_metadata(content_type=voice.mime_type), | |
| status_text=format_status("⏳", "Transcribing voice note..."), | |
| status_parse_mode="MarkdownV2", | |
| message_thread_id=thread_id, | |
| reply_to_message_id=reply_to, | |
| username=None, | |
| download_to=_download_to, | |
| reply_text=_reply_text, | |
| ) | |