From ab07cf5e1a867431edbeeeb975cd526538e992cf Mon Sep 17 00:00:00 2001 From: Your Name Date: Wed, 22 Jul 2026 17:13:36 +0500 Subject: [PATCH] Use Telegram rich messages for media posts --- src/vk_parser_app/workers/tg_poster.py | 156 ++++++++++++++++--------- 1 file changed, 103 insertions(+), 53 deletions(-) diff --git a/src/vk_parser_app/workers/tg_poster.py b/src/vk_parser_app/workers/tg_poster.py index aa7690b..82bc008 100644 --- a/src/vk_parser_app/workers/tg_poster.py +++ b/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]: - text = self.build_text(post) - 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) + def rich_media_value(self, item: dict[str, Any]) -> str | None: + stored = str(item.get("tg_file_id") or item.get("storage_attachment_id") or "").strip() + return stored or None - # 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 + def rich_text_html(self, text: str) -> str: + paragraphs = [] + for paragraph in str(text or "").strip().split("\n\n"): + body = "
".join(line for line in paragraph.splitlines() if line.strip()) + if body: + paragraphs.append(f"

{body}

") + return "\n".join(paragraphs) - 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 + def build_rich_message(self, text: str, media_rows: list[dict[str, Any]]) -> dict[str, Any] | None: + rich_media = [] + media_tags = [] + for idx, item in enumerate(media_rows[:MAX_RICH_MEDIA]): + media_type = str(item.get("media_type") or "") + if media_type not in ("photo", "video"): + continue + media = self.rich_media_value(item) + if not media: + return None + media_id = f"m{idx}" + input_type = "photo" if media_type == "photo" else "video" + rich_media.append({"id": media_id, "media": {"type": input_type, "media": media}}) + if media_type == "photo": + media_tags.append(f'') + else: + media_tags.append(f'') + if not rich_media: + return None - # 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"{''.join(media_tags)}" + 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():