Update voice bot functionality and clean up project structure
- Added voice message handling capabilities, including saving and deleting audio messages via callback queries. - Refactored audio record management in the database to remove unnecessary fields and streamline operations. - Introduced new keyboard options for voice interactions in the bot. - Updated `.gitignore` to include voice user files for better project organization. - Removed obsolete voice bot handler files to simplify the codebase.
This commit is contained in:
3
.gitignore
vendored
3
.gitignore
vendored
@@ -89,3 +89,6 @@ env/
|
||||
ENV/
|
||||
env.bak/
|
||||
venv.bak/
|
||||
|
||||
# Other files
|
||||
voice_users/
|
||||
|
||||
@@ -725,15 +725,15 @@ class AsyncBotDB:
|
||||
await conn.close()
|
||||
|
||||
# Методы для работы с аудио
|
||||
async def add_audio_record(self, file_name: str, author_id: int, file_id: str):
|
||||
async def add_audio_record(self, file_name: str, author_id: int):
|
||||
"""Добавление аудио записи."""
|
||||
conn = None
|
||||
try:
|
||||
date_added = datetime.now().strftime("%d-%m-%Y %H:%M:%S")
|
||||
conn = await self._get_connection()
|
||||
await conn.execute(
|
||||
"INSERT INTO audio_message_reference (file_name, author_id, date_added, file_id) VALUES (?, ?, ?, ?)",
|
||||
(file_name, author_id, date_added, file_id)
|
||||
"INSERT INTO audio_message_reference (file_name, author_id, date_added) VALUES (?, ?, ?)",
|
||||
(file_name, author_id, date_added)
|
||||
)
|
||||
await conn.commit()
|
||||
except Exception as e:
|
||||
@@ -778,23 +778,7 @@ class AsyncBotDB:
|
||||
if conn:
|
||||
await conn.close()
|
||||
|
||||
async def get_audio_file_id(self, user_id: int) -> Optional[str]:
|
||||
"""Получение file_id последнего аудио пользователя."""
|
||||
conn = None
|
||||
try:
|
||||
conn = await self._get_connection()
|
||||
async with conn.execute(
|
||||
"SELECT file_id FROM audio_message_reference WHERE author_id = ? ORDER BY date_added DESC LIMIT 1",
|
||||
(user_id,)
|
||||
) as cursor:
|
||||
result = await cursor.fetchone()
|
||||
return result[0] if result else None
|
||||
except Exception as e:
|
||||
self.logger.error(f"Ошибка при получении file_id аудио: {e}")
|
||||
raise
|
||||
finally:
|
||||
if conn:
|
||||
await conn.close()
|
||||
|
||||
|
||||
async def get_audio_file_name(self, user_id: int) -> Optional[str]:
|
||||
"""Получение имени файла последнего аудио пользователя."""
|
||||
|
||||
@@ -1337,21 +1337,20 @@ class BotDB:
|
||||
self.close()
|
||||
|
||||
def get_id_for_audio_record(self, user_id):
|
||||
"""Получает ID аудио сообщения пользователя"""
|
||||
"""Получает следующий номер аудио сообщения пользователя"""
|
||||
self.logger.info(
|
||||
f"Запуск функции get_id_for_audio_record. user_id={user_id}")
|
||||
try:
|
||||
self.connect()
|
||||
r = self.cursor.execute(
|
||||
"SELECT `file_id` FROM `audio_message_reference` WHERE `author_id` = ? "
|
||||
"ORDER BY date_added DESC LIMIT 1",
|
||||
"SELECT COUNT(*) FROM `audio_message_reference` WHERE `author_id` = ?",
|
||||
(user_id,))
|
||||
result = r.fetchone()[0]
|
||||
self.logger.info(
|
||||
f"Результат функции get_id_for_audio_record: {result}")
|
||||
return result
|
||||
except sqlite3.Error as error:
|
||||
self.logger.error(f"Ошибка получения последней даты войса: {error}")
|
||||
self.logger.error(f"Ошибка получения количества аудио: {error}")
|
||||
raise
|
||||
finally:
|
||||
self.close()
|
||||
|
||||
@@ -17,8 +17,7 @@ CREATE TABLE IF NOT EXISTS audio_message_reference (
|
||||
file_name TEXT NOT NULL UNIQUE,
|
||||
author_id INTEGER NOT NULL,
|
||||
date_added DATE NOT NULL,
|
||||
listen_count INTEGER NOT NULL DEFAULT 0,
|
||||
file_id INTEGER NOT NULL
|
||||
listen_count INTEGER NOT NULL DEFAULT 0
|
||||
);
|
||||
|
||||
-- Database migrations tracking
|
||||
|
||||
@@ -1,6 +1,16 @@
|
||||
import html
|
||||
import traceback
|
||||
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
from aiogram import Router, F
|
||||
from aiogram.types import CallbackQuery
|
||||
|
||||
from helper_bot.handlers.voice.constants import CALLBACK_SAVE, CALLBACK_DELETE
|
||||
from helper_bot.handlers.voice.services import AudioFileService
|
||||
from logs.custom_logger import logger
|
||||
|
||||
from aiogram import Router
|
||||
from aiogram.fsm.context import FSMContext
|
||||
from aiogram.types import CallbackQuery
|
||||
@@ -185,3 +195,60 @@ async def change_page(
|
||||
message_id=call.message.message_id,
|
||||
reply_markup=keyboard
|
||||
)
|
||||
|
||||
|
||||
@callback_router.callback_query(F.data == CALLBACK_SAVE)
|
||||
async def save_voice_message(
|
||||
call: CallbackQuery,
|
||||
bot_db: MagicData("bot_db")
|
||||
):
|
||||
try:
|
||||
# Создаем сервис для работы с аудио файлами
|
||||
audio_service = AudioFileService(bot_db)
|
||||
|
||||
# Получаем ID пользователя из базы
|
||||
user_id = bot_db.get_user_id_by_message_id_for_voice_bot(call.message.message_id)
|
||||
|
||||
# Генерируем имя файла
|
||||
file_name = audio_service.generate_file_name(user_id)
|
||||
|
||||
# Собираем инфо о сообщении
|
||||
time_UTC = int(time.time())
|
||||
date_added = datetime.fromtimestamp(time_UTC)
|
||||
|
||||
# Сохраняем в базу данных
|
||||
audio_service.save_audio_file(file_name, user_id, date_added)
|
||||
|
||||
# Скачиваем и сохраняем файл
|
||||
await audio_service.download_and_save_audio(call.bot, call.message.message_id, file_name)
|
||||
|
||||
# Удаляем сообщение из предложки
|
||||
await call.bot.delete_message(
|
||||
chat_id=bot_db.settings['Telegram']['group_for_posts'],
|
||||
message_id=call.message.message_id
|
||||
)
|
||||
|
||||
await call.answer(text='Сохранено!', cache_time=3)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка при сохранении голосового сообщения: {e}")
|
||||
await call.answer(text='Ошибка при сохранении!', cache_time=3)
|
||||
|
||||
|
||||
@callback_router.callback_query(F.data == CALLBACK_DELETE)
|
||||
async def delete_voice_message(
|
||||
call: CallbackQuery,
|
||||
bot_db: MagicData("bot_db")
|
||||
):
|
||||
try:
|
||||
# Удаляем сообщение из предложки
|
||||
await call.bot.delete_message(
|
||||
chat_id=bot_db.settings['Telegram']['group_for_posts'],
|
||||
message_id=call.message.message_id
|
||||
)
|
||||
|
||||
await call.answer(text='Удалено!', cache_time=3)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка при удалении голосового сообщения: {e}")
|
||||
await call.answer(text='Ошибка при удалении!', cache_time=3)
|
||||
|
||||
@@ -17,7 +17,8 @@ BUTTON_TEXTS: Final[Dict[str, str]] = {
|
||||
"LEAVE_CHAT": "Выйти из чата",
|
||||
"RETURN_TO_BOT": "Вернуться в бота",
|
||||
"WANT_STICKERS": "🤪Хочу стикеры",
|
||||
"CONNECT_ADMIN": "📩Связаться с админами"
|
||||
"CONNECT_ADMIN": "📩Связаться с админами",
|
||||
"VOICE_BOT": "🎤Голосовой бот"
|
||||
}
|
||||
|
||||
# Button to command mapping for metrics
|
||||
@@ -27,7 +28,8 @@ BUTTON_COMMAND_MAPPING: Final[Dict[str, str]] = {
|
||||
"Выйти из чата": "leave_chat",
|
||||
"Вернуться в бота": "return_to_bot",
|
||||
"🤪Хочу стикеры": "want_stickers",
|
||||
"📩Связаться с админами": "connect_admin"
|
||||
"📩Связаться с админами": "connect_admin",
|
||||
"🎤Голосовой бот": "voice_bot"
|
||||
}
|
||||
|
||||
# Error messages
|
||||
|
||||
@@ -73,6 +73,7 @@ class PrivateHandlers:
|
||||
self.router.message.register(self.stickers, ChatTypeFilter(chat_type=["private"]), F.text == BUTTON_TEXTS["WANT_STICKERS"])
|
||||
self.router.message.register(self.connect_with_admin, StateFilter(FSM_STATES["START"]), ChatTypeFilter(chat_type=["private"]), F.text == BUTTON_TEXTS["CONNECT_ADMIN"])
|
||||
|
||||
|
||||
# State handlers
|
||||
self.router.message.register(self.suggest_router, StateFilter(FSM_STATES["SUGGEST"]), ChatTypeFilter(chat_type=["private"]))
|
||||
self.router.message.register(self.resend_message_in_group_for_message, StateFilter(FSM_STATES["PRE_CHAT"]), ChatTypeFilter(chat_type=["private"]))
|
||||
|
||||
3
helper_bot/handlers/voice/__init__.py
Normal file
3
helper_bot/handlers/voice/__init__.py
Normal file
@@ -0,0 +1,3 @@
|
||||
from .voice_handler import VoiceHandlers
|
||||
|
||||
__all__ = ["VoiceHandlers"]
|
||||
56
helper_bot/handlers/voice/constants.py
Normal file
56
helper_bot/handlers/voice/constants.py
Normal file
@@ -0,0 +1,56 @@
|
||||
from typing import Final, Dict
|
||||
|
||||
# Voice bot constants
|
||||
VOICE_BOT_NAME = "voice"
|
||||
|
||||
# States
|
||||
STATE_START = "START"
|
||||
STATE_STANDUP_WRITE = "STANDUP_WRITE"
|
||||
|
||||
# Commands
|
||||
CMD_START = "start"
|
||||
CMD_HELP = "help"
|
||||
CMD_RESTART = "restart"
|
||||
CMD_EMOJI = "emoji"
|
||||
CMD_REFRESH = "refresh"
|
||||
|
||||
# Command to command mapping for metrics
|
||||
COMMAND_MAPPING: Final[Dict[str, str]] = {
|
||||
"start": "voice_start",
|
||||
"help": "voice_help",
|
||||
"restart": "voice_restart",
|
||||
"emoji": "voice_emoji",
|
||||
"refresh": "voice_refresh"
|
||||
}
|
||||
|
||||
# Button texts
|
||||
BTN_SPEAK = "🎤Высказаться"
|
||||
BTN_LISTEN = "🎧Послушать"
|
||||
|
||||
# Button to command mapping for metrics
|
||||
BUTTON_COMMAND_MAPPING: Final[Dict[str, str]] = {
|
||||
"🎤Высказаться": "voice_speak",
|
||||
"🎧Послушать": "voice_listen"
|
||||
}
|
||||
|
||||
# Callback data
|
||||
CALLBACK_SAVE = "save"
|
||||
CALLBACK_DELETE = "delete"
|
||||
|
||||
# Callback to command mapping for metrics
|
||||
CALLBACK_COMMAND_MAPPING: Final[Dict[str, str]] = {
|
||||
"save": "voice_save",
|
||||
"delete": "voice_delete"
|
||||
}
|
||||
|
||||
# File paths
|
||||
VOICE_USERS_DIR = "voice_users"
|
||||
STICK_DIR = "Stick"
|
||||
STICK_PATTERN = "Hello_*"
|
||||
|
||||
# Time delays
|
||||
STICKER_DELAY = 0.3
|
||||
MESSAGE_DELAY_1 = 1.0
|
||||
MESSAGE_DELAY_2 = 1.5
|
||||
MESSAGE_DELAY_3 = 1.3
|
||||
MESSAGE_DELAY_4 = 0.8
|
||||
@@ -6,8 +6,8 @@ from typing import List, Optional, Tuple
|
||||
|
||||
from aiogram.types import FSInputFile
|
||||
|
||||
from voice_bot.handlers.exceptions import VoiceMessageError, AudioProcessingError, DatabaseError, FileOperationError
|
||||
from voice_bot.handlers.constants import (
|
||||
from helper_bot.handlers.voice.exceptions import VoiceMessageError, AudioProcessingError, DatabaseError, FileOperationError
|
||||
from helper_bot.handlers.voice.constants import (
|
||||
VOICE_USERS_DIR, STICK_DIR, STICK_PATTERN, STICKER_DELAY,
|
||||
MESSAGE_DELAY_1, MESSAGE_DELAY_2, MESSAGE_DELAY_3, MESSAGE_DELAY_4
|
||||
)
|
||||
@@ -192,7 +192,7 @@ class VoiceBotService:
|
||||
|
||||
def _get_main_keyboard(self):
|
||||
"""Получить основную клавиатуру"""
|
||||
from voice_bot.keyboards.keyboards import get_main_keyboard
|
||||
from helper_bot.keyboards.keyboards import get_main_keyboard
|
||||
return get_main_keyboard()
|
||||
|
||||
async def _send_error_to_logs(self, message: str) -> None:
|
||||
@@ -239,10 +239,10 @@ class AudioFileService:
|
||||
logger.error(f"Ошибка при генерации имени файла: {e}")
|
||||
raise FileOperationError(f"Не удалось сгенерировать имя файла: {e}")
|
||||
|
||||
def save_audio_file(self, file_name: str, user_id: int, file_id: int, date_added: datetime) -> None:
|
||||
def save_audio_file(self, file_name: str, user_id: int, date_added: datetime) -> None:
|
||||
"""Сохранить информацию об аудио файле в базу данных"""
|
||||
try:
|
||||
self.bot_db.add_audio_record(file_name, user_id, date_added, 0, file_id)
|
||||
self.bot_db.add_audio_record(file_name, user_id, date_added)
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка при сохранении аудио файла в БД: {e}")
|
||||
raise DatabaseError(f"Не удалось сохранить аудио файл в БД: {e}")
|
||||
@@ -3,7 +3,7 @@ import html
|
||||
from datetime import datetime
|
||||
from typing import Optional
|
||||
|
||||
from voice_bot.handlers.exceptions import DatabaseError
|
||||
from helper_bot.handlers.voice.exceptions import DatabaseError
|
||||
from logs.custom_logger import logger
|
||||
|
||||
|
||||
336
helper_bot/handlers/voice/voice_handler.py
Normal file
336
helper_bot/handlers/voice/voice_handler.py
Normal file
@@ -0,0 +1,336 @@
|
||||
import asyncio
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from aiogram import Router, types, F
|
||||
from aiogram.filters import Command, StateFilter, MagicData
|
||||
from aiogram.fsm.context import FSMContext
|
||||
from aiogram.types import FSInputFile
|
||||
|
||||
from helper_bot.filters.main import ChatTypeFilter
|
||||
from helper_bot.middlewares.blacklist_middleware import BlacklistMiddleware
|
||||
from helper_bot.middlewares.dependencies_middleware import DependenciesMiddleware
|
||||
|
||||
from helper_bot.utils import messages
|
||||
from helper_bot.utils.helper_func import get_first_name, update_user_info, check_user_emoji, send_voice_message
|
||||
from logs.custom_logger import logger
|
||||
from helper_bot.handlers.voice.constants import *
|
||||
from helper_bot.handlers.voice.services import VoiceBotService
|
||||
from helper_bot.handlers.voice.utils import get_last_message_text, validate_voice_message, get_user_emoji_safe
|
||||
from helper_bot.keyboards.keyboards import get_main_keyboard, get_reply_keyboard_for_voice
|
||||
from helper_bot.handlers.private.constants import BUTTON_TEXTS
|
||||
|
||||
|
||||
class VoiceHandlers:
|
||||
def __init__(self, db, settings):
|
||||
self.db = db
|
||||
self.settings = settings
|
||||
self.router = Router()
|
||||
self._setup_handlers()
|
||||
self._setup_middleware()
|
||||
|
||||
def _setup_middleware(self):
|
||||
self.router.message.middleware(DependenciesMiddleware())
|
||||
self.router.message.middleware(BlacklistMiddleware())
|
||||
|
||||
def _setup_handlers(self):
|
||||
# Обработчик кнопки "Голосовой бот"
|
||||
self.router.message.register(
|
||||
self.voice_bot_button_handler,
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
F.text == BUTTON_TEXTS["VOICE_BOT"]
|
||||
)
|
||||
|
||||
# Команды
|
||||
self.router.message.register(
|
||||
self.restart_function,
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
Command(CMD_RESTART)
|
||||
)
|
||||
|
||||
self.router.message.register(
|
||||
self.handle_emoji_message,
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
Command(CMD_EMOJI)
|
||||
)
|
||||
|
||||
self.router.message.register(
|
||||
self.help_function,
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
Command(CMD_HELP)
|
||||
)
|
||||
|
||||
self.router.message.register(
|
||||
self.start,
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
Command(CMD_START)
|
||||
)
|
||||
|
||||
# Дополнительные команды
|
||||
self.router.message.register(
|
||||
self.refresh_listen_function,
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
Command(CMD_REFRESH)
|
||||
)
|
||||
|
||||
# Обработчики состояний и кнопок
|
||||
self.router.message.register(
|
||||
self.standup_write,
|
||||
StateFilter(STATE_START),
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
F.text == BTN_SPEAK
|
||||
)
|
||||
|
||||
self.router.message.register(
|
||||
self.suggest_voice,
|
||||
StateFilter(STATE_STANDUP_WRITE),
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
)
|
||||
|
||||
self.router.message.register(
|
||||
self.standup_listen_audio,
|
||||
StateFilter(STATE_START),
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
F.text == BTN_LISTEN
|
||||
)
|
||||
|
||||
async def voice_bot_button_handler(self, message: types.Message, state: FSMContext, bot_db: MagicData("bot_db"), settings: MagicData("settings")):
|
||||
"""Обработчик кнопки 'Голосовой бот' из основной клавиатуры"""
|
||||
await self.start(message, state, bot_db, settings)
|
||||
|
||||
async def restart_function(
|
||||
self,
|
||||
message: types.Message,
|
||||
state: FSMContext,
|
||||
bot_db: MagicData("bot_db")
|
||||
):
|
||||
await message.forward(chat_id=bot_db.settings['Telegram']['group_for_logs'])
|
||||
await update_user_info(VOICE_BOT_NAME, message)
|
||||
check_user_emoji(message)
|
||||
markup = get_main_keyboard()
|
||||
await message.answer(text='Я перезапущен!', reply_markup=markup)
|
||||
await state.set_state(STATE_START)
|
||||
|
||||
async def handle_emoji_message(
|
||||
self,
|
||||
message: types.Message,
|
||||
state: FSMContext,
|
||||
bot_db: MagicData("bot_db")
|
||||
):
|
||||
await message.forward(chat_id=bot_db.settings['Telegram']['group_for_logs'])
|
||||
user_emoji = check_user_emoji(message)
|
||||
await state.set_state(STATE_START)
|
||||
if user_emoji is not None:
|
||||
await message.answer(f'Твоя эмодзя - {user_emoji}', parse_mode='HTML')
|
||||
|
||||
async def help_function(
|
||||
self,
|
||||
message: types.Message,
|
||||
state: FSMContext,
|
||||
bot_db: MagicData("bot_db"),
|
||||
settings: MagicData("settings")
|
||||
):
|
||||
await message.forward(chat_id=settings['Telegram']['group_for_logs'])
|
||||
await update_user_info(VOICE_BOT_NAME, message)
|
||||
help_message = messages.get_message(get_first_name(message), 'HELP_MESSAGE')
|
||||
await message.answer(
|
||||
text=help_message,
|
||||
disable_web_page_preview=not bot_db.settings['Telegram']['preview_link']
|
||||
)
|
||||
await state.set_state(STATE_START)
|
||||
|
||||
async def start(
|
||||
self,
|
||||
message: types.Message,
|
||||
state: FSMContext,
|
||||
bot_db: MagicData("bot_db"),
|
||||
settings: MagicData("settings")
|
||||
):
|
||||
await state.set_state(STATE_START)
|
||||
await message.forward(chat_id=settings['Telegram']['group_for_logs'])
|
||||
await update_user_info(VOICE_BOT_NAME, message)
|
||||
user_emoji = get_user_emoji_safe(bot_db, message.from_user.id)
|
||||
|
||||
# Создаем сервис и отправляем приветственные сообщения
|
||||
voice_service = VoiceBotService(bot_db, settings)
|
||||
await voice_service.send_welcome_messages(message, user_emoji)
|
||||
|
||||
|
||||
async def refresh_listen_function(
|
||||
self,
|
||||
message: types.Message,
|
||||
state: FSMContext,
|
||||
bot_db: MagicData("bot_db"),
|
||||
settings: MagicData("settings")
|
||||
):
|
||||
await message.forward(chat_id=settings['Telegram']['group_for_logs'])
|
||||
await update_user_info(VOICE_BOT_NAME, message)
|
||||
markup = get_main_keyboard()
|
||||
|
||||
# Очищаем прослушивания через сервис
|
||||
voice_service = VoiceBotService(bot_db, settings)
|
||||
voice_service.clear_user_listenings(message.from_user.id)
|
||||
|
||||
listenings_cleared_message = messages.get_message(get_first_name(message), 'LISTENINGS_CLEARED_MESSAGE')
|
||||
await message.answer(
|
||||
text=listenings_cleared_message,
|
||||
disable_web_page_preview=not settings['Telegram']['preview_link'],
|
||||
reply_markup=markup
|
||||
)
|
||||
await state.set_state(STATE_START)
|
||||
|
||||
|
||||
async def standup_write(
|
||||
self,
|
||||
message: types.Message,
|
||||
state: FSMContext,
|
||||
bot_db: MagicData("bot_db"),
|
||||
settings: MagicData("settings")
|
||||
):
|
||||
await message.forward(chat_id=settings['Telegram']['group_for_logs'])
|
||||
markup = types.ReplyKeyboardRemove()
|
||||
record_voice_message = messages.get_message(get_first_name(message), 'RECORD_VOICE_MESSAGE')
|
||||
await message.answer(text=record_voice_message, reply_markup=markup)
|
||||
|
||||
try:
|
||||
message_with_date = get_last_message_text(bot_db)
|
||||
if message_with_date:
|
||||
await message.answer(text=message_with_date, parse_mode="html")
|
||||
except Exception as e:
|
||||
logger.error(f'Не удалось получить дату последнего сообщения - {e}')
|
||||
|
||||
await state.set_state(STATE_STANDUP_WRITE)
|
||||
|
||||
|
||||
async def suggest_voice(
|
||||
self,
|
||||
message: types.Message,
|
||||
state: FSMContext,
|
||||
bot_db: MagicData("bot_db"),
|
||||
settings: MagicData("settings")
|
||||
):
|
||||
logger.info(
|
||||
f"Вызов функции suggest_voice. Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}"
|
||||
)
|
||||
await message.forward(chat_id=settings['Telegram']['group_for_logs'])
|
||||
markup = get_main_keyboard()
|
||||
|
||||
if validate_voice_message(message):
|
||||
markup_for_voice = get_reply_keyboard_for_voice()
|
||||
|
||||
# Отправляем аудио в приватный канал
|
||||
sent_message = await send_voice_message(
|
||||
settings['Telegram']['group_for_posts'],
|
||||
message,
|
||||
message.voice.file_id,
|
||||
markup_for_voice
|
||||
)
|
||||
|
||||
# Сохраняем в базу инфо о посте
|
||||
bot_db.set_user_id_and_message_id_for_voice_bot(sent_message.message_id, message.from_user.id)
|
||||
|
||||
# Отправляем юзеру ответ и возвращаем его в меню
|
||||
voice_saved_message = messages.get_message(get_first_name(message), 'VOICE_SAVED_MESSAGE')
|
||||
await message.answer(text=voice_saved_message, reply_markup=markup)
|
||||
await state.set_state(STATE_START)
|
||||
else:
|
||||
unknown_content_message = messages.get_message(get_first_name(message), 'UNKNOWN_CONTENT_MESSAGE')
|
||||
await message.forward(chat_id=settings['Telegram']['group_for_logs'])
|
||||
await message.answer(text=unknown_content_message, reply_markup=markup)
|
||||
await state.set_state(STATE_STANDUP_WRITE)
|
||||
|
||||
|
||||
async def standup_listen_audio(
|
||||
self,
|
||||
message: types.Message,
|
||||
bot_db: MagicData("bot_db"),
|
||||
settings: MagicData("settings")
|
||||
):
|
||||
markup = get_main_keyboard()
|
||||
|
||||
# Создаем сервис для работы с аудио
|
||||
voice_service = VoiceBotService(bot_db, settings)
|
||||
|
||||
try:
|
||||
# Получаем случайное аудио
|
||||
audio_data = voice_service.get_random_audio(message.from_user.id)
|
||||
|
||||
if not audio_data:
|
||||
no_audio_message = messages.get_message(get_first_name(message), 'NO_AUDIO_MESSAGE')
|
||||
await message.answer(text=no_audio_message, reply_markup=markup)
|
||||
try:
|
||||
message_with_date = get_last_message_text(bot_db)
|
||||
if message_with_date:
|
||||
await message.answer(text=message_with_date, parse_mode="html")
|
||||
except Exception as e:
|
||||
logger.error(f'Не удалось получить последнюю дату {e}')
|
||||
return
|
||||
|
||||
audio_for_user, date_added, user_emoji = audio_data
|
||||
|
||||
# Получаем путь к файлу
|
||||
path = Path(f'{VOICE_USERS_DIR}/{audio_for_user}.ogg')
|
||||
|
||||
# Проверяем существование файла
|
||||
if not path.exists():
|
||||
logger.error(f"Файл не найден: {path}")
|
||||
await message.answer(
|
||||
text="Файл аудио не найден. Обратитесь к администратору.",
|
||||
reply_markup=markup
|
||||
)
|
||||
return
|
||||
|
||||
# Проверяем размер файла
|
||||
if path.stat().st_size == 0:
|
||||
logger.error(f"Файл пустой: {path}")
|
||||
await message.answer(
|
||||
text="Файл аудио поврежден. Обратитесь к администратору.",
|
||||
reply_markup=markup
|
||||
)
|
||||
return
|
||||
|
||||
voice = FSInputFile(path)
|
||||
|
||||
# Формируем подпись
|
||||
if user_emoji:
|
||||
caption = f'{user_emoji}\nДата записи: {date_added}'
|
||||
else:
|
||||
caption = f'Дата записи: {date_added}'
|
||||
|
||||
try:
|
||||
await message.bot.send_voice(
|
||||
chat_id=message.chat.id,
|
||||
voice=voice,
|
||||
caption=caption,
|
||||
reply_markup=markup
|
||||
)
|
||||
|
||||
# Маркируем сообщение как прослушанное только после успешной отправки
|
||||
voice_service.mark_audio_as_listened(audio_for_user, message.from_user.id)
|
||||
|
||||
# Получаем количество оставшихся аудио только после успешной отправки
|
||||
remaining_count = voice_service.get_remaining_audio_count(message.from_user.id) - 1
|
||||
await message.answer(
|
||||
text=f'Осталось непрослушанных: <b>{remaining_count}</b>',
|
||||
reply_markup=markup
|
||||
)
|
||||
|
||||
except Exception as voice_error:
|
||||
if "VOICE_MESSAGES_FORBIDDEN" in str(voice_error):
|
||||
# Если голосовые сообщения запрещены, отправляем информативное сообщение
|
||||
logger.info(f"Пользователь {message.from_user.id} запретил получение голосовых сообщений")
|
||||
|
||||
privacy_message = "🔇 К сожалению, у тебя закрыт доступ к получению голосовых сообщений.\n\nДля продолжения взаимодействия с ботом необходимо дать возможность мне присылать войсы в настройках приватности Telegram.\n\n💡 Как это сделать:\n1. Открой настройки Telegram\n2. Перейди в 'Конфиденциальность и безопасность'\n3. Выбери 'Голосовые сообщения'\n4. Разреши получение от 'Всех' или добавь меня в исключения"
|
||||
|
||||
await message.answer(text=privacy_message, reply_markup=markup)
|
||||
return # Выходим без записи о прослушивании
|
||||
|
||||
else:
|
||||
raise voice_error
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка при прослушивании аудио: {e}")
|
||||
await message.answer(
|
||||
text="Произошла ошибка при получении аудио. Попробуйте позже.",
|
||||
reply_markup=markup
|
||||
)
|
||||
@@ -29,6 +29,7 @@ def get_reply_keyboard(BotDB, user_id):
|
||||
builder = ReplyKeyboardBuilder()
|
||||
builder.row(types.KeyboardButton(text="📢Предложить свой пост"))
|
||||
builder.row(types.KeyboardButton(text="📩Связаться с админами"))
|
||||
builder.row(types.KeyboardButton(text=" 🎤Голосовой бот"))
|
||||
builder.row(types.KeyboardButton(text="👋🏼Сказать пока!"))
|
||||
if not BotDB.get_info_about_stickers(user_id=user_id):
|
||||
builder.row(types.KeyboardButton(text="🤪Хочу стикеры"))
|
||||
@@ -170,3 +171,23 @@ def create_keyboard_for_approve_ban():
|
||||
builder.add(types.KeyboardButton(text="Отменить"))
|
||||
markup = builder.as_markup(resize_keyboard=True, one_time_keyboard=True)
|
||||
return markup
|
||||
|
||||
|
||||
def get_main_keyboard():
|
||||
builder = ReplyKeyboardBuilder()
|
||||
builder.add(types.KeyboardButton(text="🎤Высказаться"))
|
||||
builder.add(types.KeyboardButton(text="🎧Послушать"))
|
||||
markup = builder.as_markup(resize_keyboard=True)
|
||||
return markup
|
||||
|
||||
|
||||
def get_reply_keyboard_for_voice():
|
||||
builder = InlineKeyboardBuilder()
|
||||
builder.row(types.InlineKeyboardButton(
|
||||
text="Сохранить", callback_data="save")
|
||||
)
|
||||
builder.row(types.InlineKeyboardButton(
|
||||
text="Удалить", callback_data="delete")
|
||||
)
|
||||
markup = builder.as_markup(resize_keyboard=True, one_time_keyboard=True)
|
||||
return markup
|
||||
|
||||
@@ -2,13 +2,13 @@ from aiogram import Bot, Dispatcher
|
||||
from aiogram.client.default import DefaultBotProperties
|
||||
from aiogram.fsm.storage.memory import MemoryStorage
|
||||
from aiogram.fsm.strategy import FSMStrategy
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
from helper_bot.handlers.admin import admin_router
|
||||
from helper_bot.handlers.callback import callback_router
|
||||
from helper_bot.handlers.group import group_router
|
||||
from helper_bot.handlers.private import private_router
|
||||
from helper_bot.handlers.voice import VoiceHandlers
|
||||
from helper_bot.middlewares.dependencies_middleware import DependenciesMiddleware
|
||||
from helper_bot.middlewares.blacklist_middleware import BlacklistMiddleware
|
||||
from helper_bot.middlewares.metrics_middleware import MetricsMiddleware, ErrorMetricsMiddleware
|
||||
@@ -28,13 +28,17 @@ async def start_bot(bdf):
|
||||
dp.update.outer_middleware(MetricsMiddleware())
|
||||
dp.update.outer_middleware(BlacklistMiddleware())
|
||||
|
||||
# Создаем экземпляр VoiceHandlers
|
||||
voice_handlers = VoiceHandlers(bdf, bdf.settings)
|
||||
voice_router = voice_handlers.router
|
||||
|
||||
# Добавляем middleware напрямую к роутерам для тестирования
|
||||
admin_router.message.middleware(MetricsMiddleware())
|
||||
private_router.message.middleware(MetricsMiddleware())
|
||||
callback_router.callback_query.middleware(MetricsMiddleware())
|
||||
group_router.message.middleware(MetricsMiddleware())
|
||||
|
||||
dp.include_routers(admin_router, private_router, callback_router, group_router)
|
||||
voice_router.message.middleware(MetricsMiddleware())
|
||||
dp.include_routers(admin_router, private_router, callback_router, group_router, voice_router)
|
||||
await bot.delete_webhook(drop_pending_updates=True)
|
||||
|
||||
# Запускаем HTTP сервер для метрик параллельно с ботом
|
||||
|
||||
@@ -16,12 +16,20 @@ try:
|
||||
from ..handlers.private.constants import BUTTON_COMMAND_MAPPING
|
||||
from ..handlers.callback.constants import CALLBACK_COMMAND_MAPPING
|
||||
from ..handlers.admin.constants import ADMIN_BUTTON_COMMAND_MAPPING, ADMIN_COMMANDS
|
||||
from ..handlers.voice.constants import (
|
||||
BUTTON_COMMAND_MAPPING as VOICE_BUTTON_COMMAND_MAPPING,
|
||||
COMMAND_MAPPING as VOICE_COMMAND_MAPPING,
|
||||
CALLBACK_COMMAND_MAPPING as VOICE_CALLBACK_COMMAND_MAPPING
|
||||
)
|
||||
except ImportError:
|
||||
# Fallback if constants not available
|
||||
BUTTON_COMMAND_MAPPING = {}
|
||||
CALLBACK_COMMAND_MAPPING = {}
|
||||
ADMIN_BUTTON_COMMAND_MAPPING = {}
|
||||
ADMIN_COMMANDS = {}
|
||||
VOICE_BUTTON_COMMAND_MAPPING = {}
|
||||
VOICE_COMMAND_MAPPING = {}
|
||||
VOICE_CALLBACK_COMMAND_MAPPING = {}
|
||||
|
||||
|
||||
class MetricsMiddleware(BaseMiddleware):
|
||||
@@ -183,6 +191,13 @@ class MetricsMiddleware(BaseMiddleware):
|
||||
'user_type': "admin" if message.from_user else "unknown",
|
||||
'handler_type': "admin_handler"
|
||||
}
|
||||
# Check if it's a voice bot command
|
||||
elif command_name in VOICE_COMMAND_MAPPING:
|
||||
return {
|
||||
'command': VOICE_COMMAND_MAPPING[command_name],
|
||||
'user_type': "user" if message.from_user else "unknown",
|
||||
'handler_type': "voice_command_handler"
|
||||
}
|
||||
else:
|
||||
return {
|
||||
'command': command_name,
|
||||
@@ -206,6 +221,14 @@ class MetricsMiddleware(BaseMiddleware):
|
||||
'handler_type': "button_handler"
|
||||
}
|
||||
|
||||
# Check if it's a voice bot button click
|
||||
if message.text in VOICE_BUTTON_COMMAND_MAPPING:
|
||||
return {
|
||||
'command': VOICE_BUTTON_COMMAND_MAPPING[message.text],
|
||||
'user_type': "user" if message.from_user else "unknown",
|
||||
'handler_type': "voice_button_handler"
|
||||
}
|
||||
|
||||
return None
|
||||
|
||||
def _extract_callback_command_info(self, callback: CallbackQuery) -> Optional[Dict[str, str]]:
|
||||
@@ -222,6 +245,14 @@ class MetricsMiddleware(BaseMiddleware):
|
||||
'handler_type': "callback_handler"
|
||||
}
|
||||
|
||||
# Check if it's a voice bot callback
|
||||
if parts and parts[0] in VOICE_CALLBACK_COMMAND_MAPPING:
|
||||
return {
|
||||
'command': VOICE_CALLBACK_COMMAND_MAPPING[parts[0]],
|
||||
'user_type': "user" if callback.from_user else "unknown",
|
||||
'handler_type': "voice_callback_handler"
|
||||
}
|
||||
|
||||
return None
|
||||
|
||||
|
||||
|
||||
@@ -35,12 +35,28 @@ constants = {
|
||||
"USER_ERROR": "Увы, я не понимаю тебя😐💔 Выбери один из пунктов в нижнем меню, а затем пиши.",
|
||||
"QUESTION": "Сообщение успешно отправлено❤️ Ответим, как только сможем😉",
|
||||
"SUCCESS_SEND_MESSAGE": "Пост успешно отправлен❤️ Ожидай одобрения😊",
|
||||
# Voice handler messages
|
||||
"MESSAGE_FOR_STANDUP": "Отлично, ты вошел в режим стендапа 📣"
|
||||
"&Это свободное пространство, в котором может высказаться каждый житель нашего города, и он будет услышан🙌🏼"
|
||||
"&Для того чтобы высказаться, нажми кнопку: 'Высказаться' и запиши голосовое сообщение, оно выпадет анонимно кому-то другому🗣"
|
||||
"&Для того чтобы послушать о чем говорит наш город, нажми кнопку: 'Послушать'👂"
|
||||
"&Ты можешь анонимно пообщаться, поделиться чем-то важным, обратиться напрямую к жителям🤝 Также можешь выступить перед аудиторией (спеть песню, рассказать стихотворение, шутку)🎤"
|
||||
"&❗️Но пожалуйста не оскорбляй никого, и будь вежлив."
|
||||
"&❗️Но пожалуйста не оскорбляй никого, и будь вежлив.",
|
||||
'WELCOME_MESSAGE': "<b>Привет.</b>",
|
||||
'DESCRIPTION_MESSAGE': "<i>Здесь можно послушать голосовые сообщения от совершенно незнакомых людей из Бийска</i>",
|
||||
'ANALOGY_MESSAGE': "Это почти как написать письмо, положить его в бутылку и швырнуть в океан. Никогда не узнаешь, послушал его кто-то или нет и ответить тоже не получится..",
|
||||
'RULES_MESSAGE': "Записывать можно всё что угодно — никаких правил нет. Главное — твой голос, <i>хотя бы на 5-10 секунд</i>",
|
||||
'ANONYMITY_MESSAGE': "Здесь всё анонимно: тот, кому я отправлю твое сообщение, не узнает ни твое имя, ни твой аккаунт (так что можно не стесняться говорить то, что не стал(а) бы выкладывать в собственные соцсети)",
|
||||
'SUGGESTION_MESSAGE': "Если не знаешь, что сказать, можешь просто прочитать любое текстовое сообщение из недавно полученных или отправленных (или спеть, рассказать стихотворенье)",
|
||||
'EMOJI_INFO_MESSAGE': "Любые войсы будут помечены эмоджи. <b>Твой эмоджи - </b>{emoji}Таким эмоджи будут помечены твои сообщения для других Но другие люди не узнают кто за каким эмоджи скрывается:)",
|
||||
'HELP_INFO_MESSAGE': "Так же можешь ознакомиться с инструкцией к боту по команде /help",
|
||||
'FINAL_MESSAGE': "<b>Ну всё, достаточно инструкций. записывайся! Микрофон твой - </b> 🎤",
|
||||
'HELP_MESSAGE': "Скорее всего ответы на твои вопросы есть здесь, ознакомься: https://telegra.ph/Instrukciya-k-botu-Golosa-Bijsk-10-11-2\nЕсли это не поможет, пиши в личку: @Kerrad1",
|
||||
'VOICE_SAVED_MESSAGE': "Окей, сохранил!👌",
|
||||
'LISTENINGS_CLEARED_MESSAGE': "Прослушивания очищены. Можешь начать слушать заново🤗",
|
||||
'NO_AUDIO_MESSAGE': "Прости, ты прослушал все аудио😔. Возвращайся позже, возможно наша база пополнится",
|
||||
'UNKNOWN_CONTENT_MESSAGE': "Я тебя не понимаю🤷♀️ запиши голосовое",
|
||||
'RECORD_VOICE_MESSAGE': "Хорошо, теперь пришли мне свое голосовое сообщение"
|
||||
}
|
||||
|
||||
|
||||
|
||||
@@ -136,11 +136,7 @@ async def test_audio_operations(temp_db):
|
||||
|
||||
# Добавляем аудио запись
|
||||
with pytest.raises(sqlite3.IntegrityError):
|
||||
await temp_db.add_audio_record(file_name, user_id, file_id)
|
||||
|
||||
# # Получаем file_id
|
||||
# retrieved_file_id = await temp_db.get_audio_file_id(user_id)
|
||||
# assert retrieved_file_id == file_id
|
||||
await temp_db.add_audio_record(file_name, user_id)
|
||||
|
||||
# # Получаем имя файла
|
||||
# retrieved_file_name = await temp_db.get_audio_file_name(user_id)
|
||||
|
||||
195
tests/test_voice_bot_architecture.py
Normal file
195
tests/test_voice_bot_architecture.py
Normal file
@@ -0,0 +1,195 @@
|
||||
import pytest
|
||||
from unittest.mock import Mock, AsyncMock, patch
|
||||
from datetime import datetime
|
||||
|
||||
from helper_bot.handlers.voice.services import VoiceBotService
|
||||
from helper_bot.handlers.voice.exceptions import VoiceMessageError, AudioProcessingError
|
||||
from helper_bot.handlers.voice.utils import get_last_message_text, validate_voice_message, get_user_emoji_safe
|
||||
|
||||
|
||||
class TestVoiceBotService:
|
||||
"""Тесты для VoiceBotService"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_bot_db(self):
|
||||
"""Мок для базы данных"""
|
||||
mock_db = Mock()
|
||||
mock_db.settings = {
|
||||
'Settings': {'logs': True},
|
||||
'Telegram': {'important_logs': 'test_chat_id'}
|
||||
}
|
||||
return mock_db
|
||||
|
||||
@pytest.fixture
|
||||
def mock_settings(self):
|
||||
"""Мок для настроек"""
|
||||
return {
|
||||
'Settings': {'logs': True},
|
||||
'Telegram': {'preview_link': True}
|
||||
}
|
||||
|
||||
@pytest.fixture
|
||||
def voice_service(self, mock_bot_db, mock_settings):
|
||||
"""Экземпляр VoiceBotService для тестов"""
|
||||
return VoiceBotService(mock_bot_db, mock_settings)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_welcome_sticker_success(self, voice_service, mock_settings):
|
||||
"""Тест успешного получения стикера"""
|
||||
with patch('pathlib.Path.rglob') as mock_rglob:
|
||||
mock_rglob.return_value = ['/path/to/sticker1.tgs', '/path/to/sticker2.tgs']
|
||||
|
||||
sticker = await voice_service.get_welcome_sticker()
|
||||
|
||||
assert sticker is not None
|
||||
mock_rglob.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_welcome_sticker_no_stickers(self, voice_service, mock_settings):
|
||||
"""Тест получения стикера когда их нет"""
|
||||
with patch('pathlib.Path.rglob') as mock_rglob:
|
||||
mock_rglob.return_value = []
|
||||
|
||||
sticker = await voice_service.get_welcome_sticker()
|
||||
|
||||
assert sticker is None
|
||||
|
||||
def test_get_random_audio_success(self, voice_service, mock_bot_db):
|
||||
"""Тест успешного получения случайного аудио"""
|
||||
mock_bot_db.check_listen_audio.return_value = ['audio1', 'audio2']
|
||||
mock_bot_db.get_user_id_by_file_name.return_value = 123
|
||||
mock_bot_db.get_date_by_file_name.return_value = '2025-01-01 12:00:00'
|
||||
mock_bot_db.check_emoji_for_user.return_value = '😊'
|
||||
|
||||
result = voice_service.get_random_audio(456)
|
||||
|
||||
assert result is not None
|
||||
assert len(result) == 3
|
||||
assert result[0] in ['audio1', 'audio2']
|
||||
assert result[1] == '2025-01-01 12:00:00'
|
||||
assert result[2] == '😊'
|
||||
|
||||
def test_get_random_audio_no_audio(self, voice_service, mock_bot_db):
|
||||
"""Тест получения аудио когда их нет"""
|
||||
mock_bot_db.check_listen_audio.return_value = []
|
||||
|
||||
result = voice_service.get_random_audio(456)
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_mark_audio_as_listened_success(self, voice_service, mock_bot_db):
|
||||
"""Тест успешной пометки аудио как прослушанного"""
|
||||
voice_service.mark_audio_as_listened('test_audio', 123)
|
||||
|
||||
mock_bot_db.mark_listened_audio.assert_called_once_with('test_audio', user_id=123)
|
||||
|
||||
def test_clear_user_listenings_success(self, voice_service, mock_bot_db):
|
||||
"""Тест успешной очистки прослушиваний"""
|
||||
voice_service.clear_user_listenings(123)
|
||||
|
||||
mock_bot_db.delete_listen_count_for_user.assert_called_once_with(123)
|
||||
|
||||
|
||||
class TestVoiceHandlers:
|
||||
"""Тесты для VoiceHandlers"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_db(self):
|
||||
"""Мок для базы данных"""
|
||||
return Mock()
|
||||
|
||||
@pytest.fixture
|
||||
def mock_settings(self):
|
||||
"""Мок для настроек"""
|
||||
return {
|
||||
'Telegram': {
|
||||
'group_for_logs': 'test_logs_chat',
|
||||
'group_for_posts': 'test_posts_chat',
|
||||
'preview_link': True
|
||||
}
|
||||
}
|
||||
|
||||
@pytest.fixture
|
||||
def voice_handlers(self, mock_db, mock_settings):
|
||||
"""Экземпляр VoiceHandlers для тестов"""
|
||||
from helper_bot.handlers.voice.voice_handler import VoiceHandlers
|
||||
return VoiceHandlers(mock_db, mock_settings)
|
||||
|
||||
def test_voice_handlers_initialization(self, voice_handlers):
|
||||
"""Тест инициализации VoiceHandlers"""
|
||||
assert voice_handlers.db is not None
|
||||
assert voice_handlers.settings is not None
|
||||
assert voice_handlers.router is not None
|
||||
|
||||
def test_setup_handlers(self, voice_handlers):
|
||||
"""Тест настройки обработчиков"""
|
||||
# Проверяем, что роутер содержит обработчики
|
||||
assert len(voice_handlers.router.message.handlers) > 0
|
||||
|
||||
|
||||
class TestUtils:
|
||||
"""Тесты для утилит"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_bot_db(self):
|
||||
"""Мок для базы данных"""
|
||||
return Mock()
|
||||
|
||||
def test_get_last_message_text(self, mock_bot_db):
|
||||
"""Тест получения последнего сообщения"""
|
||||
mock_bot_db.last_date_audio.return_value = "2025-01-01 12:00:00"
|
||||
|
||||
result = get_last_message_text(mock_bot_db)
|
||||
|
||||
assert result is not None
|
||||
assert "минут" in result or "часа" in result or "дня" in result
|
||||
mock_bot_db.last_date_audio.assert_called_once()
|
||||
|
||||
def test_validate_voice_message_valid(self):
|
||||
"""Тест валидации голосового сообщения"""
|
||||
mock_message = Mock()
|
||||
mock_message.content_type = 'voice'
|
||||
|
||||
result = validate_voice_message(mock_message)
|
||||
|
||||
assert result is True
|
||||
|
||||
def test_validate_voice_message_invalid(self):
|
||||
"""Тест валидации невалидного сообщения"""
|
||||
mock_message = Mock()
|
||||
mock_message.voice = None
|
||||
|
||||
result = validate_voice_message(mock_message)
|
||||
|
||||
assert result is False
|
||||
|
||||
def test_get_user_emoji_safe(self, mock_bot_db):
|
||||
"""Тест безопасного получения эмодзи пользователя"""
|
||||
mock_bot_db.check_emoji_for_user.return_value = "😊"
|
||||
|
||||
result = get_user_emoji_safe(mock_bot_db, 123)
|
||||
|
||||
assert result == "😊"
|
||||
mock_bot_db.check_emoji_for_user.assert_called_once_with(123)
|
||||
|
||||
|
||||
class TestExceptions:
|
||||
"""Тесты для исключений"""
|
||||
|
||||
def test_voice_message_error(self):
|
||||
"""Тест VoiceMessageError"""
|
||||
try:
|
||||
raise VoiceMessageError("Тестовая ошибка")
|
||||
except VoiceMessageError as e:
|
||||
assert str(e) == "Тестовая ошибка"
|
||||
|
||||
def test_audio_processing_error(self):
|
||||
"""Тест AudioProcessingError"""
|
||||
try:
|
||||
raise AudioProcessingError("Ошибка обработки")
|
||||
except AudioProcessingError as e:
|
||||
assert str(e) == "Ошибка обработки"
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
pytest.main([__file__])
|
||||
@@ -1,30 +0,0 @@
|
||||
from .voice_handler import voice_router
|
||||
from .callback_handler import callback_router
|
||||
from .dependencies import VoiceBotMiddleware, BotDB, Settings
|
||||
from .exceptions import VoiceBotError, VoiceMessageError, AudioProcessingError, DatabaseError, FileOperationError
|
||||
from .services import VoiceBotService, AudioFileService, VoiceMessage
|
||||
from .utils import (
|
||||
format_time_ago, plural_time, get_last_message_text,
|
||||
validate_voice_message, get_user_emoji_safe
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
'voice_router',
|
||||
'callback_router',
|
||||
'VoiceBotMiddleware',
|
||||
'BotDB',
|
||||
'Settings',
|
||||
'VoiceBotError',
|
||||
'VoiceMessageError',
|
||||
'AudioProcessingError',
|
||||
'DatabaseError',
|
||||
'FileOperationError',
|
||||
'VoiceBotService',
|
||||
'AudioFileService',
|
||||
'VoiceMessage',
|
||||
'format_time_ago',
|
||||
'plural_time',
|
||||
'get_last_message_text',
|
||||
'validate_voice_message',
|
||||
'get_user_emoji_safe'
|
||||
]
|
||||
@@ -1,71 +0,0 @@
|
||||
import time
|
||||
from datetime import datetime
|
||||
|
||||
from aiogram import Router, F
|
||||
from aiogram.types import CallbackQuery
|
||||
|
||||
from voice_bot.handlers.constants import CALLBACK_SAVE, CALLBACK_DELETE, VOICE_USERS_DIR
|
||||
from voice_bot.handlers.dependencies import VoiceBotMiddleware, BotDB
|
||||
from voice_bot.handlers.services import AudioFileService
|
||||
from logs.custom_logger import logger
|
||||
|
||||
callback_router = Router()
|
||||
|
||||
# Middleware
|
||||
callback_router.callback_query.middleware(VoiceBotMiddleware())
|
||||
|
||||
|
||||
@callback_router.callback_query(F.data == CALLBACK_SAVE)
|
||||
async def save_voice_message(call: CallbackQuery, bot_db: BotDB):
|
||||
try:
|
||||
# Создаем сервис для работы с аудио файлами
|
||||
audio_service = AudioFileService(bot_db)
|
||||
|
||||
# Получаем ID пользователя из базы
|
||||
user_id = bot_db.get_user_id_by_message_id_for_voice_bot(call.message.message_id)
|
||||
|
||||
# Генерируем имя файла
|
||||
file_name = audio_service.generate_file_name(user_id)
|
||||
|
||||
# Собираем инфо о сообщении
|
||||
time_UTC = int(time.time())
|
||||
date_added = datetime.fromtimestamp(time_UTC)
|
||||
|
||||
# Определяем file_id
|
||||
file_id = 1
|
||||
if bot_db.get_last_user_audio_record(user_id=user_id):
|
||||
file_id = bot_db.get_id_for_audio_record(user_id) + 1
|
||||
|
||||
# Сохраняем в базу данных
|
||||
audio_service.save_audio_file(file_name, user_id, file_id, date_added)
|
||||
|
||||
# Скачиваем и сохраняем файл
|
||||
await audio_service.download_and_save_audio(call.bot, call.message.message_id, file_name)
|
||||
|
||||
# Удаляем сообщение из предложки
|
||||
await call.bot.delete_message(
|
||||
chat_id=bot_db.settings['Telegram']['group_for_posts'],
|
||||
message_id=call.message.message_id
|
||||
)
|
||||
|
||||
await call.answer(text='Сохранено!', cache_time=3)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка при сохранении голосового сообщения: {e}")
|
||||
await call.answer(text='Ошибка при сохранении!', cache_time=3)
|
||||
|
||||
|
||||
@callback_router.callback_query(F.data == CALLBACK_DELETE)
|
||||
async def delete_voice_message(call: CallbackQuery, bot_db: BotDB):
|
||||
try:
|
||||
# Удаляем сообщение из предложки
|
||||
await call.bot.delete_message(
|
||||
chat_id=bot_db.settings['Telegram']['group_for_posts'],
|
||||
message_id=call.message.message_id
|
||||
)
|
||||
|
||||
await call.answer(text='Удалено!', cache_time=3)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка при удалении голосового сообщения: {e}")
|
||||
await call.answer(text='Ошибка при удалении!', cache_time=3)
|
||||
@@ -1,56 +0,0 @@
|
||||
# Voice bot constants
|
||||
VOICE_BOT_NAME = "voice"
|
||||
|
||||
# States
|
||||
STATE_START = "START"
|
||||
STATE_STANDUP_WRITE = "STANDUP_WRITE"
|
||||
|
||||
# Commands
|
||||
CMD_START = "start"
|
||||
CMD_HELP = "help"
|
||||
CMD_RESTART = "restart"
|
||||
CMD_EMOJI = "emoji"
|
||||
CMD_REFRESH = "refresh"
|
||||
|
||||
# Button texts
|
||||
BTN_SPEAK = "🎤Высказаться"
|
||||
BTN_LISTEN = "🎧Послушать"
|
||||
|
||||
# Callback data
|
||||
CALLBACK_SAVE = "save"
|
||||
CALLBACK_DELETE = "delete"
|
||||
|
||||
# File paths
|
||||
VOICE_USERS_DIR = "voice_users"
|
||||
STICK_DIR = "Stick"
|
||||
STICK_PATTERN = "Hello_*"
|
||||
|
||||
# Messages
|
||||
WELCOME_MESSAGE = "<b>Привет.</b>"
|
||||
DESCRIPTION_MESSAGE = "<i>Здесь можно послушать голосовые сообщения от совершенно незнакомых людей из Бийска</i>"
|
||||
ANALOGY_MESSAGE = "Это почти как написать письмо, положить его в бутылку и швырнуть в океан. Никогда не узнаешь, послушал его кто-то или нет и ответить тоже не получится.."
|
||||
RULES_MESSAGE = "Записывать можно всё что угодно — никаких правил нет. Главное — твой голос, <i>хотя бы на 5-10 секунд</i>"
|
||||
ANONYMITY_MESSAGE = "Здесь всё анонимно: тот, кому я отправлю твое сообщение, не узнает ни твое имя, ни твой аккаунт (так что можно не стесняться говорить то, что не стал(а) бы выкладывать в собственные соцсети)"
|
||||
SUGGESTION_MESSAGE = "Если не знаешь, что сказать, можешь просто прочитать любое текстовое сообщение из недавно полученных или отправленных (или спеть, рассказать стихотворенье)"
|
||||
EMOJI_INFO_MESSAGE = "Любые войсы будут помечены эмоджи. <b>Твой эмоджи - </b>{emoji}Таким эмоджи будут помечены твои сообщения для других Но другие люди не узнают кто за каким эмоджи скрывается:)"
|
||||
HELP_INFO_MESSAGE = "Так же можешь ознакомиться с инструкцией к боту по команде /help"
|
||||
FINAL_MESSAGE = "<b>Ну всё, достаточно инструкций. записывайся! Микрофон твой - </b> 🎤"
|
||||
|
||||
# Help message
|
||||
HELP_MESSAGE = "Скорее всего ответы на твои вопросы есть здесь, ознакомься: https://telegra.ph/Instrukciya-k-botu-Golosa-Bijsk-10-11-2\nЕсли это не поможет, пиши в личку: @Kerrad1"
|
||||
|
||||
# Success messages
|
||||
VOICE_SAVED_MESSAGE = "Окей, сохранил!👌"
|
||||
LISTENINGS_CLEARED_MESSAGE = "Прослушивания очищены. Можешь начать слушать заново🤗"
|
||||
NO_AUDIO_MESSAGE = "Прости, ты прослушал все аудио😔. Возвращайся позже, возможно наша база пополнится"
|
||||
|
||||
# Error messages
|
||||
UNKNOWN_CONTENT_MESSAGE = "Я тебя не понимаю🤷♀️ запиши голосовое"
|
||||
RECORD_VOICE_MESSAGE = "Хорошо, теперь пришли мне свое голосовое сообщение"
|
||||
|
||||
# Time delays
|
||||
STICKER_DELAY = 0.3
|
||||
MESSAGE_DELAY_1 = 1.0
|
||||
MESSAGE_DELAY_2 = 1.5
|
||||
MESSAGE_DELAY_3 = 1.3
|
||||
MESSAGE_DELAY_4 = 0.8
|
||||
@@ -1,48 +0,0 @@
|
||||
from typing import Dict, Any
|
||||
try:
|
||||
from typing import Annotated
|
||||
except ImportError:
|
||||
from typing_extensions import Annotated
|
||||
from aiogram import BaseMiddleware
|
||||
from aiogram.types import TelegramObject
|
||||
|
||||
from helper_bot.utils.base_dependency_factory import get_global_instance
|
||||
from logs.custom_logger import logger
|
||||
|
||||
|
||||
class VoiceBotMiddleware(BaseMiddleware):
|
||||
"""Middleware для voice_bot с dependency injection"""
|
||||
|
||||
async def __call__(self, handler, event: TelegramObject, data: Dict[str, Any]) -> Any:
|
||||
try:
|
||||
# Вызываем хендлер с data
|
||||
return await handler(event, data)
|
||||
except TypeError as e:
|
||||
if "missing 1 required positional argument: 'data'" in str(e):
|
||||
logger.error(f"Ошибка в VoiceBotMiddleware: {e}. Хендлер не принимает параметр 'data'")
|
||||
# Пытаемся вызвать хендлер без data (для совместимости с MagicData)
|
||||
return await handler(event)
|
||||
else:
|
||||
logger.error(f"TypeError в VoiceBotMiddleware: {e}")
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error(f"Неожиданная ошибка в VoiceBotMiddleware: {e}")
|
||||
raise
|
||||
|
||||
|
||||
# Dependency providers
|
||||
def get_bot_db():
|
||||
"""Провайдер для получения экземпляра БД"""
|
||||
bdf = get_global_instance()
|
||||
return bdf.get_db()
|
||||
|
||||
|
||||
def get_settings():
|
||||
"""Провайдер для получения настроек"""
|
||||
bdf = get_global_instance()
|
||||
return bdf.settings
|
||||
|
||||
|
||||
# Type aliases for dependency injection
|
||||
BotDB = Annotated[object, get_bot_db()]
|
||||
Settings = Annotated[dict, get_settings()]
|
||||
@@ -1,215 +0,0 @@
|
||||
import asyncio
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from aiogram import Router, types, F
|
||||
from aiogram.filters import Command, StateFilter
|
||||
from aiogram.fsm.context import FSMContext
|
||||
from aiogram.types import FSInputFile
|
||||
|
||||
from helper_bot.filters.main import ChatTypeFilter
|
||||
from helper_bot.middlewares.blacklist_middleware import BlacklistMiddleware
|
||||
from helper_bot.utils.helper_func import update_user_info, check_user_emoji, send_voice_message
|
||||
from logs.custom_logger import logger
|
||||
from voice_bot.handlers.constants import *
|
||||
from voice_bot.handlers.dependencies import VoiceBotMiddleware, BotDB, Settings
|
||||
from voice_bot.handlers.services import VoiceBotService
|
||||
from voice_bot.handlers.utils import get_last_message_text, validate_voice_message, get_user_emoji_safe
|
||||
from voice_bot.keyboards.keyboards import get_main_keyboard, get_reply_keyboard_for_voice
|
||||
|
||||
voice_router = Router()
|
||||
|
||||
# Middleware
|
||||
voice_router.message.middleware(VoiceBotMiddleware())
|
||||
voice_router.message.middleware(BlacklistMiddleware())
|
||||
|
||||
|
||||
@voice_router.message(
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
Command(CMD_RESTART)
|
||||
)
|
||||
async def restart_function(message: types.Message, state: FSMContext, bot_db: BotDB):
|
||||
await message.forward(chat_id=bot_db.settings['Telegram']['group_for_logs'])
|
||||
await update_user_info(VOICE_BOT_NAME, message)
|
||||
check_user_emoji(message)
|
||||
markup = get_main_keyboard()
|
||||
await message.answer(text='Я перезапущен!', reply_markup=markup)
|
||||
await state.set_state(STATE_START)
|
||||
|
||||
|
||||
@voice_router.message(
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
Command(CMD_EMOJI)
|
||||
)
|
||||
async def handle_emoji_message(message: types.Message, state: FSMContext, bot_db: BotDB):
|
||||
await message.forward(chat_id=bot_db.settings['Telegram']['group_for_logs'])
|
||||
user_emoji = check_user_emoji(message)
|
||||
await state.set_state(STATE_START)
|
||||
if user_emoji is not None:
|
||||
await message.answer(f'Твоя эмодзя - {user_emoji}', parse_mode='HTML')
|
||||
|
||||
|
||||
@voice_router.message(
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
Command(CMD_HELP)
|
||||
)
|
||||
async def help_function(message: types.Message, state: FSMContext, bot_db: BotDB):
|
||||
await message.forward(chat_id=bot_db.settings['Telegram']['group_for_logs'])
|
||||
await update_user_info(VOICE_BOT_NAME, message)
|
||||
await message.answer(
|
||||
text=HELP_MESSAGE,
|
||||
disable_web_page_preview=not bot_db.settings['Telegram']['preview_link']
|
||||
)
|
||||
await state.set_state(STATE_START)
|
||||
|
||||
|
||||
@voice_router.message(
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
Command(CMD_START)
|
||||
)
|
||||
async def start(message: types.Message, state: FSMContext, bot_db: BotDB, settings: Settings):
|
||||
await state.set_state(STATE_START)
|
||||
await message.forward(chat_id=settings['Telegram']['group_for_logs'])
|
||||
await update_user_info(VOICE_BOT_NAME, message)
|
||||
user_emoji = get_user_emoji_safe(bot_db, message.from_user.id)
|
||||
|
||||
# Создаем сервис и отправляем приветственные сообщения
|
||||
voice_service = VoiceBotService(bot_db, settings)
|
||||
await voice_service.send_welcome_messages(message, user_emoji)
|
||||
|
||||
|
||||
@voice_router.message(
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
Command(CMD_REFRESH)
|
||||
)
|
||||
async def refresh_listen_function(message: types.Message, state: FSMContext, bot_db: BotDB):
|
||||
await message.forward(chat_id=bot_db.settings['Telegram']['group_for_logs'])
|
||||
await update_user_info(VOICE_BOT_NAME, message)
|
||||
markup = get_main_keyboard()
|
||||
|
||||
# Очищаем прослушивания через сервис
|
||||
voice_service = VoiceBotService(bot_db, bot_db.settings)
|
||||
voice_service.clear_user_listenings(message.from_user.id)
|
||||
|
||||
await message.answer(
|
||||
text=LISTENINGS_CLEARED_MESSAGE,
|
||||
disable_web_page_preview=not bot_db.settings['Telegram']['preview_link'],
|
||||
reply_markup=markup
|
||||
)
|
||||
await state.set_state(STATE_START)
|
||||
|
||||
|
||||
@voice_router.message(
|
||||
StateFilter(STATE_START),
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
F.text == BTN_SPEAK
|
||||
)
|
||||
async def standup_write(message: types.Message, state: FSMContext, bot_db: BotDB):
|
||||
await message.forward(chat_id=bot_db.settings['Telegram']['group_for_logs'])
|
||||
markup = types.ReplyKeyboardRemove()
|
||||
await message.answer(text=RECORD_VOICE_MESSAGE, reply_markup=markup)
|
||||
|
||||
try:
|
||||
message_with_date = get_last_message_text(bot_db)
|
||||
if message_with_date:
|
||||
await message.answer(text=message_with_date, parse_mode="html")
|
||||
except Exception as e:
|
||||
logger.error(f'Не удалось получить дату последнего сообщения - {e}')
|
||||
|
||||
await state.set_state(STATE_STANDUP_WRITE)
|
||||
|
||||
|
||||
@voice_router.message(
|
||||
StateFilter(STATE_STANDUP_WRITE),
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
)
|
||||
async def suggest_voice(message: types.Message, state: FSMContext, bot_db: BotDB):
|
||||
logger.info(
|
||||
f"Вызов функции suggest_voice. Пользователь: {message.from_user.id} Имя автора сообщения: {message.from_user.full_name}"
|
||||
)
|
||||
await message.forward(chat_id=bot_db.settings['Telegram']['group_for_logs'])
|
||||
markup = get_main_keyboard()
|
||||
|
||||
if validate_voice_message(message):
|
||||
markup_for_voice = get_reply_keyboard_for_voice()
|
||||
|
||||
# Отправляем аудио в приватный канал
|
||||
sent_message = await send_voice_message(
|
||||
bot_db.settings['Telegram']['group_for_posts'],
|
||||
message,
|
||||
message.voice.file_id,
|
||||
markup_for_voice
|
||||
)
|
||||
|
||||
# Сохраняем в базу инфо о посте
|
||||
bot_db.set_user_id_and_message_id_for_voice_bot(sent_message.message_id, message.from_user.id)
|
||||
|
||||
# Отправляем юзеру ответ и возвращаем его в меню
|
||||
await message.answer(text=VOICE_SAVED_MESSAGE, reply_markup=markup)
|
||||
await state.set_state(STATE_START)
|
||||
else:
|
||||
await message.forward(chat_id=bot_db.settings['Telegram']['group_for_logs'])
|
||||
await message.answer(text=UNKNOWN_CONTENT_MESSAGE, reply_markup=markup)
|
||||
await state.set_state(STATE_STANDUP_WRITE)
|
||||
|
||||
|
||||
@voice_router.message(
|
||||
StateFilter(STATE_START),
|
||||
ChatTypeFilter(chat_type=["private"]),
|
||||
F.text == BTN_LISTEN
|
||||
)
|
||||
async def standup_listen_audio(message: types.Message, bot_db: BotDB):
|
||||
markup = get_main_keyboard()
|
||||
|
||||
# Создаем сервис для работы с аудио
|
||||
voice_service = VoiceBotService(bot_db, bot_db.settings)
|
||||
|
||||
try:
|
||||
# Получаем случайное аудио
|
||||
audio_data = voice_service.get_random_audio(message.from_user.id)
|
||||
|
||||
if not audio_data:
|
||||
await message.answer(text=NO_AUDIO_MESSAGE, reply_markup=markup)
|
||||
try:
|
||||
message_with_date = get_last_message_text(bot_db)
|
||||
if message_with_date:
|
||||
await message.answer(text=message_with_date, parse_mode="html")
|
||||
except Exception as e:
|
||||
logger.error(f'Не удалось получить последнюю дату {e}')
|
||||
return
|
||||
|
||||
audio_for_user, date_added, user_emoji = audio_data
|
||||
|
||||
# Получаем путь к файлу
|
||||
path = Path(f'{VOICE_USERS_DIR}/{audio_for_user}.ogg')
|
||||
voice = FSInputFile(path)
|
||||
|
||||
# Маркируем сообщение как прослушанное
|
||||
voice_service.mark_audio_as_listened(audio_for_user, message.from_user.id)
|
||||
|
||||
# Формируем подпись
|
||||
if user_emoji:
|
||||
caption = f'{user_emoji}\nДата записи: {date_added}'
|
||||
else:
|
||||
caption = f'Дата записи: {date_added}'
|
||||
|
||||
await message.bot.send_voice(
|
||||
chat_id=message.chat.id,
|
||||
voice=voice,
|
||||
caption=caption,
|
||||
reply_markup=markup
|
||||
)
|
||||
|
||||
# Получаем количество оставшихся аудио
|
||||
remaining_count = voice_service.get_remaining_audio_count(message.from_user.id) - 1
|
||||
await message.answer(
|
||||
text=f'Осталось непрослушанных: <b>{remaining_count}</b>',
|
||||
reply_markup=markup
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(f"Ошибка при прослушивании аудио: {e}")
|
||||
await message.answer(
|
||||
text="Произошла ошибка при получении аудио. Попробуйте позже.",
|
||||
reply_markup=markup
|
||||
)
|
||||
@@ -1,22 +0,0 @@
|
||||
from aiogram import types
|
||||
from aiogram.utils.keyboard import ReplyKeyboardBuilder, InlineKeyboardBuilder
|
||||
|
||||
|
||||
def get_main_keyboard():
|
||||
builder = ReplyKeyboardBuilder()
|
||||
builder.add(types.KeyboardButton(text="🎤Высказаться"))
|
||||
builder.add(types.KeyboardButton(text="🎧Послушать"))
|
||||
markup = builder.as_markup(resize_keyboard=True)
|
||||
return markup
|
||||
|
||||
|
||||
def get_reply_keyboard_for_voice():
|
||||
builder = InlineKeyboardBuilder()
|
||||
builder.row(types.InlineKeyboardButton(
|
||||
text="Сохранить", callback_data="save")
|
||||
)
|
||||
builder.row(types.InlineKeyboardButton(
|
||||
text="Удалить", callback_data="delete")
|
||||
)
|
||||
markup = builder.as_markup(resize_keyboard=True, one_time_keyboard=True)
|
||||
return markup
|
||||
@@ -1,232 +0,0 @@
|
||||
import pytest
|
||||
from unittest.mock import Mock, AsyncMock, patch
|
||||
from datetime import datetime
|
||||
|
||||
from voice_bot.handlers.services import VoiceBotService, AudioFileService
|
||||
from voice_bot.handlers.exceptions import VoiceMessageError, AudioProcessingError
|
||||
from voice_bot.handlers.utils import format_time_ago, plural_time
|
||||
|
||||
|
||||
class TestVoiceBotService:
|
||||
"""Тесты для VoiceBotService"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_bot_db(self):
|
||||
"""Мок для базы данных"""
|
||||
mock_db = Mock()
|
||||
mock_db.settings = {
|
||||
'Settings': {'logs': True},
|
||||
'Telegram': {'important_logs': 'test_chat_id'}
|
||||
}
|
||||
return mock_db
|
||||
|
||||
@pytest.fixture
|
||||
def mock_settings(self):
|
||||
"""Мок для настроек"""
|
||||
return {
|
||||
'Settings': {'logs': True},
|
||||
'Telegram': {'preview_link': True}
|
||||
}
|
||||
|
||||
@pytest.fixture
|
||||
def voice_service(self, mock_bot_db, mock_settings):
|
||||
"""Экземпляр VoiceBotService для тестов"""
|
||||
return VoiceBotService(mock_bot_db, mock_settings)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_welcome_sticker_success(self, voice_service, mock_settings):
|
||||
"""Тест успешного получения стикера"""
|
||||
with patch('pathlib.Path.rglob') as mock_rglob:
|
||||
mock_rglob.return_value = ['/path/to/sticker1.tgs', '/path/to/sticker2.tgs']
|
||||
|
||||
sticker = await voice_service.get_welcome_sticker()
|
||||
|
||||
assert sticker is not None
|
||||
mock_rglob.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_get_welcome_sticker_no_stickers(self, voice_service, mock_settings):
|
||||
"""Тест получения стикера когда их нет"""
|
||||
with patch('pathlib.Path.rglob') as mock_rglob:
|
||||
mock_rglob.return_value = []
|
||||
|
||||
sticker = await voice_service.get_welcome_sticker()
|
||||
|
||||
assert sticker is None
|
||||
|
||||
def test_get_random_audio_success(self, voice_service, mock_bot_db):
|
||||
"""Тест успешного получения случайного аудио"""
|
||||
mock_bot_db.check_listen_audio.return_value = ['audio1', 'audio2']
|
||||
mock_bot_db.get_user_id_by_file_name.return_value = 123
|
||||
mock_bot_db.get_date_by_file_name.return_value = '2025-01-01 12:00:00'
|
||||
mock_bot_db.check_emoji_for_user.return_value = '😊'
|
||||
|
||||
result = voice_service.get_random_audio(456)
|
||||
|
||||
assert result is not None
|
||||
assert len(result) == 3
|
||||
assert result[0] in ['audio1', 'audio2']
|
||||
assert result[1] == '2025-01-01 12:00:00'
|
||||
assert result[2] == '😊'
|
||||
|
||||
def test_get_random_audio_no_audio(self, voice_service, mock_bot_db):
|
||||
"""Тест получения аудио когда их нет"""
|
||||
mock_bot_db.check_listen_audio.return_value = []
|
||||
|
||||
result = voice_service.get_random_audio(456)
|
||||
|
||||
assert result is None
|
||||
|
||||
def test_mark_audio_as_listened_success(self, voice_service, mock_bot_db):
|
||||
"""Тест успешной пометки аудио как прослушанного"""
|
||||
voice_service.mark_audio_as_listened('test_audio', 123)
|
||||
|
||||
mock_bot_db.mark_listened_audio.assert_called_once_with('test_audio', user_id=123)
|
||||
|
||||
def test_clear_user_listenings_success(self, voice_service, mock_bot_db):
|
||||
"""Тест успешной очистки прослушиваний"""
|
||||
voice_service.clear_user_listenings(123)
|
||||
|
||||
mock_bot_db.delete_listen_count_for_user.assert_called_once_with(123)
|
||||
|
||||
|
||||
class TestAudioFileService:
|
||||
"""Тесты для AudioFileService"""
|
||||
|
||||
@pytest.fixture
|
||||
def mock_bot_db(self):
|
||||
"""Мок для базы данных"""
|
||||
return Mock()
|
||||
|
||||
@pytest.fixture
|
||||
def audio_service(self, mock_bot_db):
|
||||
"""Экземпляр AudioFileService для тестов"""
|
||||
return AudioFileService(mock_bot_db)
|
||||
|
||||
def test_generate_file_name_first_audio(self, audio_service, mock_bot_db):
|
||||
"""Тест генерации имени для первого аудио пользователя"""
|
||||
mock_bot_db.get_last_user_audio_record.return_value = False
|
||||
|
||||
file_name = audio_service.generate_file_name(123)
|
||||
|
||||
assert file_name == 'message_from_123_number_1'
|
||||
|
||||
def test_generate_file_name_subsequent_audio(self, audio_service, mock_bot_db):
|
||||
"""Тест генерации имени для последующих аудио пользователя"""
|
||||
mock_bot_db.get_last_user_audio_record.return_value = True
|
||||
mock_bot_db.get_path_for_audio_record.return_value = 'existing_file'
|
||||
mock_bot_db.get_id_for_audio_record.return_value = 5
|
||||
|
||||
with patch('pathlib.Path.exists') as mock_exists:
|
||||
mock_exists.return_value = True
|
||||
|
||||
file_name = audio_service.generate_file_name(123)
|
||||
|
||||
assert file_name == 'message_from_123_number_6'
|
||||
|
||||
def test_save_audio_file_success(self, audio_service, mock_bot_db):
|
||||
"""Тест успешного сохранения аудио файла в БД"""
|
||||
file_name = 'test_file'
|
||||
user_id = 123
|
||||
file_id = 1
|
||||
date_added = datetime.now()
|
||||
|
||||
audio_service.save_audio_file(file_name, user_id, file_id, date_added)
|
||||
|
||||
mock_bot_db.add_audio_record.assert_called_once_with(
|
||||
file_name, user_id, date_added, 0, file_id
|
||||
)
|
||||
|
||||
|
||||
class TestUtils:
|
||||
"""Тесты для утилит"""
|
||||
|
||||
def test_plural_time_minutes(self):
|
||||
"""Тест множественного числа для минут"""
|
||||
assert plural_time(1, 1) == '1 минуту'
|
||||
assert plural_time(1, 2) == '2 минуты'
|
||||
assert plural_time(1, 5) == '5 минут'
|
||||
assert plural_time(1, 11) == '11 минут'
|
||||
assert plural_time(1, 21) == '21 минуту'
|
||||
|
||||
def test_plural_time_hours(self):
|
||||
"""Тест множественного числа для часов"""
|
||||
assert plural_time(2, 1) == '1 час'
|
||||
assert plural_time(2, 2) == '2 часа'
|
||||
assert plural_time(2, 5) == '5 часов'
|
||||
assert plural_time(2, 11) == '11 часов'
|
||||
assert plural_time(2, 21) == '21 час'
|
||||
|
||||
def test_plural_time_days(self):
|
||||
"""Тест множественного числа для дней"""
|
||||
assert plural_time(3, 1) == '1 день'
|
||||
assert plural_time(3, 2) == '2 дня'
|
||||
assert plural_time(3, 5) == '5 дней'
|
||||
assert plural_time(3, 11) == '11 дней'
|
||||
assert plural_time(3, 21) == '21 день'
|
||||
|
||||
def test_format_time_ago_minutes(self):
|
||||
"""Тест форматирования времени для минут"""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
# Создаем время 30 минут назад
|
||||
past_time = datetime.now() - timedelta(minutes=30)
|
||||
time_str = past_time.strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
result = format_time_ago(time_str)
|
||||
|
||||
assert '30 минут' in result
|
||||
assert 'назад' in result
|
||||
|
||||
def test_format_time_ago_hours(self):
|
||||
"""Тест форматирования времени для часов"""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
# Создаем время 2 часа назад
|
||||
past_time = datetime.now() - timedelta(hours=2)
|
||||
time_str = past_time.strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
result = format_time_ago(time_str)
|
||||
|
||||
assert '2 часа' in result
|
||||
assert 'назад' in result
|
||||
|
||||
def test_format_time_ago_days(self):
|
||||
"""Тест форматирования времени для дней"""
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
# Создаем время 3 дня назад
|
||||
past_time = datetime.now() - timedelta(days=3)
|
||||
time_str = past_time.strftime("%Y-%m-%d %H:%M:%S")
|
||||
|
||||
result = format_time_ago(time_str)
|
||||
|
||||
assert '3 дня' in result
|
||||
assert 'назад' in result
|
||||
|
||||
|
||||
class TestExceptions:
|
||||
"""Тесты для исключений"""
|
||||
|
||||
def test_voice_bot_error_inheritance(self):
|
||||
"""Тест наследования исключений"""
|
||||
assert issubclass(VoiceMessageError, VoiceBotError)
|
||||
assert issubclass(AudioProcessingError, VoiceBotError)
|
||||
assert issubclass(DatabaseError, VoiceBotError)
|
||||
assert issubclass(FileOperationError, VoiceBotError)
|
||||
|
||||
def test_exception_messages(self):
|
||||
"""Тест сообщений исключений"""
|
||||
try:
|
||||
raise VoiceMessageError("Тестовая ошибка")
|
||||
except VoiceMessageError as e:
|
||||
assert str(e) == "Тестовая ошибка"
|
||||
|
||||
try:
|
||||
raise AudioProcessingError("Ошибка обработки")
|
||||
except AudioProcessingError as e:
|
||||
assert str(e) == "Ошибка обработки"
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
pytest.main([__file__])
|
||||
@@ -1,6 +0,0 @@
|
||||
def get_message(username: str, type_message: str):
|
||||
constants = {
|
||||
|
||||
}
|
||||
message = constants[type_message]
|
||||
return message.replace('username', username).replace('&', '\n')
|
||||
Reference in New Issue
Block a user