Туториал: Telegram-бот для конвертации валют

python --version

python -m venv .venv

# активировать:
.venv\Scripts\activate      # Windows
source .venv/bin/activate   # Linux / macOS

pip install -r requirements.txt

API_URL = "https://open.er-api.com/v6/latest/{base}"
TIMEOUT_S = 5

CURRENCIES = ["USD", "EUR", "RUB", "GBP", "CNY", "JPY", "CHF", "TRY"]
BUTTONS_PER_ROW = 4

# Состояния ConversationHandler
SELECT_FROM  = 0
SELECT_TO    = 1
ENTER_AMOUNT = 2

def load_env_file(path: str = ".env") -> None:
    if not os.path.exists(path):
        return
    with open(path, "r", encoding="utf-8") as f:
        for raw_line in f:
            line = raw_line.strip()
            # пропускаем пустые строки и комментарии
            if not line or line.startswith("#") or "=" not in line:
                continue
            key, value = line.split("=", 1)
            key   = key.strip()
            value = value.strip().strip('"').strip("'")
            # не перезаписываем уже установленные переменные
            if key and key not in os.environ and value:
                os.environ[key] = value

async def fetch_rates(base: str) -> Optional[dict]:
    url = API_URL.format(base=base)

    def _do_request() -> dict:
        return requests.get(url, timeout=TIMEOUT_S).json()

    try:
        # requests — синхронная библиотека.
        # asyncio.to_thread() запускает её в пуле потоков,
        # не блокируя event loop бота.
        data = await asyncio.to_thread(_do_request)
    except Exception:
        logging.exception("API request failed: %s", url)
        return None

    # валидируем ответ
    if data.get("result") != "success":
        logging.error("API error: %s", data.get("error-type"))
        return None

    rates = data.get("rates")
    if not isinstance(rates, dict):
        return None

    return rates


async def get_rate(from_code: str, to_code: str) -> Optional[float]:
    rates = await fetch_rates(from_code)
    if not rates:
        return None
    try:
        return float(rates.get(to_code))
    except (TypeError, ValueError):
        return None

def build_currency_keyboard(
    *, exclude: Optional[str] = None, prefix: str
) -> InlineKeyboardMarkup:
    items = [c for c in CURRENCIES if c != exclude]
    rows = []
    for i in range(0, len(items), BUTTONS_PER_ROW):
        row = [
            InlineKeyboardButton(text=code, callback_data=f"{prefix}:{code}")
            for code in items[i : i + BUTTONS_PER_ROW]
        ]
        rows.append(row)
    return InlineKeyboardMarkup(rows)

# Шаг 1 — клавиатура ИЗ
build_currency_keyboard(prefix="FROM")
# → callback_data = "FROM:USD", "FROM:EUR", ...

# Шаг 2 — клавиатура В (без уже выбранной валюты)
build_currency_keyboard(exclude="USD", prefix="TO")
# → callback_data = "TO:EUR", "TO:RUB", ...

async def rate_cmd(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None:
    msg  = update.effective_message
    args = context.args

    if len(args) < 2 or len(args) > 3:
        await msg.reply_text("Формат: /rate <ИЗ> <В> [сумма]")
        return

    from_code = args[0].upper()
    to_code   = args[1].upper()

    if from_code not in CURRENCIES or to_code not in CURRENCIES:
        await msg.reply_text(f"Неподдерживаемая валюта. Доступны: {', '.join(CURRENCIES)}")
        return

    if from_code == to_code:
        await msg.reply_text("Валюты «ИЗ» и «В» должны отличаться.")
        return

    amount: Optional[float] = None
    if len(args) == 3:
        try:
            amount = float(args[2].replace(",", ""))
        except ValueError:
            await msg.reply_text("Сумма должна быть числом, например 100 или 12.5")
            return
        if amount <= 0:
            await msg.reply_text("Сумма должна быть больше 0.")
            return

    rate = await get_rate(from_code, to_code)
    if rate is None:
        await msg.reply_text(f"Не удалось получить курс для {from_code} → {to_code}.")
        return

    if amount is None:
        await msg.reply_text(f"1 {from_code} = {format_rate(rate)} {to_code}")
    else:
        result = amount * rate
        await msg.reply_text(
            f"{format_amount(amount)} {from_code} = {format_amount(result)} {to_code}\n"
            f"Курс: {format_rate(rate)}"
        )

async def convert_entry(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    await update.effective_message.reply_text(
        "Выбери валюту, ИЗ которой конвертируем:",
        reply_markup=build_currency_keyboard(prefix="FROM"),
    )
    return SELECT_FROM

async def convert_select_from(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    query = update.callback_query
    await query.answer()
    _, code = query.data.split(":", 1)          # "FROM:USD" → "USD"
    context.user_data["from"] = code
    await query.edit_message_text(
        f"Валюта ИЗ: {code}\n\nТеперь выбери валюту, В которую:",
        reply_markup=build_currency_keyboard(exclude=code, prefix="TO"),
    )
    return SELECT_TO

async def convert_enter_amount(update: Update, context: ContextTypes.DEFAULT_TYPE) -> int:
    amount    = float(update.effective_message.text.replace(",", ""))
    from_code = context.user_data["from"]
    to_code   = context.user_data["to"]
    rate      = await get_rate(from_code, to_code)
    result    = amount * rate
    await update.effective_message.reply_text(
        f"{format_amount(amount)} {from_code} = {format_amount(result)} {to_code}\n"
        f"Курс: {format_rate(rate)}"
    )
    context.user_data.pop("from", None)
    context.user_data.pop("to", None)
    return ConversationHandler.END

def build_app(token: str) -> Application:
    request = HTTPXRequest(
        connect_timeout=20, read_timeout=20,
        write_timeout=20,   pool_timeout=20
    )
    app = (
        Application.builder()
        .token(token)
        .request(request)
        .post_init(post_init)
        .build()
    )

    app.add_handler(CommandHandler("start",  start_cmd))
    app.add_handler(CommandHandler("help",   help_cmd))
    app.add_handler(CommandHandler("rate",   rate_cmd))

    conv = ConversationHandler(
        entry_points=[CommandHandler("convert", convert_entry)],
        states={
            SELECT_FROM:  [CallbackQueryHandler(convert_select_from,  pattern=r"^FROM:")],
            SELECT_TO:    [CallbackQueryHandler(convert_select_to,    pattern=r"^TO:")],
            ENTER_AMOUNT: [MessageHandler(filters.TEXT & ~filters.COMMAND, convert_enter_amount)],
        },
        fallbacks=[CommandHandler("cancel", cancel_cmd)],
        allow_reentry=True,
    )
    app.add_handler(conv)
    app.add_handler(CommandHandler("cancel", cancel_cmd))
    return app

python bot.py