app/bot/bot_handler.py - master - dayekb/vibecoding_tgbot

app/bot/bot_handler.py
443 строки20 KB
Your Name
Initial commit: TG Daily Bot - Agile Team Standup Bot
19 авг 2025, 23:23
19 авг 2025, 23:231bc4aac
from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup
from telegram.ext import (
    Application, CommandHandler, MessageHandler, CallbackQueryHandler,
    ContextTypes, filters
)
from sqlalchemy.orm import Session
from app.services.user_service import UserService
from app.services.team_service import TeamService
from app.services.daily_service import DailyService
from app.services.telegram_service import TelegramService
from app.models.daily_question import DailyQuestion
from app.schemas.user import UserCreate
from app.schemas.team import TeamCreate
from app.schemas.team_member import TeamMemberCreate
from app.schemas.daily_response import DailyResponseCreate
from config.database import get_db
from config.settings import settings
from typing import Dict, Any
from datetime import datetime, date
import secrets
import asyncio


class TelegramBotHandler:
    def __init__(self):
        self.application = Application.builder().token(settings.telegram_bot_token).build()
        self.user_states: Dict[int, Dict[str, Any]] = {}  # Состояния пользователей
        self.invitation_links: Dict[str, Dict[str, Any]] = {}  # Ссылки-приглашения
        
        self.setup_handlers()
    
    def setup_handlers(self):
        """Настройка обработчиков команд и сообщений"""
        # Команды
        self.application.add_handler(CommandHandler("start", self.start_command))
        self.application.add_handler(CommandHandler("help", self.help_command))
        self.application.add_handler(CommandHandler("menu", self.menu_command))
        
        # Обработка callback'ов от inline кнопок
        self.application.add_handler(CallbackQueryHandler(self.handle_callback))
        
        # Обработка текстовых сообщений
        self.application.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, self.handle_message))
    
    async def start_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Обработка команды /start"""
        user = update.effective_user
        chat_id = update.effective_chat.id
        
        # Проверяем, зарегистрирован ли пользователь
        db = next(get_db())
        user_service = UserService(db)
        existing_user = user_service.get_user_by_telegram_id(user.id)
        
        if existing_user:
            # Пользователь уже зарегистрирован - проверяем его роли в командах
            team_service = TeamService(db)
            user_teams = team_service.get_user_teams(user.id)
            
            # Проверяем, есть ли команды где пользователь менеджер
            manager_teams = [tm for tm in user_teams if tm.is_manager]
            
            if manager_teams:
                # Пользователь менеджер в хотя бы одной команде
                await self.send_manager_menu(chat_id)
            else:
                # Пользователь только участник команд
                await self.send_team_member_menu(chat_id)
        else:
            # Новый пользователь - предлагаем выбрать роль
            await self.send_welcome_message(chat_id, user.first_name)
            await self.send_role_selection(chat_id)
            
            # Сохраняем состояние пользователя
            self.user_states[user.id] = {
                "state": "selecting_role",
                "telegram_id": user.id,
                "first_name": user.first_name,
                "last_name": user.last_name,
                "username": user.username
            }
    
    async def help_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Обработка команды /help"""
        help_text = """
❓ Справка по TG Daily Bot

🔧 Для менеджеров:
• /start - Начать работу с ботом
• /menu - Главное меню
• Создание команд и управление участниками
• Настройка времени опросов и отчетов

👤 Для участников команды:
• /start - Начать работу с ботом
• /menu - Главное меню
• Ежедневные опросы
• Просмотр своих ответов

📝 Основные функции:
• Автоматические ежедневные опросы
• Структурированные отчеты
• Кастомизируемые вопросы
• Интеграция с Telegram группами

По всем вопросам обращайтесь к менеджеру команды.
        """
        
        await update.message.reply_text(help_text.strip())
    
    async def menu_command(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Обработка команды /menu"""
        user = update.effective_user
        chat_id = update.effective_chat.id
        
        db = next(get_db())
        user_service = UserService(db)
        existing_user = user_service.get_user_by_telegram_id(user.id)
        
        if existing_user:
            # Проверяем роли пользователя в командах
            team_service = TeamService(db)
            user_teams = team_service.get_user_teams(user.id)
            
            # Проверяем, есть ли команды где пользователь менеджер
            manager_teams = [tm for tm in user_teams if tm.is_manager]
            
            if manager_teams:
                # Пользователь менеджер в хотя бы одной команде
                await self.send_manager_menu(chat_id)
            else:
                # Пользователь только участник команд
                await self.send_team_member_menu(chat_id)
        else:
            await update.message.reply_text("Сначала зарегистрируйтесь с помощью /start")
    
    async def send_manager_menu(self, chat_id: int):
        """Отправка меню менеджера"""
        telegram_service = TelegramService(settings.telegram_bot_token)
        await telegram_service.send_manager_menu(chat_id)
    
    async def send_team_member_menu(self, chat_id: int):
        """Отправка меню участника команды"""
        telegram_service = TelegramService(settings.telegram_bot_token)
        await telegram_service.send_team_member_menu(chat_id)
    
    async def send_welcome_message(self, chat_id: int, user_name: str):
        """Отправка приветственного сообщения"""
        telegram_service = TelegramService(settings.telegram_bot_token)
        await telegram_service.send_welcome_message(chat_id, user_name)
    
    async def send_role_selection(self, chat_id: int):
        """Отправка меню выбора роли"""
        telegram_service = TelegramService(settings.telegram_bot_token)
        await telegram_service.send_role_selection(chat_id)
    
    async def handle_callback(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Обработка callback'ов от inline кнопок"""
        query = update.callback_query
        await query.answer()
        
        user_id = query.from_user.id
        chat_id = query.message.chat_id
        data = query.data
        
        if data.startswith("role_"):
            await self.handle_role_selection(query, data)
        elif data.startswith("time_"):
            await self.handle_time_selection(query, data)
        elif data == "create_team":
            await self.handle_create_team(query)
        elif data == "manage_teams":
            await self.handle_manage_teams(query)
        elif data == "take_survey":
            await self.handle_take_survey(query)
        elif data.startswith("answer_"):
            await self.handle_answer_question(query, data)
        else:
            await query.message.reply_text("Неизвестная команда")
    
    async def handle_role_selection(self, query, data: str):
        """Обработка выбора роли"""
        user_id = query.from_user.id
        role_mapping = {
            "role_ds": "data_scientist",
            "role_frontend": "frontend",
            "role_backend": "backend",
            "role_fullstack": "fullstack",
            "role_designer": "designer",
            "role_custom": "custom"
        }
        
        if data in role_mapping:
            # Обновляем состояние пользователя
            if user_id in self.user_states:
                self.user_states[user_id]["team_role"] = role_mapping[data]
                self.user_states[user_id]["state"] = "selecting_time"
                
                # Отправляем меню выбора времени
                telegram_service = TelegramService(settings.telegram_bot_token)
                await telegram_service.send_time_selection(query.message.chat_id)
            else:
                await query.message.reply_text("Ошибка: пользователь не найден. Попробуйте /start")
    
    async def handle_time_selection(self, query, data: str):
        """Обработка выбора времени ответов"""
        user_id = query.from_user.id
        
        if user_id in self.user_states:
            time_mapping = {
                "time_evening": "evening",
                "time_morning": "morning"
            }
            
            if data in time_mapping:
                self.user_states[user_id]["response_time"] = time_mapping[data]
                
                # Регистрируем пользователя
                await self.register_user(user_id, query.message.chat_id)
            else:
                await query.message.reply_text("Ошибка: неверное время")
        else:
            await query.message.reply_text("Ошибка: пользователь не найден. Попробуйте /start")
    
    async def register_user(self, user_id: int, chat_id: int):
        """Регистрация нового пользователя"""
        if user_id not in self.user_states:
            await self.send_error_message(chat_id, "Ошибка регистрации")
            return
        
        user_data = self.user_states[user_id]
        
        try:
            db = next(get_db())
            user_service = UserService(db)
            
            # Создаем пользователя (без роли, так как теперь роли на уровне команды)
            user_create = UserCreate(
                telegram_id=user_data["telegram_id"],
                username=user_data.get("username"),
                first_name=user_data["first_name"],
                last_name=user_data.get("last_name")
            )
            
            user = user_service.create_user(user_create)
            
            # Очищаем состояние
            del self.user_states[user_id]
            
            # Отправляем меню участника команды (по умолчанию)
            await self.send_team_member_menu(chat_id)
                
            await self.send_success_message(chat_id, "Регистрация успешно завершена!")
            
        except Exception as e:
            await self.send_error_message(chat_id, f"Ошибка регистрации: {str(e)}")
    
    async def handle_create_team(self, query):
        """Обработка создания команды"""
        user_id = query.from_user.id
        chat_id = query.message.chat_id
        
        # Проверяем, что пользователь зарегистрирован
        db = next(get_db())
        user_service = UserService(db)
        user = user_service.get_user_by_telegram_id(user_id)
        
        if not user:
            await query.message.reply_text("Сначала зарегистрируйтесь с помощью /start")
            return
        
        # Переводим в состояние создания команды
        self.user_states[user_id] = {
            "state": "creating_team",
            "step": "team_name"
        }
        
        await query.message.reply_text("Введите название команды:")
    
    async def handle_manage_teams(self, query):
        """Обработка управления командами"""
        user_id = query.from_user.id
        chat_id = query.message.chat_id
        
        # TODO: Реализовать управление командами
        await query.message.reply_text("Функция управления командами в разработке")
    
    async def handle_take_survey(self, query):
        """Обработка прохождения опроса"""
        user_id = query.from_user.id
        chat_id = query.message.chat_id
        
        # TODO: Реализовать прохождение опроса
        await query.message.reply_text("Функция прохождения опроса в разработке")
    
    async def handle_answer_question(self, query, data: str):
        """Обработка ответа на вопрос"""
        question_id = int(data.split("_")[1])
        
        # Переводим в состояние ответа на вопрос
        user_id = query.from_user.id
        self.user_states[user_id] = {
            "state": "answering_question",
            "question_id": question_id
        }
        
        await query.message.reply_text("Введите ваш ответ:")
    
    async def handle_message(self, update: Update, context: ContextTypes.DEFAULT_TYPE):
        """Обработка текстовых сообщений"""
        user_id = update.effective_user.id
        chat_id = update.effective_chat.id
        text = update.message.text
        
        if user_id in self.user_states:
            state = self.user_states[user_id]["state"]
            
            if state == "creating_team":
                await self.handle_team_creation(update, text)
            elif state == "answering_question":
                await self.handle_question_answer(update, text)
            else:
                await update.message.reply_text("Неизвестное состояние. Попробуйте /start")
        else:
            await update.message.reply_text("Используйте /start для начала работы с ботом")
    
    async def handle_team_creation(self, update: Update, text: str):
        """Обработка создания команды пошагово"""
        user_id = update.effective_user.id
        chat_id = update.effective_chat.id
        
        if self.user_states[user_id]["step"] == "team_name":
            # Сохраняем название команды
            self.user_states[user_id]["team_name"] = text
            self.user_states[user_id]["step"] = "team_description"
            
            await update.message.reply_text("Введите описание команды (или отправьте '-' для пропуска):")
        
        elif self.user_states[user_id]["step"] == "team_description":
            # Сохраняем описание команды
            description = text if text != "-" else None
            self.user_states[user_id]["team_description"] = description
            self.user_states[user_id]["step"] = "telegram_group"
            
            await update.message.reply_text("Введите ID Telegram группы (или отправьте '-' для пропуска):")
        
        elif self.user_states[user_id]["step"] == "telegram_group":
            # Сохраняем ID группы
            group_id = text if text != "-" else None
            self.user_states[user_id]["telegram_group_id"] = group_id
            self.user_states[user_id]["step"] = "telegram_topic"
            
            await update.message.reply_text("Введите ID топика в группе (или отправьте '-' для пропуска):")
        
        elif self.user_states[user_id]["step"] == "telegram_topic":
            # Создаем команду
            topic_id = text if text != "-" else None
            self.user_states[user_id]["telegram_topic_id"] = topic_id
            
            await self.create_team_from_state(user_id, chat_id)
    
    async def create_team_from_state(self, user_id: int, chat_id: int):
        """Создание команды из сохраненного состояния"""
        try:
            db = next(get_db())
            team_service = TeamService(db)
            user_service = UserService(db)
            
            # Получаем пользователя
            user = user_service.get_user_by_telegram_id(user_id)
            if not user:
                raise Exception("Пользователь не найден")
            
            # Создаем команду
            team_data = TeamCreate(
                name=self.user_states[user_id]["team_name"],
                description=self.user_states[user_id].get("team_description"),
                telegram_group_id=self.user_states[user_id].get("telegram_group_id"),
                telegram_topic_id=self.user_states[user_id].get("telegram_topic_id")
            )
            
            team = team_service.create_team(team_data, user.id)
            
            # Очищаем состояние
            del self.user_states[user_id]
            
            # Отправляем подтверждение
            await self.send_success_message(chat_id, f"Команда '{team.name}' успешно создана!")
            
            # Отправляем меню менеджера
            await self.send_manager_menu(chat_id)
            
        except Exception as e:
            await self.send_error_message(chat_id, f"Ошибка создания команды: {str(e)}")
    
    async def handle_question_answer(self, update: Update, text: str):
        """Обработка ответа на вопрос"""
        user_id = update.effective_user.id
        chat_id = update.effective_chat.id
        
        try:
            question_id = self.user_states[user_id]["question_id"]
            
            # TODO: Сохранение ответа в базу данных
            
            # Очищаем состояние
            del self.user_states[user_id]
            
            await self.send_success_message(chat_id, "Ответ сохранен!")
            
        except Exception as e:
            await self.send_error_message(chat_id, f"Ошибка сохранения ответа: {str(e)}")
    
    async def send_success_message(self, chat_id: int, text: str):
        """Отправка сообщения об успехе"""
        await self.application.bot.send_message(
            chat_id=chat_id,
            text=f"✅ {text}"
        )
    
    async def send_error_message(self, chat_id: int, text: str):
        """Отправка сообщения об ошибке"""
        await self.application.bot.send_message(
            chat_id=chat_id,
            text=f"❌ {text}"
        )
    
    def run(self):
        """Запуск бота"""
        self.application.run_polling()
    
    async def run_async(self):
        """Асинхронный запуск бота"""
        await self.application.initialize()
        await self.application.start()
        await self.application.updater.start_polling()
        
        try:
            await asyncio.Event().wait()
        finally:
            await self.application.updater.stop()
            await self.application.stop()
            await self.application.shutdown()