import asyncio import logging from aiogram import Bot, Dispatcher, F from aiogram.filters import Command, CommandObject from aiogram.types import ( CallbackQuery, InlineKeyboardButton, InlineKeyboardMarkup, KeyboardButton, Message, ReplyKeyboardMarkup, WebAppInfo, ) from app.core.config import settings from bot.api_client import ApiClient logging.basicConfig(level=logging.INFO) dp = Dispatcher() api = ApiClient() def main_keyboard() -> ReplyKeyboardMarkup: return ReplyKeyboardMarkup( keyboard=[ [KeyboardButton(text="Открыть CarPass")], [KeyboardButton(text="Мои авто"), KeyboardButton(text="Помощь")], ], resize_keyboard=True, ) def webapp_inline_keyboard() -> InlineKeyboardMarkup: return InlineKeyboardMarkup( inline_keyboard=[ [InlineKeyboardButton(text="Открыть CarPass", web_app=WebAppInfo(url=settings.effective_webapp_url))], ] ) @dp.message(Command("start")) async def start(message: Message) -> None: user = await api.upsert_user(message.from_user) text = ( f"Готово, {user.get('first_name') or 'водитель'}.\n\n" "CarPass — цифровой паспорт автомобиля: заправки, обслуживание, напоминания, подтвержденная история и стоимость владения.\n\n" "Нажми «Открыть CarPass», чтобы перейти в приложение." ) await message.answer(text, reply_markup=webapp_inline_keyboard()) await message.answer("Клавиатура ниже открывает меню бота. Сам Mini App запускается кнопкой в сообщении выше.", reply_markup=main_keyboard()) @dp.message(Command("add_car")) async def add_car(message: Message, command: CommandObject) -> None: user = await api.upsert_user(message.from_user) name = command.args.strip() if command.args else "" if not name: await message.answer("Напиши так: /add_car Toyota Camry") return car = await api.create_car(user["id"], name, message.from_user.id) await message.answer(f"Добавил авто: {car['name']}") @dp.message(Command("cars")) @dp.message(F.text == "Мои авто") async def cars(message: Message) -> None: user = await api.upsert_user(message.from_user) items = await api.list_cars(user["id"], message.from_user.id) if not items: await message.answer("Автомобилей пока нет. Добавь через mini app или командой /add_car Название.") return buttons = [ [InlineKeyboardButton(text=car["name"], callback_data=f"stats:{car['id']}")] for car in items ] await message.answer("Твой гараж:", reply_markup=InlineKeyboardMarkup(inline_keyboard=buttons)) @dp.callback_query(F.data.startswith("stats:")) async def show_stats(callback: CallbackQuery) -> None: car_id = int(callback.data.split(":", 1)[1]) stats = await api.stats(car_id, callback.from_user.id) consumption = stats["avg_consumption_l_per_100km"] cost_per_km = stats["cost_per_km"] await callback.message.answer( "\n".join( [ "Статистика авто:", f"Расходы всего: {stats['total_cost']}", f"Топливо: {stats['fuel_cost']}", f"Сервис и ремонты: {stats['service_cost']}", f"Пробег по записям: {stats['distance_km']} км", f"Средний расход: {consumption:.2f} л/100 км" if consumption else "Средний расход: нет данных", f"Стоимость 1 км: {cost_per_km:.2f}" if cost_per_km else "Стоимость 1 км: нет данных", ] ) ) await callback.answer() @dp.message(F.text == "Помощь") @dp.message(Command("help")) async def help_message(message: Message) -> None: await message.answer( "CarPass помогает вести цифровой паспорт автомобиля.\n\n" "Что можно делать:\n" "• добавлять автомобили и параметры обслуживания;\n" "• вести заправки, ТО, ремонт, страховку, налоги и штрафы;\n" "• видеть стоимость владения, стоимость 1 км и прогноз расходов;\n" "• загрузить чек, проверить распознанные данные и сохранить запись;\n" "• привязать авто к проверенному СТО и подтверждать сервисную историю;\n" "• зарегистрировать СТО и отправить заявку на проверку.\n\n" "Mini App нужно открывать кнопкой под этим сообщением: так Telegram передает защищенную авторизацию.", reply_markup=webapp_inline_keyboard(), ) @dp.message(F.text == "Открыть CarPass") @dp.message(F.text == "Открыть гараж") async def open_carpass(message: Message) -> None: await message.answer( "Открой CarPass кнопкой ниже. Это правильный Telegram Mini App вход с авторизацией.", reply_markup=webapp_inline_keyboard(), ) async def main() -> None: if not settings.bot_token: raise RuntimeError("BOT_TOKEN is empty") if not settings.internal_api_token: raise RuntimeError("INTERNAL_API_TOKEN is empty") settings.validate_webapp_url_for_telegram() bot = Bot(settings.bot_token) await dp.start_polling(bot) if __name__ == "__main__": asyncio.run(main())