81 lines
2.8 KiB
Python
81 lines
2.8 KiB
Python
"""Отправка последовательностей сообщений (Блок 4).
|
||
|
||
Последовательность читается из БД; каждое сообщение отправляется по порядку,
|
||
с картинкой из общего media-volume и своими inline-кнопками.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
from pathlib import Path
|
||
from typing import Any
|
||
|
||
from aiogram import Bot
|
||
from aiogram.types import FSInputFile, InlineKeyboardButton, InlineKeyboardMarkup
|
||
|
||
from common.config import get_settings
|
||
from common.repositories import sequences_repo
|
||
from common.texts.render import render
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
#: Лимит подписи к фото в Telegram.
|
||
CAPTION_LIMIT = 1024
|
||
|
||
|
||
def _markup(buttons: list[dict[str, Any]]) -> InlineKeyboardMarkup | None:
|
||
rows = [[InlineKeyboardButton(text=str(button["text"]), url=str(button["url"]))] for button in buttons]
|
||
return InlineKeyboardMarkup(inline_keyboard=rows) if rows else None
|
||
|
||
|
||
def _image_path(item: dict[str, Any]) -> Path | None:
|
||
name = item.get("image_name")
|
||
if not name:
|
||
return None
|
||
|
||
filename = Path(str(name)).name
|
||
candidates = [
|
||
Path(get_settings().media_dir) / filename,
|
||
Path("images") / filename,
|
||
]
|
||
for candidate in candidates:
|
||
if candidate.is_file():
|
||
return candidate
|
||
|
||
logger.warning("Картинка %s для последовательности не найдена", filename)
|
||
return None
|
||
|
||
|
||
async def send_sequence(bot: Bot, chat_id: int, key: str, context: dict[str, Any]) -> bool:
|
||
"""Отправить последовательность. False — если она не настроена/пуста."""
|
||
try:
|
||
items = await sequences_repo.get_items(key)
|
||
except Exception:
|
||
logger.warning("Не удалось прочитать последовательность %s", key, exc_info=True)
|
||
return False
|
||
|
||
enabled = [item for item in items if item.get("enabled")]
|
||
if not enabled:
|
||
return False
|
||
|
||
for item in enabled:
|
||
text = render(str(item.get("text") or ""), context) if item.get("text") else ""
|
||
path = _image_path(item)
|
||
markup = _markup(item.get("buttons") or [])
|
||
|
||
if path is not None:
|
||
if text and len(text) > CAPTION_LIMIT:
|
||
await bot.send_photo(chat_id=chat_id, photo=FSInputFile(path))
|
||
await bot.send_message(chat_id=chat_id, text=text, reply_markup=markup)
|
||
else:
|
||
await bot.send_photo(
|
||
chat_id=chat_id,
|
||
photo=FSInputFile(path),
|
||
caption=text or None,
|
||
reply_markup=markup,
|
||
)
|
||
elif text:
|
||
await bot.send_message(chat_id=chat_id, text=text, reply_markup=markup)
|
||
|
||
return True
|