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():