Use Telegram rich messages for media posts

src/vk_parser_app/workers/tg_poster.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import asyncio
+import aiohttp
 import json
 import os
 import re
@@ -12,7 +13,7 @@ from zoneinfo import ZoneInfo
 
 from aiogram import Bot
 from aiogram.exceptions import TelegramRetryAfter
-from aiogram.types import FSInputFile, InputMediaPhoto, InputMediaVideo, LinkPreviewOptions
+from aiogram.types import FSInputFile, InputMediaPhoto, InputMediaVideo
 from loguru import logger
 
 from ..config import settings
@@ -29,6 +30,8 @@ from ..text_utils import build_publication_text
 
 LOCAL_TZ = ZoneInfo("Asia/Yekaterinburg")
 MAX_MEDIA_GROUP = 10
+MAX_RICH_MEDIA = 50
+MAX_RICH_TEXT = 32768
 PROJECT_ROOT = Path(__file__).resolve().parents[3]
 UPLOAD_ROOT = PROJECT_ROOT / "uploads"
 
@@ -85,6 +88,10 @@ def local_upload_path(url: str) -> Path | None:
     return path if path.is_file() else None
 
 
+class RichMessageUnavailable(RuntimeError):
+    pass
+
+
 class TelegramPoster:
     def __init__(self) -> None:
         self.pool = None
@@ -353,59 +360,77 @@ class TelegramPoster:
         messages = await self.tg_retry(lambda: self.bot.send_media_group(media=group, **self.chat_kwargs()))
         return [int(message.message_id) for message in messages]
 
-    async def send_post(self, post: dict[str, Any]) -> list[int]:
+    def rich_media_value(self, item: dict[str, Any]) -> str | None:
-        text = self.build_text(post)
+        stored = str(item.get("tg_file_id") or item.get("storage_attachment_id") or "").strip()
-        media_rows = await self.load_media(int(post["id"]))
+        return stored or None
-        
-        # 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)
+    def rich_text_html(self, text: str) -> str:
-        # We use LinkPreviewOptions with the first media URL to show it above the text.
+        paragraphs = []
-        first_url = None
+        for paragraph in str(text or "").strip().split("\n\n"):
-        for row in media_rows:
+            body = "<br/>".join(line for line in paragraph.splitlines() if line.strip())
-            url = str(row.get("original_url") or "").strip()
+            if body:
-            if url.startswith("http"):
+                paragraphs.append(f"<p>{body}</p>")
-                first_url = url
+        return "\n".join(paragraphs)
-                break
 
-        if first_url:
+    def build_rich_message(self, text: str, media_rows: list[dict[str, Any]]) -> dict[str, Any] | None:
-            # Send as a Rich Message with Link Preview
+        rich_media = []
-            message = await self.tg_retry(
+        media_tags = []
-                lambda: self.bot.send_message(
+        for idx, item in enumerate(media_rows[:MAX_RICH_MEDIA]):
-                    text=text,
+            media_type = str(item.get("media_type") or "")
-                    link_preview_options=LinkPreviewOptions(
+            if media_type not in ("photo", "video"):
-                        url=first_url,
+                continue
-                        show_above_text=True
+            media = self.rich_media_value(item)
-                    ),
+            if not media:
-                    **self.chat_kwargs()
+                return None
-                )
+            media_id = f"m{idx}"
-            )
+            input_type = "photo" if media_type == "photo" else "video"
-            message_ids = [int(message.message_id)]
+            rich_media.append({"id": media_id, "media": {"type": input_type, "media": media}})
-            
+            if media_type == "photo":
-            # If it's an album (multiple media), we can send the rest as a media group without caption
+                media_tags.append(f'<img src="tg://photo?id={media_id}"/>')
-            if len(media_rows) > 1:
+            else:
-                media_items = []
+                media_tags.append(f'<video src="tg://video?id={media_id}"></video>')
-                for item in media_rows[1:]:
+        if not rich_media:
-                    media = self.media_value(item)
+            return None
-                    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
+        rich_text = self.rich_text_html(text)
+        if len(rich_text) > MAX_RICH_TEXT:
+            return None
+
+        media_html = media_tags[0] if len(media_tags) == 1 else f"<tg-collage>{''.join(media_tags)}</tg-collage>"
+        return {
+            "html": f"{media_html}\n{rich_text}" if rich_text else media_html,
+            "media": rich_media,
+        }
+
+    async def send_rich_message(self, rich_message: dict[str, Any]) -> list[int]:
+        data: dict[str, Any] = {
+            "chat_id": int(self.chat_id),
+            "rich_message": rich_message,
+        }
+        if self.thread_id:
+            data["message_thread_id"] = int(self.thread_id)
+
+        url = f"https://api.telegram.org/bot{self.bot_token}/sendRichMessage"
+        timeout = aiohttp.ClientTimeout(total=90)
+        async with aiohttp.ClientSession(timeout=timeout) as session:
+            async with session.post(url, json=data) as response:
+                payload = await response.json(content_type=None)
+
+        if payload.get("ok"):
+            message_id = payload.get("result", {}).get("message_id")
+            if message_id:
+                return [int(message_id)]
+            raise RichMessageUnavailable("sendRichMessage returned no message_id")
+
+        description = str(payload.get("description") or f"HTTP {response.status}")
+        if "Too Many Requests" in description and isinstance(payload.get("parameters"), dict):
+            retry_after = float(payload["parameters"].get("retry_after") or 0)
+            if retry_after > 0:
+                logger.warning("Telegram rich message flood control, sleep {}s", retry_after)
+                await asyncio.sleep(retry_after + 0.5)
+        raise RichMessageUnavailable(description)
+
+    async def send_legacy_media_post(self, text: str, media_rows: list[dict[str, Any]]) -> list[int]:
         media_items = []
         for item in media_rows:
             media = self.media_value(item)
@@ -416,11 +441,17 @@ class TelegramPoster:
                 media_items.append({"type": "photo", "media": media})
             elif media_type == "video":
                 media_items.append({"type": "video", "media": media})
-                
+
         if not media_items:
             return await self.send_text(text)
 
-        first_caption = text[: self.caption_limit]
+        if len(text) <= self.caption_limit:
+            first_caption = text
+            text_chunks: list[str] = []
+        else:
+            first_caption = self.overflow_caption[: self.caption_limit]
+            text_chunks = split_message_chunks(text, self.message_limit)
+
         first = media_items[: self.media_group_max_items]
         message_ids = []
         message_ids.extend(await self.send_media_items(first, first_caption))
@@ -434,8 +465,27 @@ class TelegramPoster:
             if self.send_delay_sec:
                 await asyncio.sleep(self.send_delay_sec)
 
+        for chunk in text_chunks:
+            message_ids.extend(await self.send_text(chunk))
+
         return message_ids
 
+    async def send_post(self, post: dict[str, Any]) -> list[int]:
+        text = self.build_text(post)
+        media_rows = await self.load_media(int(post["id"]))
+
+        if not media_rows:
+            return await self.send_text(text)
+
+        rich_message = self.build_rich_message(text, media_rows)
+        if rich_message:
+            try:
+                return await self.send_rich_message(rich_message)
+            except RichMessageUnavailable as exc:
+                logger.warning("Telegram rich message failed, fallback to legacy media post: {}", exc)
+
+        return await self.send_legacy_media_post(text, media_rows)
+
     async def mark_published(self, post_id: int, message_ids: list[int]) -> None:
         async with self.pool.acquire() as conn:
             async with conn.transaction():