104 lines
3.0 KiB
Python
104 lines
3.0 KiB
Python
from aiogram.types import Message
|
|
|
|
from app.models.domain import TgIncomingMessage
|
|
|
|
TG_CAPTION_MAX_LEN = 1024
|
|
TG_MESSAGE_MAX_LEN = 4096
|
|
|
|
|
|
def split_text_for_media_caption(text: str) -> tuple[str | None, str | None]:
|
|
if not text:
|
|
return None, None
|
|
if len(text) <= TG_CAPTION_MAX_LEN:
|
|
return text, None
|
|
return None, text
|
|
|
|
|
|
def format_tg_author(message: Message) -> tuple[str, str | None]:
|
|
user = message.from_user
|
|
if user is None:
|
|
return "Unknown", None
|
|
parts = [user.first_name or "", user.last_name or ""]
|
|
name = " ".join(p for p in parts if p).strip() or (user.username or "User")
|
|
return name, user.username
|
|
|
|
|
|
def format_topic_pin_text(
|
|
*,
|
|
chat_name: str,
|
|
is_dm: bool,
|
|
max_chat_id: int,
|
|
participants_count: int,
|
|
max_chat_link: str | None,
|
|
) -> str:
|
|
chat_type = "PRIVATE" if is_dm else "CHAT"
|
|
lines = [
|
|
f"{chat_name} · Тип: {chat_type}",
|
|
f"id: {max_chat_id}",
|
|
f"Участников: {participants_count}",
|
|
]
|
|
if max_chat_link:
|
|
lines.extend(["", f"🔗 {max_chat_link}"])
|
|
return "\n".join(lines)
|
|
|
|
|
|
def format_tg_to_max_text(author_name: str, username: str | None, text: str) -> str:
|
|
handle = f" (@{username})" if username else ""
|
|
return f"{author_name}{handle}:\n{text}"
|
|
|
|
|
|
def extract_tg_media(message: Message) -> list[dict]:
|
|
items: list[dict] = []
|
|
if message.photo:
|
|
photo = message.photo[-1]
|
|
items.append({"kind": "photo", "file_id": photo.file_id})
|
|
elif message.video:
|
|
items.append({"kind": "video", "file_id": message.video.file_id})
|
|
elif message.document:
|
|
items.append(
|
|
{
|
|
"kind": "document",
|
|
"file_id": message.document.file_id,
|
|
"file_name": message.document.file_name,
|
|
"mime_type": message.document.mime_type,
|
|
}
|
|
)
|
|
elif message.audio:
|
|
items.append({"kind": "audio", "file_id": message.audio.file_id})
|
|
elif message.voice:
|
|
items.append({"kind": "voice", "file_id": message.voice.file_id})
|
|
elif message.sticker or message.animation or message.video_note:
|
|
items.append({"kind": "unsupported", "file_id": None})
|
|
return items
|
|
|
|
|
|
def build_tg_incoming(message: Message, forum_channel_id: int) -> TgIncomingMessage | None:
|
|
if message.chat.id != forum_channel_id:
|
|
return None
|
|
if message.is_topic_message is False and message.message_thread_id is None:
|
|
return None
|
|
|
|
author_name, username = format_tg_author(message)
|
|
text = message.text or message.caption or ""
|
|
media = extract_tg_media(message)
|
|
|
|
if not text and not media:
|
|
return None
|
|
|
|
if media and not text and any(m.get("kind") == "unsupported" for m in media):
|
|
text = "[Telegram files]"
|
|
|
|
return TgIncomingMessage(
|
|
tg_chat_id=message.chat.id,
|
|
tg_message_id=message.message_id,
|
|
tg_thread_id=message.message_thread_id,
|
|
text=text,
|
|
author_name=author_name,
|
|
author_username=username,
|
|
is_bot=bool(message.from_user and message.from_user.is_bot),
|
|
reply_to_tg_message_id=message.reply_to_message.message_id
|
|
if message.reply_to_message
|
|
else None,
|
|
media=media,
|
|
)
|