Hua/nanobot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Handle media files (voice messages, audio, images, documents) on Telegram Channel

Browse Source
This commit is contained in:
Darye
2026-02-16 16:29:03 +01:00
parent 1207b89adb
commit 23b7e1ef5e
2 changed files with 65 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

64
nanobot/channels/telegram.py
View File

@@ -198,6 +198,18 @@ class TelegramChannel(BaseChannel):
await self._app.shutdown()
self._app = None
def _get_media_type(self, path: str) -> str:
"""Guess media type from file extension."""
path = path.lower()
if path.endswith(('.jpg', '.jpeg', '.png', '.gif', '.webp')):
return "photo"
elif path.endswith('.ogg'):
return "voice"
elif path.endswith(('.mp3', '.m4a', '.wav', '.aac')):
return "audio"
else:
return "document"
async def send(self, msg: OutboundMessage) -> None:
"""Send a message through Telegram."""
if not self._app:
@@ -212,16 +224,50 @@ class TelegramChannel(BaseChannel):
logger.error(f"Invalid chat_id: {msg.chat_id}")
return
for chunk in _split_message(msg.content):
try:
html = _markdown_to_telegram_html(chunk)
await self._app.bot.send_message(chat_id=chat_id, text=html, parse_mode="HTML")
except Exception as e:
logger.warning(f"HTML parse failed, falling back to plain text: {e}")
# Handle media files
if msg.media:
for media_path in msg.media:
try:
await self._app.bot.send_message(chat_id=chat_id, text=chunk)
except Exception as e2:
logger.error(f"Error sending Telegram message: {e2}")
media_type = self._get_media_type(media_path)
# Determine caption (only for first media or if explicitly set,
# but here we keep it simple: content is sent separately if media is present
# to avoid length issues, unless we want to attach it to the first media)
# For simplicity: send media first, then text if present.
# Or: if single media, attach text as caption.
# Let's attach content as caption to the last media if single,
# otherwise send text separately.
with open(media_path, 'rb') as f:
if media_type == "photo":
await self._app.bot.send_photo(chat_id=chat_id, photo=f)
elif media_type == "voice":
await self._app.bot.send_voice(chat_id=chat_id, voice=f)
elif media_type == "audio":
await self._app.bot.send_audio(chat_id=chat_id, audio=f)
else:
await self._app.bot.send_document(chat_id=chat_id, document=f)
except Exception as e:
logger.error(f"Failed to send media {media_path}: {e}")
await self._app.bot.send_message(
chat_id=chat_id,
text=f"[Failed to send file: {media_path}]"
)
# Send text content if present
if msg.content and msg.content != "[empty message]":
for chunk in _split_message(msg.content):
try:
html = _markdown_to_telegram_html(chunk)
await self._app.bot.send_message(chat_id=chat_id, text=html, parse_mode="HTML")
except Exception as e:
logger.warning(f"HTML parse failed, falling back to plain text: {e}")
try:
await self._app.bot.send_message(chat_id=chat_id, text=chunk)
except Exception as e2:
logger.error(f"Error sending Telegram message: {e2}")
async def _on_start(self, update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
"""Handle /start command."""