121 lines
4.2 KiB
Python
121 lines
4.2 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
|
||
from common.texts.store import get_text
|
||
|
||
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],
|
||
*,
|
||
reply_markup: Any = None,
|
||
) -> 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 index, item in enumerate(enabled):
|
||
text = render(str(item.get("text") or ""), context) if item.get("text") else ""
|
||
path = _image_path(item)
|
||
is_last = index == len(enabled) - 1
|
||
markup = reply_markup if is_last and reply_markup is not None else _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
|
||
|
||
|
||
async def template_text(key: str, context: dict[str, Any] | None = None) -> str:
|
||
"""Текст первого включённого сообщения последовательности (для edit_message_text)."""
|
||
try:
|
||
items = await sequences_repo.get_items(key)
|
||
except Exception:
|
||
logger.warning("Не удалось прочитать последовательность %s", key, exc_info=True)
|
||
items = []
|
||
|
||
for item in items:
|
||
if item.get("enabled") and item.get("text"):
|
||
return render(str(item["text"]), context or {})
|
||
return render(await get_text(key), context or {})
|
||
|
||
|
||
async def send_template(
|
||
bot: Bot,
|
||
chat_id: int,
|
||
key: str,
|
||
context: dict[str, Any] | None = None,
|
||
*,
|
||
reply_markup: Any = None,
|
||
) -> bool:
|
||
"""Отправить последовательность, а при её отсутствии — старый шаблон из кода."""
|
||
if await send_sequence(bot, chat_id, key, context or {}, reply_markup=reply_markup):
|
||
return True
|
||
|
||
text = render(await get_text(key), context or {})
|
||
await bot.send_message(chat_id=chat_id, text=text, reply_markup=reply_markup)
|
||
return True
|