timer_bot.py - master - KKleinikov/telegram_bot_timer

timer_bot.py
396 строк13 KB
Konstantin-Kleinikov
1.05 refactor: Add full type annotations and improve code quality.
07 ноя 2025, 23:41
07 ноя 2025, 23:417591d70
"""
Telegram Timer Bot

A fully asynchronous bot that allows users to set timers using commands like /timer 30s, /timer 10m, or /timer 1h.
Supports time ranges from 30 seconds to 24 hours with proper validation, rate limiting, and access control.

Key features:
- Time parsing with support for seconds (s), minutes (m), and hours (h)
- Access control via whitelist and blacklist (configured through environment variables)
- Rate limiting to prevent abuse (5 seconds between requests)
- Graceful error handling and detailed logging
- Reply keyboard shortcuts for quick timer setup
- Proper task cancellation and cleanup
- Internationalized Russian time unit forms (секунда, минута, час)
- Configurable via environment variables (BOT_TOKEN, WHITELIST, BLACKLIST)

Environment variables:
    BOT_TOKEN: Telegram bot token (required)
    WHITELIST: Comma-separated user IDs allowed to use the bot (optional, empty = all allowed)
    BLACKLIST: Comma-separated user IDs blocked from using the bot (optional)

Usage:
    /start - Start the bot
    /timer 30s - Set a 30-second timer
    /cancel - Cancel the active timer

The bot uses aiogram 3.x, Python 3.13+, and follows best practices for logging, error handling, and code quality.
"""

import asyncio
import contextlib
import logging
import os
import re
import time

from enum import Enum
from re import Pattern
from typing import Any, Optional

from aiogram import Bot, Dispatcher, F
from aiogram.client.default import DefaultBotProperties
from aiogram.enums import ParseMode
from aiogram.filters import Command, CommandStart
from aiogram.types import (
    BotCommand,
    KeyboardButton,
    Message,
    ReplyKeyboardMarkup,
)
from dotenv import load_dotenv

load_dotenv()

# Configuration
BOT_TOKEN: str = os.getenv('BOT_TOKEN')  # type: ignore
if not BOT_TOKEN:
    raise ValueError('BOT_TOKEN not found in environment variables')

# Whitelist and Blacklist
WHITELIST_STR: str = os.getenv('WHITELIST', '')
WHITELIST: list[int] = [
    int(uid.strip()) for uid in WHITELIST_STR.split(',') if uid.strip().isdigit()
] if WHITELIST_STR else []

BLACKLIST_STR: str = os.getenv('BLACKLIST', '')
BLACKLIST: list[int] = [
    int(uid.strip()) for uid in BLACKLIST_STR.split(',') if uid.strip().isdigit()
] if BLACKLIST_STR else []

# Constants
MIN_TIMER_SECONDS: int = 30
MAX_TIMER_SECONDS: int = 86400  # 24 hours
MIN_REQUEST_INTERVAL: float = 5.0

# Logging setup
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
)
logger: logging.Logger = logging.getLogger(__name__)

# Timer storage
timers: dict[int, asyncio.Task] = {}
last_request_time: dict[int, float] = {}

# Regex pattern for time parsing
_TIME_PATTERN: Pattern[str] = re.compile(r'^(\d+)([smh])$')


class TimeUnit(Enum):
    """
    Enumeration for time units with symbol, multiplier, and Russian forms.
    """
    SECONDS = ('s', 1, 'секунда', 'секунды', 'секунд')
    MINUTES = ('m', 60, 'минута', 'минуты', 'минут')
    HOURS = ('h', 3600, 'час', 'часа', 'часов')

    def __init__(self, symbol: str, multiplier: int, one: str, few: str, many: str) -> None:
        self.symbol: str = symbol
        self.multiplier: int = multiplier
        self.forms: tuple[str, str, str] = (one, few, many)

    def get_name(self, number: int) -> str:
        """
        Returns correct Russian form based on number.
        """
        if number % 10 == 1 and number % 100 != 11:
            return self.forms[0]  # singular
        if 2 <= number % 10 <= 4 and (number % 100 < 10 or number % 100 >= 20):
            return self.forms[1]  # few
        return self.forms[2]  # many

    @classmethod
    def from_symbol(cls, symbol: str) -> Optional['TimeUnit']:
        """
        Returns TimeUnit by symbol (s, m, h).
        """
        for unit in cls:
            if unit.symbol == symbol:
                return unit
        return None


def parse_time(time_str: str) -> tuple[int, int, str] | None:
    """
    Parses time string like '10m', '30s', '2h'.
    Returns (seconds, number, unit_full) or None.
    """
    match = _TIME_PATTERN.match(time_str.lower().strip())
    if not match:
        return None

    number = int(match.group(1))
    unit_symbol = match.group(2)

    unit = TimeUnit.from_symbol(unit_symbol)
    if not unit:
        return None

    seconds = number * unit.multiplier
    unit_full = unit.get_name(number)

    if seconds < MIN_TIMER_SECONDS or seconds > MAX_TIMER_SECONDS:
        return None

    return seconds, number, unit_full


def check_rate_limit(user_id: int) -> bool:
    """
    Checks if enough time has passed since last request.
    Uses monotonic time to prevent issues with system clock adjustments.
    Returns True if request can be processed.
    """
    current_time: float = time.monotonic()
    last_time: float = last_request_time.get(user_id, 0)

    if current_time - last_time < MIN_REQUEST_INTERVAL:
        return False

    last_request_time[user_id] = current_time
    return True


def check_access(user_id: int) -> bool:
    """
    Checks if user has access to the bot.
    Returns False if user is blacklisted or not in whitelist.
    """
    if user_id in BLACKLIST:
        logger.warning('Blocked user %s attempted to use bot', user_id)
        return False

    if WHITELIST and user_id not in WHITELIST:
        logger.warning('Unauthorized user %s attempted to use bot', user_id)
        return False

    return True


async def timer_task(bot: Bot, chat_id: int, user_id: int, delay: int) -> None:
    """
    Asynchronous timer task: waits and sends notification.
    Handles cancellation and errors during execution.
    """
    try:
        await asyncio.sleep(delay)
        await bot.send_message(chat_id, '⏰ Время вышло!')
        logger.info('Timer expired for user %s', user_id)
    except asyncio.CancelledError:
        logger.info('Timer for user %s was cancelled', user_id)
        raise
    except Exception:
        logger.exception('Error in timer_task for user %s', user_id)
    finally:
        timers.pop(user_id, None)


async def cancel_timer(user_id: int) -> bool:
    """
    Cancels active timer for user.
    Returns True if timer was cancelled.
    """
    if user_id not in timers:
        return False

    task: asyncio.Task = timers[user_id]
    if not task.done():
        task.cancel()
        with contextlib.suppress(asyncio.CancelledError):
            await task

    timers.pop(user_id, None)
    logger.info('Timer cancelled for user %s', user_id)
    return True


def get_main_keyboard() -> ReplyKeyboardMarkup:
    """
    Returns reply keyboard with time shortcut buttons.
    """
    return ReplyKeyboardMarkup(
        keyboard=[
            [KeyboardButton(text='30s'), KeyboardButton(text='10m'), KeyboardButton(text='1h')],
            [KeyboardButton(text='/cancel')],
        ],
        resize_keyboard=True,
        one_time_keyboard=False,
        input_field_placeholder='Press to insert example',
    )


async def set_timer(message: Message, bot: Bot, time_str: str) -> None:
    """
    Shared helper to set a timer.
    Validates user access, rate limit, time format, and starts the timer.
    """
    user_id: int = message.from_user.id

    if not check_access(user_id):
        await message.answer('❌ У вас нет доступа к этому боту.')  # noqa: RUF001
        return

    if not check_rate_limit(user_id):
        await message.answer('⏳ Пожалуйста, подождите 5 секунд между запросами.')
        return

    parsed: tuple[int, int, str] | None = parse_time(time_str)
    if parsed is None:
        await message.answer(
            '❌ Неверный формат времени.\n'
            'Допустимые диапазоны: от 30s до 24h\n'
            'Примеры: 30s, 10m, 2h'
        )
        return

    seconds, number, unit_full = parsed
    await cancel_timer(user_id)

    task: asyncio.Task = asyncio.create_task(timer_task(bot, message.chat.id, user_id, seconds))
    timers[user_id] = task

    logger.info('Timer set for user %s: %s seconds (%s %s)', user_id, seconds, number, unit_full)
    await message.answer(
        f'✅ Таймер на {number} {unit_full} установлен',
        reply_markup=get_main_keyboard(),
    )


async def main() -> None:
    """
    Main function to start the bot.
    """
    logger.info('Starting bot...')

    bot: Bot = Bot(token=BOT_TOKEN, default=DefaultBotProperties(parse_mode=ParseMode.HTML))
    dp: Dispatcher = Dispatcher()

    @dp.message(CommandStart())
    async def cmd_start(message: Message) -> None:
        if message.chat.type != 'private':
            return

        user_id: int = message.from_user.id
        if not check_access(user_id):
            await message.answer('❌ У вас нет доступа к этому боту.')  # noqa: RUF001
            return

        logger.info('User %s started bot', user_id)
        await message.answer(
            '👋 Привет! Я бот для таймеров.\n\n'
            '💡 Используй команды ниже или введи:\n'
            '/timer 30s — таймер на 30 секунд\n'
            '/timer 10m — таймер на 10 минут\n'
            '/timer 1h — таймер на 1 час\n\n'
            '🚫 /cancel — отменить таймер',
            parse_mode=None,
        )
        await message.answer(
            'Выберите пример времени:',
            reply_markup=get_main_keyboard()
        )

    @dp.message(Command('timer'))
    async def cmd_timer(message: Message) -> None:
        if message.chat.type != 'private':
            return

        args: list[str] = message.text.split(maxsplit=1)
        if len(args) < 2:
            await message.answer(
                '❌ Неверный формат команды.\n'
                'Используйте: /timer &lt;время&gt;\n'
                'Примеры: /timer 30s, /timer 10m, /timer 2h'
            )
            return

        time_str: str = args[1].strip()
        await set_timer(message, bot, time_str)

    @dp.message(Command('cancel'))
    async def cmd_cancel(message: Message) -> None:
        if message.chat.type != 'private':
            return

        user_id: int = message.from_user.id
        if not check_access(user_id):
            await message.answer('❌ У вас нет доступа к этому боту.')  # noqa: RUF001
            return

        if not check_rate_limit(user_id):
            await message.answer('⏳ Пожалуйста, подождите 5 секунд между запросами.')
            return

        if await cancel_timer(user_id):
            await message.answer('🚫 Таймер отменён')
        else:
            await message.answer('ℹ️ У вас нет активного таймера')  # noqa: RUF001

    @dp.message(F.chat.type == 'private')
    async def other_messages(message: Message) -> None:
        user_id: int = message.from_user.id
        if not check_access(user_id):
            await message.answer('❌ У вас нет доступа к этому боту.')  # noqa: RUF001
            return

        text: str = message.text.strip()
        # Use set for O(1) membership test
        if text in {'30s', '1m', '10m', '30m', '1h', '2h'}:
            await set_timer(message, bot, text)
            return

        await message.answer(
            '💡 Используйте команды:\n'
            '/timer &lt;время&gt; — установить таймер\n'
            'Например: /timer 30s, /timer 10m\n\n'
            'Или выберите пример ниже:',
            reply_markup=get_main_keyboard(),
        )

    @dp.message(F.text.in_(['30s', '10m', '1h']))
    async def handle_time_shortcut(message: Message) -> None:
        await set_timer(message, bot, message.text)

    # Register handlers
    # NOTE: handlers are used by dp.start_polling() — vulture cannot detect this
    _handlers: list[Any] = [
        cmd_start,
        cmd_timer,
        cmd_cancel,
        other_messages,
        handle_time_shortcut,
    ]

    await bot.set_my_commands([
        BotCommand(command='start', description='Запустить бота'),
        BotCommand(command='timer', description='Например: /timer 30s'),
        BotCommand(command='cancel', description='Отменить таймер'),
    ])

    logger.info('Commands set: /start, /timer, /cancel')

    try:
        await dp.start_polling(bot)
    except Exception:
        logger.exception('Bot stopped with error')
    finally:
        await bot.session.close()


if __name__ == '__main__':
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logger.info('Bot stopped by user')