bot rafactor and bugfix
This commit is contained in:
@@ -1,19 +1,214 @@
|
||||
from typing import Iterable
|
||||
from __future__ import annotations
|
||||
import logging
|
||||
from typing import Dict, Any, Optional, Iterable, Tuple
|
||||
|
||||
def make_keyboard_payload(buttons: Iterable[tuple[str, str]] | None):
|
||||
from telegram import Bot, Message, InlineKeyboardMarkup
|
||||
from telegram.error import InvalidToken, TelegramError
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
def make_keyboard_payload(buttons: Optional[Iterable[Tuple[str, str]]]) -> Optional[Dict]:
|
||||
"""
|
||||
Создает структуру inline-клавиатуры для API Telegram.
|
||||
|
||||
Args:
|
||||
buttons: Список кнопок в формате [(text, url), ...]
|
||||
|
||||
Returns:
|
||||
Dict в формате {"rows": [[{"text": text, "url": url}], ...]}
|
||||
"""
|
||||
if not buttons:
|
||||
return None
|
||||
rows = [[{"text": t, "url": u}] for t, u in buttons]
|
||||
return {"rows": rows}
|
||||
|
||||
def build_payload(
|
||||
ptype: str,
|
||||
text: Optional[str] = None,
|
||||
media_file_id: Optional[str] = None,
|
||||
parse_mode: Optional[str] = None,
|
||||
keyboard: Optional[Dict[str, Any]] = None
|
||||
) -> Dict[str, Any]:
|
||||
"""
|
||||
Строит payload для отправки поста.
|
||||
|
||||
Args:
|
||||
ptype: Тип поста (text/photo/video/animation)
|
||||
text: Текст сообщения
|
||||
media_file_id: ID медиафайла в Telegram
|
||||
parse_mode: Формат разметки (HTML/MarkdownV2)
|
||||
keyboard: Inline клавиатура
|
||||
|
||||
Returns:
|
||||
Dict содержащий все необходимые поля для отправки
|
||||
"""
|
||||
payload: Dict[str, Any] = {
|
||||
"type": str(ptype),
|
||||
"text": text if text is not None else "",
|
||||
"parse_mode": str(parse_mode) if parse_mode is not None else "html",
|
||||
"keyboard": keyboard if keyboard is not None else {},
|
||||
}
|
||||
if media_file_id:
|
||||
payload["media_file_id"] = media_file_id
|
||||
return payload
|
||||
|
||||
def build_payload(ptype: str, text: str | None, media_file_id: str | None,
|
||||
parse_mode: str | None, keyboard: dict | None) -> dict:
|
||||
# ptype: "text" | "photo" | "video" | "animation"
|
||||
return {
|
||||
"type": ptype,
|
||||
"text": text,
|
||||
"media_file_id": media_file_id,
|
||||
"parse_mode": parse_mode,
|
||||
"keyboard": keyboard,
|
||||
}
|
||||
async def validate_bot_token(token: str) -> Tuple[bool, Optional[str], Optional[int]]:
|
||||
"""
|
||||
Проверяет валидность токена бота и возвращает его username и ID.
|
||||
|
||||
Args:
|
||||
token: Токен бота для проверки
|
||||
|
||||
Returns:
|
||||
tuple[bool, Optional[str], Optional[int]]: (is_valid, username, bot_id)
|
||||
"""
|
||||
try:
|
||||
bot = Bot(token)
|
||||
me = await bot.get_me()
|
||||
return True, me.username, me.id
|
||||
except InvalidToken:
|
||||
logger.warning(f"Invalid bot token provided: {token[:10]}...")
|
||||
return False, None, None
|
||||
except TelegramError as e:
|
||||
logger.error(f"Telegram error while validating bot token: {e}")
|
||||
return False, None, None
|
||||
except Exception as e:
|
||||
logger.exception(f"Unexpected error while validating bot token: {e}")
|
||||
return False, None, None
|
||||
finally:
|
||||
if 'bot' in locals():
|
||||
await bot.close()
|
||||
|
||||
def validate_message_length(text: str) -> bool:
|
||||
"""
|
||||
Проверяет длину сообщения на соответствие лимитам Telegram.
|
||||
|
||||
Args:
|
||||
text: Текст для проверки
|
||||
|
||||
Returns:
|
||||
bool: True если длина в пределах лимита
|
||||
"""
|
||||
return len(text) <= 4096 # Максимальная длина текста в Telegram
|
||||
|
||||
def is_valid_webhook_url(url: str) -> bool:
|
||||
"""Проверяет соответствие URL требованиям Telegram для вебхуков.
|
||||
|
||||
Args:
|
||||
url: URL для проверки
|
||||
|
||||
Returns:
|
||||
bool: True если URL валидный, иначе False
|
||||
"""
|
||||
if not url:
|
||||
return False
|
||||
|
||||
return True # TODO: implement proper validation
|
||||
|
||||
|
||||
class PostService:
|
||||
"""Сервис для работы с постами."""
|
||||
|
||||
@staticmethod
|
||||
async def preview_post(message: Message, post_data: Dict[str, Any]) -> None:
|
||||
"""Показывает предпросмотр поста.
|
||||
|
||||
Args:
|
||||
message (Message): Telegram сообщение
|
||||
post_data (Dict[str, Any]): Данные поста из сессии
|
||||
"""
|
||||
text = post_data.get('text', '')
|
||||
parse_mode = post_data.get('parse_mode', 'HTML')
|
||||
keyboard = post_data.get('keyboard')
|
||||
|
||||
if keyboard:
|
||||
# Создаем разметку клавиатуры
|
||||
rows = keyboard.get('rows', [])
|
||||
markup = InlineKeyboardMarkup(rows) if rows else None
|
||||
else:
|
||||
markup = None
|
||||
|
||||
media_file_id = post_data.get('media_file_id')
|
||||
if media_file_id:
|
||||
# Отправляем медиафайл с подписью
|
||||
try:
|
||||
await message.reply_photo(
|
||||
photo=media_file_id,
|
||||
caption=text,
|
||||
parse_mode=parse_mode,
|
||||
reply_markup=markup
|
||||
)
|
||||
except TelegramError as e:
|
||||
# В случае ошибки отправляем только текст
|
||||
logger.error(f"Error sending photo preview: {e}")
|
||||
await message.reply_text(
|
||||
text=text,
|
||||
parse_mode=parse_mode,
|
||||
reply_markup=markup
|
||||
)
|
||||
else:
|
||||
# Отправляем только текст
|
||||
await message.reply_text(
|
||||
text=text,
|
||||
parse_mode=parse_mode,
|
||||
reply_markup=markup
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
async def create_post(bot: Bot, chat_id: int, post_data: Dict[str, Any]) -> bool:
|
||||
"""Создает новый пост в канале.
|
||||
|
||||
Args:
|
||||
bot (Bot): Экземпляр бота
|
||||
chat_id (int): ID канала
|
||||
post_data (Dict[str, Any]): Данные поста
|
||||
|
||||
Returns:
|
||||
bool: Успешность создания
|
||||
"""
|
||||
try:
|
||||
text = post_data.get('text', '')
|
||||
parse_mode = post_data.get('parse_mode', 'HTML')
|
||||
keyboard = post_data.get('keyboard')
|
||||
|
||||
if keyboard:
|
||||
rows = keyboard.get('rows', [])
|
||||
markup = InlineKeyboardMarkup(rows) if rows else None
|
||||
else:
|
||||
markup = None
|
||||
|
||||
media_file_id = post_data.get('media_file_id')
|
||||
if media_file_id:
|
||||
await bot.send_photo(
|
||||
chat_id=chat_id,
|
||||
photo=media_file_id,
|
||||
caption=text,
|
||||
parse_mode=parse_mode,
|
||||
reply_markup=markup
|
||||
)
|
||||
else:
|
||||
await bot.send_message(
|
||||
chat_id=chat_id,
|
||||
text=text,
|
||||
parse_mode=parse_mode,
|
||||
reply_markup=markup
|
||||
)
|
||||
return True
|
||||
|
||||
except TelegramError as e:
|
||||
logger.error(f"Error creating post: {e}")
|
||||
return False
|
||||
def validate_url(url: str) -> bool:
|
||||
"""Проверяет соответствие URL требованиям.
|
||||
|
||||
Args:
|
||||
url (str): URL для проверки
|
||||
|
||||
Returns:
|
||||
bool: True если URL соответствует требованиям
|
||||
"""
|
||||
return (
|
||||
url.startswith("https://") and
|
||||
not url.startswith("https://telegram.org") and
|
||||
len(url) <= 512
|
||||
)
|
||||
Reference in New Issue
Block a user