114 lines
5.3 KiB
Python
114 lines
5.3 KiB
Python
import logging
|
||
from typing import Any, Dict
|
||
|
||
from aiogram_dialog import DialogManager
|
||
from sqlalchemy import select
|
||
from sqlalchemy.exc import SQLAlchemyError
|
||
|
||
from shopbot.bot.database.db import async_session
|
||
from shopbot.bot.database.models import Account
|
||
from shopbot.bot.getters.filters import FilterDTO, FilterType
|
||
|
||
|
||
def _format_account_short_info(account: Account) -> str:
|
||
return (
|
||
f"🌍 {account.region} | 🎮 {account.current_rank} | "
|
||
f"🔪 {account.knife_count} | 💰 {account.price:.2f}₽"
|
||
)
|
||
|
||
|
||
def _format_account_details(account: Account) -> str:
|
||
return (
|
||
f"<b>Подробная информация об аккаунте:</b>\n"
|
||
f"🌍 Регион: {account.region}\n"
|
||
f"🎮 Ранг: {account.current_rank}\n"
|
||
f"🔪 Количество ножей: {account.knife_count}\n"
|
||
f"🛡️ Количество агентов: {account.agent_count}\n"
|
||
f"🖌️ Скины: {account.skin_count}\n"
|
||
f"🗡️ Стоимость инвентаря: {account.inventory_value}VP\n"
|
||
f"📈 Последний ранг: {account.last_rank}\n"
|
||
f"🏅 Уровень: {account.level}\n"
|
||
f"💰 Цена: {account.price:.2f}₽"
|
||
)
|
||
|
||
|
||
async def get_accounts(dialog_manager: DialogManager, **kwargs) -> Dict[str, Any]:
|
||
_filters: dict = dialog_manager.dialog_data.get("filters", {})
|
||
filters = [
|
||
FilterDTO(name=_.value, id=_.name, value=_filters.get(_.name, 0))
|
||
for _ in [FilterType[__] for __ in _filters]
|
||
]
|
||
try:
|
||
async with async_session() as session:
|
||
query = select(Account).where(Account.is_sold == False) # noqa: E712
|
||
if filters:
|
||
for _filter in filters:
|
||
if _filter.id == FilterType.PRICE_MAX.name:
|
||
query = query.where(Account.price <= _filter.value)
|
||
elif _filter.id == FilterType.PRICE_MIN.name:
|
||
query = query.where(Account.price >= _filter.value)
|
||
elif _filter.id == FilterType.LAST_RANK_MAX.name:
|
||
query = query.where(Account.last_rank <= _filter.value)
|
||
elif _filter.id == FilterType.LAST_RANK_MIN.name:
|
||
query = query.where(Account.last_rank >= _filter.value)
|
||
elif _filter.id == FilterType.SKIN_COUNT_MAX.name:
|
||
query = query.where(Account.skin_count <= _filter.value)
|
||
elif _filter.id == FilterType.SKIN_COUNT_MIN.name:
|
||
query = query.where(Account.skin_count >= _filter.value)
|
||
elif _filter.id == FilterType.KNIFE_COUNT_MAX.name:
|
||
query = query.where(Account.knife_count <= _filter.value)
|
||
elif _filter.id == FilterType.KNIFE_COUNT_MIN.name:
|
||
query = query.where(Account.knife_count >= _filter.value)
|
||
|
||
result = await session.execute(query)
|
||
accounts = result.scalars().all()
|
||
except SQLAlchemyError as e:
|
||
logging.error(f"Database error: {e}", exc_info=True)
|
||
return {
|
||
"accounts": [],
|
||
"text": "❌ Произошла ошибка при работе с базой данных.",
|
||
}
|
||
except Exception as e:
|
||
logging.error(f"An unexpected error occurred: {e}", exc_info=True)
|
||
return {"accounts": [], "text": "❌ Произошла непредвиденная ошибка."}
|
||
if accounts:
|
||
text = "😊 Пожалуйста, выберите один из доступных аккаунтов для продолжения."
|
||
else:
|
||
text = "😔 У нас сейчас нет аккаунтов. Приходи позже."
|
||
|
||
return {
|
||
"accounts": [
|
||
(_format_account_short_info(account), account.id) for account in accounts
|
||
],
|
||
"text": text,
|
||
}
|
||
|
||
|
||
async def get_account_details(dialog_manager, **kwargs) -> Dict[str, str]:
|
||
account_id = dialog_manager.current_context().dialog_data.get("selected_account_id")
|
||
if not account_id:
|
||
return {"account_info": "❌ Не выбран аккаунт!"}
|
||
|
||
try:
|
||
async with async_session() as session:
|
||
account = await session.scalar(
|
||
select(Account).where(Account.id == account_id)
|
||
)
|
||
if not account:
|
||
logging.warning(f"Account with id {account_id} not found.")
|
||
return {"account_info": "Аккаунт не найден!"}
|
||
return {"account_info": _format_account_details(account)}
|
||
except SQLAlchemyError as e:
|
||
logging.error(f"Database error: {e}", exc_info=True)
|
||
return {"account_info": "❌ Произошла ошибка при работе с базой данных."}
|
||
except Exception as e:
|
||
logging.error(f"An unexpected error occurred: {e}", exc_info=True)
|
||
return {"account_info": "❌ Произошла непредвиденная ошибка."}
|
||
|
||
|
||
async def get_account_info(dialog_manager, **kwargs) -> Dict[str, str]:
|
||
account_info = dialog_manager.current_context().dialog_data.get("account_info")
|
||
if not account_info:
|
||
return {"account_info": "Информация об аккаунте недоступна."}
|
||
return {"account_info": account_info}
|