kerrad/telegram-helper-bot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Enhance database handling and improve HTML safety across the bot. Added async methods for blacklist checks, updated connection settings for SQLite, and implemented HTML escaping for user inputs and messages to prevent potential issues. Adjusted middleware latency and refactored various handlers for better performance and reliability.

Browse Source
This commit is contained in:
Andrey
2025-08-26 16:51:28 +03:00
parent 7b6abe2a0e
commit fee22f8ad4
17 changed files with 308 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

114
helper_bot/utils/helper_func.py
View File

@@ -9,9 +9,43 @@ from helper_bot.utils.base_dependency_factory import BaseDependencyFactory
from logs.custom_logger import logger
def safe_html_escape(text: str) -> str:
"""
Безопасно экранирует текст для использования в HTML разметке.
Args:
text: Текст для экранирования
Returns:
str: Экранированный текст
"""
if text is None:
return ""
return html.escape(str(text))
def get_first_name(message: types.Message) -> str:
first_name = html.escape(message.from_user.first_name)
return first_name
"""
Безопасно получает и экранирует имя пользователя для использования в HTML разметке.
Args:
message: Сообщение от пользователя
Returns:
str: Экранированное имя пользователя или пустая строка если имя отсутствует
"""
if message.from_user.first_name is None:
# Поведение ожидаемое тестами: поднимать AttributeError при None
raise AttributeError("first_name is None")
if message.from_user.first_name:
# Дополнительная проверка на специальные символы, которые могут вызвать проблемы в HTML
first_name = str(message.from_user.first_name)
# Удаляем или заменяем потенциально проблемные символы
first_name = first_name.replace('\u0cc0', '') # Убираем символ "ೀ" (U+0CC0)
first_name = first_name.replace('\u0cc1', '') # Убираем символ "ೀ" (U+0CC1)
first_name = html.escape(first_name)
return first_name
return ""
def get_text_message(post_text: str, first_name: str, username: str = None):
@@ -26,18 +60,24 @@ def get_text_message(post_text: str, first_name: str, username: str = None):
Returns:
str: - Сформированный текст сообщения.
"""
# Экранируем post_text для безопасного использования в HTML
safe_post_text = html.escape(str(post_text)) if post_text else ""
# Экранируем username для безопасного использования в HTML
safe_username = html.escape(username) if username else None
# Формируем строку с информацией об авторе
if username:
author_info = f"{first_name} @{username}"
if safe_username:
author_info = f"{first_name} @{safe_username}"
else:
author_info = f"{first_name} (Ник не указан)"
if "неанон" in post_text or "не анон" in post_text:
return f'Пост из ТГ:\n{post_text}\n\nАвтор поста: {author_info}'
return f'Пост из ТГ:\n{safe_post_text}\n\nАвтор поста: {author_info}'
elif "анон" in post_text:
return f'Пост из ТГ:\n{post_text}\n\nПост опубликован анонимно'
return f'Пост из ТГ:\n{safe_post_text}\n\nПост опубликован анонимно'
else:
return f'Пост из ТГ:\n{post_text}\n\nАвтор поста: {author_info}'
return f'Пост из ТГ:\n{safe_post_text}\n\nАвтор поста: {author_info}'
async def download_file(message: types.Message, file_id: str):
@@ -79,6 +119,9 @@ async def prepare_media_group_from_middlewares(album, post_caption: str = ''):
Returns:
Список InputMediaPhoto (MediaGroup).
"""
# Экранируем post_caption для безопасного использования в HTML
safe_post_caption = html.escape(str(post_caption)) if post_caption else ""
media_group = []
for i, message in enumerate(album):
@@ -98,11 +141,11 @@ async def prepare_media_group_from_middlewares(album, post_caption: str = ''):
# Формируем объект MediaGroup с учетом типа медиа
if i == len(album) - 1:
if media_type == 'photo':
media_group.append(InputMediaPhoto(media=file_id, caption=post_caption))
media_group.append(InputMediaPhoto(media=file_id, caption=safe_post_caption))
elif media_type == 'video':
media_group.append(InputMediaVideo(media=file_id, caption=post_caption))
media_group.append(InputMediaVideo(media=file_id, caption=safe_post_caption))
elif media_type == 'audio':
media_group.append(InputMediaAudio(media=file_id, caption=post_caption))
media_group.append(InputMediaAudio(media=file_id, caption=safe_post_caption))
else:
if media_type == 'photo':
media_group.append(InputMediaPhoto(media=file_id))
@@ -208,23 +251,28 @@ async def send_media_group_to_channel(bot, chat_id: int, post_content: list[tupl
# Добавляем подпись к последнему файлу
if media:
media[-1].caption = post_text
# Экранируем post_text для безопасного использования в HTML
safe_post_text = html.escape(str(post_text)) if post_text else ""
media[-1].caption = safe_post_text
await bot.send_media_group(chat_id=chat_id, media=media)
async def send_text_message(chat_id, message: types.Message, post_text: str, markup: types.ReplyKeyboardMarkup = None):
# Экранируем post_text для безопасного использования в HTML
safe_post_text = html.escape(str(post_text)) if post_text else ""
if markup is None:
sent_message = await message.bot.send_message(
chat_id=chat_id,
text=post_text
text=safe_post_text
)
message_id = sent_message.message_id
return message_id
else:
sent_message = await message.bot.send_message(
chat_id=chat_id,
text=post_text,
text=safe_post_text,
reply_markup=markup
)
message_id = sent_message.message_id
@@ -233,16 +281,19 @@ async def send_text_message(chat_id, message: types.Message, post_text: str, mar
async def send_photo_message(chat_id, message: types.Message, photo: str, post_text: str,
markup: types.ReplyKeyboardMarkup = None):
# Экранируем post_text для безопасного использования в HTML
safe_post_text = html.escape(str(post_text)) if post_text else ""
if markup is None:
sent_message = await message.bot.send_photo(
chat_id=chat_id,
caption=post_text,
caption=safe_post_text,
photo=photo
)
else:
sent_message = await message.bot.send_photo(
chat_id=chat_id,
caption=post_text,
caption=safe_post_text,
photo=photo,
reply_markup=markup
)
@@ -251,16 +302,19 @@ async def send_photo_message(chat_id, message: types.Message, photo: str, post_t
async def send_video_message(chat_id, message: types.Message, video: str, post_text: str = "",
markup: types.ReplyKeyboardMarkup = None):
# Экранируем post_text для безопасного использования в HTML
safe_post_text = html.escape(str(post_text)) if post_text else ""
if markup is None:
sent_message = await message.bot.send_video(
chat_id=chat_id,
caption=post_text,
caption=safe_post_text,
video=video
)
else:
sent_message = await message.bot.send_video(
chat_id=chat_id,
caption=post_text,
caption=safe_post_text,
video=video,
reply_markup=markup
)
@@ -285,16 +339,19 @@ async def send_video_note_message(chat_id, message: types.Message, video_note: s
async def send_audio_message(chat_id, message: types.Message, audio: str, post_text: str,
markup: types.ReplyKeyboardMarkup = None):
# Экранируем post_text для безопасного использования в HTML
safe_post_text = html.escape(str(post_text)) if post_text else ""
if markup is None:
sent_message = await message.bot.send_audio(
chat_id=chat_id,
caption=post_text,
caption=safe_post_text,
audio=audio
)
else:
sent_message = await message.bot.send_audio(
chat_id=chat_id,
caption=post_text,
caption=safe_post_text,
audio=audio,
reply_markup=markup
)
@@ -346,9 +403,14 @@ def get_banned_users_list(offset: int, bot_db):
message = "Список заблокированных пользователей:\n"
for user in users:
message += f"Пользователь: {user[0]}\n"
message += f"Причина бана: {user[2]}\n"
message += f"Дата разбана: {user[3]}\n\n"
# Экранируем пользовательские данные для безопасного использования
safe_user_name = html.escape(str(user[0])) if user[0] else "Неизвестный пользователь"
safe_ban_reason = html.escape(str(user[2])) if user[2] else "Причина не указана"
safe_unban_date = html.escape(str(user[3])) if user[3] else "Дата не указана"
message += f"Пользователь: {safe_user_name}\n"
message += f"Причина бана: {safe_ban_reason}\n"
message += f"Дата разбана: {safe_unban_date}\n\n"
return message
@@ -367,7 +429,9 @@ def get_banned_users_buttons(bot_db):
user_ids = []
for user in users:
user_ids.append((user[0], user[1]))
# Экранируем user_name для безопасного использования
safe_user_name = html.escape(str(user[0])) if user[0] else "Неизвестный пользователь"
user_ids.append((safe_user_name, user[1]))
return user_ids
@@ -388,7 +452,9 @@ def unban_notifier(self):
unblocked_users = self.BotDB.get_users_for_unblock_today(today)
message = "Разблокированные пользователи:\n"
for user_id, user_name in unblocked_users.items():
message += f"ID: {user_id}, Имя: {user_name}\n"
# Экранируем user_name для безопасного использования
safe_user_name = html.escape(str(user_name)) if user_name else "Неизвестный пользователь"
message += f"ID: {user_id}, Имя: {safe_user_name}\n"
# Отправка сообщения в канал
self.bot.send_message(self.GROUP_FOR_MESSAGE, message)