fix: implement true Telegram Rich Messages via LinkPreviewOptions

This commit is contained in:
Your Name
2026-07-22 16:53:33 +05:00
parent acc7e8657d
commit 459f281a5d
+53 -2
View File
@@ -12,7 +12,7 @@ from zoneinfo import ZoneInfo
from aiogram import Bot from aiogram import Bot
from aiogram.exceptions import TelegramRetryAfter from aiogram.exceptions import TelegramRetryAfter
from aiogram.types import FSInputFile, InputMediaPhoto, InputMediaVideo from aiogram.types import FSInputFile, InputMediaPhoto, InputMediaVideo, LinkPreviewOptions
from loguru import logger from loguru import logger
from ..config import settings from ..config import settings
@@ -356,6 +356,56 @@ class TelegramPoster:
async def send_post(self, post: dict[str, Any]) -> list[int]: async def send_post(self, post: dict[str, Any]) -> list[int]:
text = self.build_text(post) text = self.build_text(post)
media_rows = await self.load_media(int(post["id"])) media_rows = await self.load_media(int(post["id"]))
# If no media, just send text
if not media_rows:
return await self.send_text(text)
# Telegram Rich Message (Single message, block layout, up to 4096 chars)
# We use LinkPreviewOptions with the first media URL to show it above the text.
first_url = None
for row in media_rows:
url = str(row.get("original_url") or "").strip()
if url.startswith("http"):
first_url = url
break
if first_url:
# Send as a Rich Message with Link Preview
message = await self.tg_retry(
lambda: self.bot.send_message(
text=text,
link_preview_options=LinkPreviewOptions(
url=first_url,
show_above_text=True
),
**self.chat_kwargs()
)
)
message_ids = [int(message.message_id)]
# If it's an album (multiple media), we can send the rest as a media group without caption
if len(media_rows) > 1:
media_items = []
for item in media_rows[1:]:
media = self.media_value(item)
if not media: continue
media_type = str(item.get("media_type") or "")
if media_type == "photo":
media_items.append({"type": "photo", "media": media})
elif media_type == "video":
media_items.append({"type": "video", "media": media})
rest = media_items
for start in range(0, len(rest), self.media_group_max_items):
chunk = rest[start : start + self.media_group_max_items]
message_ids.extend(await self.send_media_items(chunk, caption=None))
if self.send_delay_sec:
await asyncio.sleep(self.send_delay_sec)
return message_ids
# Fallback if no valid HTTP URL was found for Rich Message
media_items = [] media_items = []
for item in media_rows: for item in media_rows:
media = self.media_value(item) media = self.media_value(item)
@@ -366,12 +416,13 @@ class TelegramPoster:
media_items.append({"type": "photo", "media": media}) media_items.append({"type": "photo", "media": media})
elif media_type == "video": elif media_type == "video":
media_items.append({"type": "video", "media": media}) media_items.append({"type": "video", "media": media})
message_ids: list[int] = []
if not media_items: if not media_items:
return await self.send_text(text) return await self.send_text(text)
first_caption = text[: self.caption_limit] first_caption = text[: self.caption_limit]
first = media_items[: self.media_group_max_items] first = media_items[: self.media_group_max_items]
message_ids = []
message_ids.extend(await self.send_media_items(first, first_caption)) message_ids.extend(await self.send_media_items(first, first_caption))
if self.send_delay_sec: if self.send_delay_sec:
await asyncio.sleep(self.send_delay_sec) await asyncio.sleep(self.send_delay_sec)