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

Add media download functionality

Browse Source
This commit is contained in:
John Doe
2026-03-12 06:43:59 +07:00
parent 1eedee0c40
commit 3f799531cc
2 changed files with 210 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files

133
nanobot/channels/telegram.py
View File

@@ -477,21 +477,75 @@ class TelegramChannel(BaseChannel):
+ ("..." if len(text) > TELEGRAM_REPLY_CONTEXT_MAX_LEN else "")
)
return f"[Reply to: {truncated}]"
# Reply has no text/caption; use type placeholder when it has media
# Reply has no text/caption; use type placeholder when it has media.
# Note: replied-to media is not attached to this message, so the agent won't receive it.
if getattr(reply, "photo", None):
return "[Reply to: (image)]"
return "[Reply to: (image — not attached)]"
if getattr(reply, "document", None):
return "[Reply to: (document)]"
return "[Reply to: (document — not attached)]"
if getattr(reply, "voice", None):
return "[Reply to: (voice)]"
return "[Reply to: (voice — not attached)]"
if getattr(reply, "video_note", None) or getattr(reply, "video", None):
return "[Reply to: (video)]"
return "[Reply to: (video — not attached)]"
if getattr(reply, "audio", None):
return "[Reply to: (audio)]"
return "[Reply to: (audio — not attached)]"
if getattr(reply, "animation", None):
return "[Reply to: (animation)]"
return "[Reply to: (animation — not attached)]"
return "[Reply to: (no text)]"
async def _download_message_media(
self, msg, *, add_failure_content: bool = False
) -> tuple[list[str], list[str]]:
"""Download media from a message (current or reply). Returns (media_paths, content_parts)."""
media_file = None
media_type = None
if getattr(msg, "photo", None):
media_file = msg.photo[-1]
media_type = "image"
elif getattr(msg, "voice", None):
media_file = msg.voice
media_type = "voice"
elif getattr(msg, "audio", None):
media_file = msg.audio
media_type = "audio"
elif getattr(msg, "document", None):
media_file = msg.document
media_type = "file"
elif getattr(msg, "video", None):
media_file = msg.video
media_type = "video"
elif getattr(msg, "video_note", None):
media_file = msg.video_note
media_type = "video"
elif getattr(msg, "animation", None):
media_file = msg.animation
media_type = "animation"
if not media_file or not self._app:
return [], []
try:
file = await self._app.bot.get_file(media_file.file_id)
ext = self._get_extension(
media_type,
getattr(media_file, "mime_type", None),
getattr(media_file, "file_name", None),
)
media_dir = get_media_dir("telegram")
file_path = media_dir / f"{media_file.file_id[:16]}{ext}"
await file.download_to_drive(str(file_path))
path_str = str(file_path)
if media_type in ("voice", "audio"):
transcription = await self.transcribe_audio(file_path)
if transcription:
logger.info("Transcribed {}: {}...", media_type, transcription[:50])
return [path_str], [f"[transcription: {transcription}]"]
return [path_str], [f"[{media_type}: {path_str}]"]
return [path_str], [f"[{media_type}: {path_str}]"]
except Exception as e:
logger.warning("Failed to download message media: {}", e)
if add_failure_content:
return [], [f"[{media_type}: download failed]"]
return [], []
async def _ensure_bot_identity(self) -> tuple[int | None, str | None]:
"""Load bot identity once and reuse it for mention/reply checks."""
if self._bot_user_id is not None or self._bot_username is not None:
@@ -612,56 +666,25 @@ class TelegramChannel(BaseChannel):
if message.caption:
content_parts.append(message.caption)
# Handle media files
media_file = None
media_type = None
if message.photo:
media_file = message.photo[-1] # Largest photo
media_type = "image"
elif message.voice:
media_file = message.voice
media_type = "voice"
elif message.audio:
media_file = message.audio
media_type = "audio"
elif message.document:
media_file = message.document
media_type = "file"
# Download media if present
if media_file and self._app:
try:
file = await self._app.bot.get_file(media_file.file_id)
ext = self._get_extension(
media_type,
getattr(media_file, 'mime_type', None),
getattr(media_file, 'file_name', None),
)
media_dir = get_media_dir("telegram")
file_path = media_dir / f"{media_file.file_id[:16]}{ext}"
await file.download_to_drive(str(file_path))
media_paths.append(str(file_path))
if media_type in ("voice", "audio"):
transcription = await self.transcribe_audio(file_path)
if transcription:
logger.info("Transcribed {}: {}...", media_type, transcription[:50])
content_parts.append(f"[transcription: {transcription}]")
else:
content_parts.append(f"[{media_type}: {file_path}]")
else:
content_parts.append(f"[{media_type}: {file_path}]")
logger.debug("Downloaded {} to {}", media_type, file_path)
except Exception as e:
logger.error("Failed to download media: {}", e)
content_parts.append(f"[{media_type}: download failed]")
# Download current message media
current_media_paths, current_media_parts = await self._download_message_media(
message, add_failure_content=True
)
media_paths.extend(current_media_paths)
content_parts.extend(current_media_parts)
if current_media_paths:
logger.debug("Downloaded message media to {}", current_media_paths[0])
# Reply context: include replied-to content; if reply has media, try to attach it
reply = getattr(message, "reply_to_message", None)
reply_ctx = self._extract_reply_context(message)
if reply_ctx is not None:
if reply_ctx is not None and reply is not None:
if "not attached)]" in reply_ctx:
reply_media_paths, reply_media_parts = await self._download_message_media(reply)
if reply_media_paths and reply_media_parts:
reply_ctx = f"[Reply to: {reply_media_parts[0]}]"
media_paths = reply_media_paths + media_paths
logger.debug("Attached replied-to media: {}", reply_media_paths[0])
content_parts.insert(0, reply_ctx)
content = "\n".join(content_parts) if content_parts else "[empty message]"