"""Handlers for private messages.""" import re from typing import Optional, Dict import telebot from telebot import types from sqlalchemy.orm import Session from database import get_db_session, User, Chat, UserChat from messages import THEMES, get_theme_emoji # User states for conversation flow user_states: Dict[int, str] = {} def register_private_handlers(bot: telebot.TeleBot) -> None: """Register all private message handlers.""" @bot.message_handler(commands=['start'], chat_types=['private']) def handle_start(message: telebot.types.Message) -> None: """Handle /start command in private chat.""" if not message.from_user: return user_id = message.from_user.id username: Optional[str] = message.from_user.username first_name: str = message.from_user.first_name or "Пользователь" db = get_db_session() try: # Check if user exists user = db.query(User).filter(User.user_id == user_id).first() if user: # User already exists - ask if they want to update bot.send_message( user_id, f"Привет, {first_name}! Ты уже поделился со мной днем рождения.\n" f"Используй /update, чтобы обновить свои данные." ) else: # New user - ask for birthday user_states[user_id] = 'waiting_birthday' bot.send_message( user_id, f"Привет, {first_name}! 👋\n\n" f"Пришли мне свой день рождения в формате ДД.ММ или ДД.ММ.ГГГГ\n" f"Например: 15.03 или 15.03.1990" ) finally: db.close() @bot.message_handler(commands=['update'], chat_types=['private']) def handle_update(message: telebot.types.Message) -> None: """Handle /update command in private chat.""" if not message.from_user: return user_id = message.from_user.id first_name: str = message.from_user.first_name or "Пользователь" db = get_db_session() try: user = db.query(User).filter(User.user_id == user_id).first() if user: user_states[user_id] = 'waiting_birthday' bot.send_message( user_id, f"Хорошо, {first_name}! Пришли мне свой день рождения в формате ДД.ММ или ДД.ММ.ГГГГ" ) else: bot.send_message( user_id, "Ты еще не поделился со мной днем рождения. Используй /start для начала." ) finally: db.close() @bot.message_handler(func=lambda m: m.chat.type == 'private' and m.from_user and m.from_user.id in user_states and m.text) def handle_birthday_input(message: telebot.types.Message) -> None: """Handle birthday input from user.""" if not message.from_user or not message.text: return user_id = message.from_user.id state = user_states.get(user_id) if state == 'waiting_birthday': # Parse birthday birthday_text = message.text.strip() parsed = parse_birthday(birthday_text) if not parsed: bot.send_message( user_id, "Неверный формат даты. Используй формат ДД.ММ или ДД.ММ.ГГГГ\n" "Например: 15.03 или 15.03.1990" ) return day, month, year = parsed # Validate date if not is_valid_date(day, month, year): bot.send_message( user_id, "Неверная дата. Проверь правильность дня и месяца." ) return # Save birthday and ask for preference db = get_db_session() try: if not message.from_user: return username: Optional[str] = message.from_user.username first_name: str = message.from_user.first_name or "Пользователь" user = db.query(User).filter(User.user_id == user_id).first() if user: # Update existing user user.birthday_day = day user.birthday_month = month user.birthday_year = year user.first_name = first_name user.username = username else: # Create new user user = User( user_id=user_id, username=username, first_name=first_name, birthday_day=day, birthday_month=month, birthday_year=year, preference_theme="Музыка" # Default theme ) db.add(user) db.commit() # Ask for preference theme user_states[user_id] = 'waiting_preference' ask_preference_theme(bot, user_id) finally: db.close() elif state == 'waiting_preference': # User selected preference theme if not message.text: return theme_text = message.text.strip() if theme_text not in THEMES: bot.send_message( user_id, "Пожалуйста, выбери одну из предложенных тем, нажав на кнопку." ) return # Save preference db = get_db_session() try: user = db.query(User).filter(User.user_id == user_id).first() if user: user.preference_theme = theme_text db.commit() # Update user in all chats update_user_in_chats(bot, db, user) bot.send_message( user_id, f"Отлично! Я запомнил твои предпочтения: {theme_text}\n\n" f"Теперь я буду поздравлять тебя с днем рождения во всех чатах, где ты состоишь!" ) user_states.pop(user_id, None) finally: db.close() @bot.callback_query_handler(func=lambda call: call.data and call.data.startswith('theme_')) def handle_theme_selection(call: telebot.types.CallbackQuery) -> None: """Handle theme selection from inline keyboard.""" if not call.from_user or not call.data or not call.message: return user_id = call.from_user.id if user_states.get(user_id) != 'waiting_preference': bot.answer_callback_query(call.id, "Это действие недоступно сейчас.") return theme = call.data.replace('theme_', '') if theme not in THEMES: bot.answer_callback_query(call.id, "Неверная тема.") return # Save preference db = get_db_session() try: user = db.query(User).filter(User.user_id == user_id).first() if user: user.preference_theme = theme db.commit() # Update user in all chats update_user_in_chats(bot, db, user) bot.answer_callback_query(call.id, f"Выбрано: {theme}") bot.edit_message_text( f"Отлично! Я запомнил твои предпочтения: {theme}\n\n" f"Теперь я буду поздравлять тебя с днем рождения во всех чатах, где ты состоишь!", chat_id=call.message.chat.id, message_id=call.message.message_id ) user_states.pop(user_id, None) finally: db.close() def parse_birthday(text: str) -> Optional[tuple[int, int, Optional[int]]]: """Parse birthday from text. Returns (day, month, year) or None.""" # Try DD.MM.YYYY format match = re.match(r'^(\d{1,2})\.(\d{1,2})\.(\d{4})$', text) if match: day, month, year = int(match.group(1)), int(match.group(2)), int(match.group(3)) return (day, month, year) # Try DD.MM format match = re.match(r'^(\d{1,2})\.(\d{1,2})$', text) if match: day, month = int(match.group(1)), int(match.group(2)) return (day, month, None) return None def is_valid_date(day: int, month: int, year: Optional[int]) -> bool: """Check if date is valid.""" if month < 1 or month > 12: return False if day < 1 or day > 31: return False # Check specific month limits if month in [4, 6, 9, 11] and day > 30: return False if month == 2: if year: # Check leap year if (year % 4 == 0 and year % 100 != 0) or (year % 400 == 0): max_day = 29 else: max_day = 28 else: max_day = 29 # Assume leap year if year not provided if day > max_day: return False return True def ask_preference_theme(bot: telebot.TeleBot, user_id: int) -> None: """Ask user to select preference theme.""" keyboard = types.InlineKeyboardMarkup(row_width=2) # Show hobbies (themes) in 2 buttons per row for i in range(0, len(THEMES), 2): theme1 = THEMES[i] emoji1 = get_theme_emoji(theme1) btn1 = types.InlineKeyboardButton( text=f"{emoji1} {theme1}", callback_data=f'theme_{theme1}' ) # Optional second button in the same row if i + 1 < len(THEMES): theme2 = THEMES[i + 1] emoji2 = get_theme_emoji(theme2) btn2 = types.InlineKeyboardButton( text=f"{emoji2} {theme2}", callback_data=f'theme_{theme2}' ) keyboard.add(btn1, btn2) else: keyboard.add(btn1) bot.send_message( user_id, "Что тебе нравится? Выбери одну из тем:", reply_markup=keyboard ) def update_user_in_chats(bot: telebot.TeleBot, db: Session, user: User) -> None: """Update user information in all chats where bot is admin.""" # Get all chats where bot is admin admin_chats = db.query(Chat).filter(Chat.bot_is_admin == True).all() for chat in admin_chats: try: # Check if user is member of this chat member = bot.get_chat_member(chat.chat_id, user.user_id) if member.status not in ['left', 'kicked']: # User is in chat - add/update relationship user_chat = db.query(UserChat).filter( UserChat.user_id == user.user_id, UserChat.chat_id == chat.chat_id ).first() if not user_chat: user_chat = UserChat(user_id=user.user_id, chat_id=chat.chat_id) db.add(user_chat) db.commit() except Exception: # User might have blocked bot or bot was removed from chat pass