shopbot/bot/getters/history.py

60 lines
2.2 KiB
Python
Raw Normal View History

2024-12-24 18:08:03 +00:00
import logging
2024-12-25 01:30:33 +00:00
from typing import Any, Dict
2024-12-24 18:08:03 +00:00
from aiogram_dialog import DialogManager
from sqlalchemy import select
from sqlalchemy.exc import SQLAlchemyError
2024-12-25 01:30:33 +00:00
from shopbot.bot.database.db import async_session
from shopbot.bot.database.models import Account
2024-12-24 18:08:03 +00:00
def _format_account_short_info(account: Account) -> str:
2024-12-25 01:30:33 +00:00
return f"🌍 {account.region} | 🎮 {account.current_rank} | 💰 {account.price:.2f}"
2024-12-24 18:08:03 +00:00
2024-12-25 01:30:33 +00:00
async def get_purchase_history(
dialog_manager: DialogManager, **kwargs
) -> Dict[str, Any]:
2024-12-24 18:08:03 +00:00
user_tg_id = dialog_manager.event.from_user.id
try:
async with async_session() as session:
2024-12-25 01:30:33 +00:00
result = await session.execute(
2024-12-24 18:08:03 +00:00
select(Account).where(Account.buyer == user_tg_id)
)
2024-12-25 01:30:33 +00:00
purchased_accounts = result.scalars().all()
2024-12-24 18:08:03 +00:00
if purchased_accounts:
2024-12-25 01:30:33 +00:00
text = "🛍️ Ваша история покупок:"
2024-12-24 18:08:03 +00:00
else:
logging.warning(f"User with tg_id {user_tg_id} has no purchases")
text = "🛍️ У вас пока нет покупок, пора купить что-нибудь)"
return {
"purchased_accounts": [
(_format_account_short_info(account), account.id)
for account in purchased_accounts
],
2024-12-25 01:30:33 +00:00
"text": text,
2024-12-24 18:08:03 +00:00
}
except SQLAlchemyError as e:
logging.error(f"Database error: {e}", exc_info=True)
2024-12-25 01:30:33 +00:00
return {
"purchased_accounts": [],
"text": "❌ Произошла ошибка при работе с базой данных.",
}
2024-12-24 18:08:03 +00:00
except Exception as e:
logging.error(f"An unexpected error occurred: {e}", exc_info=True)
return {"purchased_accounts": [], "text": "❌ Произошла непредвиденная ошибка."}
2024-12-25 01:30:33 +00:00
async def get_history_account_details(
dialog_manager: DialogManager, **kwargs
) -> Dict[str, str]:
2024-12-24 18:08:03 +00:00
account_info = dialog_manager.current_context().dialog_data.get("account_info")
if not account_info:
logging.warning("No account info found in dialog data")
return {"account_info": "❌ Информация об аккаунте недоступна."}
2024-12-25 01:30:33 +00:00
return {"account_info": account_info}