405 lines
16 KiB
Python
405 lines
16 KiB
Python
from __future__ import annotations
|
|
from loguru import logger
|
|
|
|
import asyncio
|
|
import os
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any, Optional
|
|
import aiohttp
|
|
from aiogram import Bot
|
|
from aiogram.client.session.aiohttp import AiohttpSession
|
|
from aiogram.client.telegram import TelegramAPIServer
|
|
from aiogram.exceptions import TelegramRetryAfter
|
|
from aiogram.types import FSInputFile, InputMediaPhoto, InputMediaVideo
|
|
try:
|
|
from .config import settings
|
|
from .media_processor import ProcessedMedia
|
|
from .text_formatter import format_post_text, split_message_chunks
|
|
except (ImportError, ValueError):
|
|
from config import settings
|
|
from media_processor import ProcessedMedia
|
|
from text_formatter import format_post_text, split_message_chunks
|
|
|
|
MAX_MEDIA_GROUP = 10
|
|
MAX_RICH_MEDIA = 50
|
|
MAX_RICH_TEXT = 32768
|
|
|
|
|
|
def parse_topic(value: str) -> tuple[int, Optional[int]]:
|
|
raw = str(value or "").strip()
|
|
if not raw:
|
|
raise ValueError("Telegram chat ID is empty")
|
|
if ":" in raw:
|
|
chat_id, thread_id = raw.split(":", 1)
|
|
return int(chat_id), int(thread_id)
|
|
return int(raw), None
|
|
|
|
|
|
def tg_message_url(chat_id: int, message_id: int) -> str:
|
|
chat = str(abs(int(chat_id)))
|
|
if chat.startswith("100"):
|
|
chat = chat[3:]
|
|
return f"https://t.me/c/{chat}/{message_id}"
|
|
|
|
|
|
class RichMessageUnavailable(RuntimeError):
|
|
pass
|
|
|
|
|
|
class TelegramPoster:
|
|
def __init__(self) -> None:
|
|
self.bot: Optional[Bot] = None
|
|
self.chat_id: int = 0
|
|
self.thread_id: Optional[int] = None
|
|
self.storage_chat_id: Optional[int] = None
|
|
self.storage_thread_id: Optional[int] = None
|
|
self.is_local_api: bool = False
|
|
self.caption_limit: int = 1024
|
|
self.message_limit: int = 4096
|
|
|
|
async def init(self) -> None:
|
|
if not settings.tg_bot_token:
|
|
raise ValueError("TG_BOT_TOKEN is not configured")
|
|
if not settings.tg_chat_id:
|
|
raise ValueError("TG_CHAT_ID is not configured")
|
|
|
|
self.chat_id, self.thread_id = parse_topic(settings.tg_chat_id)
|
|
if settings.tg_media_channel_id:
|
|
try:
|
|
self.storage_chat_id, self.storage_thread_id = parse_topic(settings.tg_media_channel_id)
|
|
except Exception as exc:
|
|
logger.warning("Could not parse TG_MEDIA_CHANNEL_ID: {}", exc)
|
|
|
|
local_url = (settings.local_bot_api_url or "").strip().rstrip("/")
|
|
if local_url:
|
|
try:
|
|
session = AiohttpSession(
|
|
api=TelegramAPIServer.from_base(local_url, is_local=True)
|
|
)
|
|
test_bot = Bot(token=settings.tg_bot_token, session=session)
|
|
me = await test_bot.get_me()
|
|
self.bot = test_bot
|
|
self.is_local_api = True
|
|
logger.info("Connected to local Telegram Bot API at {}: @{}", local_url, me.username)
|
|
except Exception as exc:
|
|
logger.warning(
|
|
"Local Telegram Bot API unavailable at {}: {}. Falling back to standard API.",
|
|
local_url,
|
|
exc,
|
|
)
|
|
self.bot = Bot(token=settings.tg_bot_token)
|
|
self.is_local_api = False
|
|
else:
|
|
self.bot = Bot(token=settings.tg_bot_token)
|
|
self.is_local_api = False
|
|
|
|
async def close(self) -> None:
|
|
if self.bot and self.bot.session:
|
|
await self.bot.session.close()
|
|
|
|
def chat_kwargs(self, is_storage: bool = False) -> dict[str, Any]:
|
|
c_id = self.storage_chat_id if is_storage and self.storage_chat_id else self.chat_id
|
|
t_id = self.storage_thread_id if is_storage and self.storage_chat_id else self.thread_id
|
|
kwargs: dict[str, Any] = {"chat_id": int(c_id)}
|
|
if t_id:
|
|
kwargs["message_thread_id"] = int(t_id)
|
|
return kwargs
|
|
|
|
async def tg_retry(self, fn):
|
|
for attempt in range(1, 4):
|
|
try:
|
|
return await fn()
|
|
except TelegramRetryAfter as exc:
|
|
delay = float(exc.retry_after) + 0.5
|
|
logger.warning("Telegram flood control: retry after {}s", delay)
|
|
await asyncio.sleep(delay)
|
|
except Exception as exc:
|
|
if attempt >= 3:
|
|
raise
|
|
logger.warning("Telegram request error attempt {}: {}", attempt, exc)
|
|
await asyncio.sleep(2 ** attempt)
|
|
raise RuntimeError("Telegram retries exhausted")
|
|
|
|
async def upload_media_for_file_ids(
|
|
self, media_items: list[ProcessedMedia]
|
|
) -> dict[str, str]:
|
|
"""Uploads files to destination/storage to obtain tg_file_ids"""
|
|
file_ids: dict[str, str] = {}
|
|
valid_items = [m for m in media_items if m.local_path and m.local_path.exists()]
|
|
if not valid_items or not self.bot:
|
|
return file_ids
|
|
|
|
# If storage channel is configured, we send there; otherwise we send directly
|
|
use_storage = bool(self.storage_chat_id)
|
|
group = []
|
|
for item in valid_items[:MAX_MEDIA_GROUP]:
|
|
fs = FSInputFile(str(item.local_path))
|
|
if item.media_type == "photo":
|
|
group.append(InputMediaPhoto(media=fs))
|
|
else:
|
|
group.append(InputMediaVideo(media=fs))
|
|
|
|
if not group:
|
|
return file_ids
|
|
|
|
try:
|
|
if len(group) == 1:
|
|
item = valid_items[0]
|
|
fs = FSInputFile(str(item.local_path))
|
|
if item.media_type == "photo":
|
|
msg = await self.tg_retry(
|
|
lambda: self.bot.send_photo(photo=fs, **self.chat_kwargs(use_storage))
|
|
)
|
|
if msg.photo:
|
|
file_ids[item.attachment_id] = msg.photo[-1].file_id
|
|
else:
|
|
msg = await self.tg_retry(
|
|
lambda: self.bot.send_video(video=fs, **self.chat_kwargs(use_storage))
|
|
)
|
|
if msg.video:
|
|
file_ids[item.attachment_id] = msg.video.file_id
|
|
else:
|
|
msgs = await self.tg_retry(
|
|
lambda: self.bot.send_media_group(media=group, **self.chat_kwargs(use_storage))
|
|
)
|
|
for item, msg in zip(valid_items, msgs):
|
|
if item.media_type == "photo" and msg.photo:
|
|
file_ids[item.attachment_id] = msg.photo[-1].file_id
|
|
elif item.media_type == "video" and msg.video:
|
|
file_ids[item.attachment_id] = msg.video.file_id
|
|
except Exception as exc:
|
|
logger.warning("Error uploading media items for file_ids: {}", exc)
|
|
|
|
return file_ids
|
|
|
|
def build_rich_text_html(self, text: str) -> str:
|
|
paragraphs = []
|
|
for p in text.strip().split("\n\n"):
|
|
body = "<br/>".join(line for line in p.splitlines() if line.strip())
|
|
if body:
|
|
paragraphs.append(f"<p>{body}</p>")
|
|
return "\n".join(paragraphs)
|
|
|
|
def build_rich_message(
|
|
self, text: str, media_items: list[ProcessedMedia], file_ids: dict[str, str]
|
|
) -> Optional[dict[str, Any]]:
|
|
rich_media = []
|
|
media_tags = []
|
|
|
|
valid_items = [
|
|
m for m in media_items if m.attachment_id in file_ids
|
|
][:MAX_RICH_MEDIA]
|
|
|
|
if not valid_items:
|
|
return None
|
|
|
|
for idx, item in enumerate(valid_items):
|
|
f_id = file_ids[item.attachment_id]
|
|
media_id = f"m{idx}"
|
|
m_type = "photo" if item.media_type == "photo" else "video"
|
|
rich_media.append({"id": media_id, "media": {"type": m_type, "media": f_id}})
|
|
if item.media_type == "photo":
|
|
media_tags.append(f'<img src="tg://photo?id={media_id}"/>')
|
|
else:
|
|
media_tags.append(f'<video src="tg://video?id={media_id}"></video>')
|
|
|
|
rich_text = self.build_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)
|
|
|
|
base_url = (settings.local_bot_api_url or "").strip().rstrip("/") or "https://api.telegram.org"
|
|
url = f"{base_url}/bot{settings.tg_bot_token}/sendRichMessage"
|
|
|
|
timeout = aiohttp.ClientTimeout(total=90)
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
async with session.post(url, json=data) as resp:
|
|
payload = await resp.json(content_type=None)
|
|
|
|
if payload.get("ok"):
|
|
res = payload.get("result", {})
|
|
mid = res.get("message_id")
|
|
if mid:
|
|
return [int(mid)]
|
|
raise RichMessageUnavailable("sendRichMessage ok but no message_id")
|
|
|
|
desc = str(payload.get("description") or f"HTTP {resp.status}")
|
|
raise RichMessageUnavailable(desc)
|
|
|
|
async def send_legacy_media_post(
|
|
self, text: str, media_items: list[ProcessedMedia], file_ids: dict[str, str]
|
|
) -> list[int]:
|
|
if not self.bot:
|
|
raise RuntimeError("Bot not initialized")
|
|
|
|
input_media = []
|
|
valid_items = [
|
|
m for m in media_items if m.local_path or m.attachment_id in file_ids
|
|
]
|
|
|
|
for item in valid_items:
|
|
fid = file_ids.get(item.attachment_id)
|
|
val = fid if fid else (FSInputFile(str(item.local_path)) if item.local_path else None)
|
|
if not val:
|
|
continue
|
|
if item.media_type == "photo":
|
|
input_media.append({"type": "photo", "media": val})
|
|
else:
|
|
input_media.append({"type": "video", "media": val})
|
|
|
|
if not input_media:
|
|
# Text only post
|
|
return await self.send_text_post(text)
|
|
|
|
# Handle caption vs overflow
|
|
text_chunks: list[str] = []
|
|
if len(text) <= self.caption_limit:
|
|
first_caption = text
|
|
else:
|
|
first_caption = ""
|
|
text_chunks = split_message_chunks(text, self.message_limit)
|
|
|
|
first_group = input_media[:MAX_MEDIA_GROUP]
|
|
message_ids: list[int] = []
|
|
|
|
if len(first_group) == 1:
|
|
item = first_group[0]
|
|
if item["type"] == "photo":
|
|
msg = await self.tg_retry(
|
|
lambda: self.bot.send_photo(
|
|
photo=item["media"],
|
|
caption=first_caption or None,
|
|
parse_mode="HTML",
|
|
**self.chat_kwargs(),
|
|
)
|
|
)
|
|
else:
|
|
msg = await self.tg_retry(
|
|
lambda: self.bot.send_video(
|
|
video=item["media"],
|
|
caption=first_caption or None,
|
|
parse_mode="HTML",
|
|
**self.chat_kwargs(),
|
|
)
|
|
)
|
|
message_ids.append(int(msg.message_id))
|
|
else:
|
|
group = []
|
|
for idx, item in enumerate(first_group):
|
|
cap = first_caption if idx == 0 and first_caption else None
|
|
if item["type"] == "photo":
|
|
group.append(InputMediaPhoto(media=item["media"], caption=cap, parse_mode="HTML"))
|
|
else:
|
|
group.append(InputMediaVideo(media=item["media"], caption=cap, parse_mode="HTML"))
|
|
msgs = await self.tg_retry(
|
|
lambda: self.bot.send_media_group(media=group, **self.chat_kwargs())
|
|
)
|
|
message_ids.extend(int(m.message_id) for m in msgs)
|
|
|
|
# Remaining media chunks if more than 10
|
|
rest = input_media[MAX_MEDIA_GROUP:]
|
|
for start in range(0, len(rest), MAX_MEDIA_GROUP):
|
|
chunk = rest[start : start + MAX_MEDIA_GROUP]
|
|
g = [
|
|
InputMediaPhoto(media=item["media"])
|
|
if item["type"] == "photo"
|
|
else InputMediaVideo(media=item["media"])
|
|
for item in chunk
|
|
]
|
|
msgs = await self.tg_retry(
|
|
lambda: self.bot.send_media_group(media=g, **self.chat_kwargs())
|
|
)
|
|
message_ids.extend(int(m.message_id) for m in msgs)
|
|
|
|
# Overflow text chunks if caption exceeded 1024 chars
|
|
for chunk in text_chunks:
|
|
msg = await self.tg_retry(
|
|
lambda c=chunk: self.bot.send_message(
|
|
text=c,
|
|
parse_mode="HTML",
|
|
disable_web_page_preview=True,
|
|
**self.chat_kwargs(),
|
|
)
|
|
)
|
|
message_ids.append(int(msg.message_id))
|
|
|
|
return message_ids
|
|
|
|
async def send_text_post(self, text: str) -> list[int]:
|
|
if not self.bot:
|
|
raise RuntimeError("Bot not initialized")
|
|
chunks = split_message_chunks(text, self.message_limit)
|
|
mids: list[int] = []
|
|
for chunk in chunks:
|
|
msg = await self.tg_retry(
|
|
lambda c=chunk: self.bot.send_message(
|
|
text=c,
|
|
parse_mode="HTML",
|
|
disable_web_page_preview=True,
|
|
**self.chat_kwargs(),
|
|
)
|
|
)
|
|
mids.append(int(msg.message_id))
|
|
return mids
|
|
|
|
async def post_to_telegram(
|
|
self,
|
|
raw_text: str,
|
|
media_items: list[ProcessedMedia],
|
|
vk_url: Optional[str] = None,
|
|
) -> tuple[list[int], Optional[str]]:
|
|
"""
|
|
Main Telegram posting routine:
|
|
1. Formats text for HTML parse mode.
|
|
2. Uploads media (or storage channel if configured) to get file_ids.
|
|
3. Tries sendRichMessage first.
|
|
4. If unavailable, falls back to legacy media groups / single media / text.
|
|
"""
|
|
formatted_text = format_post_text(
|
|
raw_text,
|
|
parse_mode="html",
|
|
bold_first_line=settings.format_first_line_bold,
|
|
vk_url=vk_url,
|
|
)
|
|
|
|
valid_media = [m for m in media_items if not m.is_link_only and m.local_path]
|
|
|
|
# 1. Obtain file_ids if we have storage channel or if we want rich message
|
|
file_ids: dict[str, str] = {}
|
|
if valid_media and self.storage_chat_id:
|
|
file_ids = await self.upload_media_for_file_ids(valid_media)
|
|
|
|
# 2. Try Rich Message if file_ids are available
|
|
if file_ids:
|
|
rich_msg = self.build_rich_message(formatted_text, valid_media, file_ids)
|
|
if rich_msg:
|
|
try:
|
|
mids = await self.send_rich_message(rich_msg)
|
|
url = tg_message_url(self.chat_id, mids[0]) if mids else None
|
|
logger.info("Sent Telegram Rich Message: {}", mids)
|
|
return mids, url
|
|
except RichMessageUnavailable as exc:
|
|
logger.warning("Telegram sendRichMessage failed: {}. Falling back to legacy.", exc)
|
|
|
|
# 3. Fallback to legacy media group / text
|
|
mids = await self.send_legacy_media_post(formatted_text, valid_media, file_ids)
|
|
url = tg_message_url(self.chat_id, mids[0]) if mids else None
|
|
logger.info("Sent Telegram Legacy Message: {}", mids)
|
|
return mids, url
|