49 lines
2.0 KiB
Python
49 lines
2.0 KiB
Python
"""Доступ к данным экономики (деньги, опыт, очищение, яички)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from common.db.pool import get_pool
|
|
|
|
|
|
async def money_action(id_user: int, amount: int, id_action: int = 1) -> None:
|
|
"""Начислить/списать деньги (запись в журнал money)."""
|
|
async with get_pool().connection() as conn:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute("CALL bot.money_action(%s, %s, %s)", (id_user, amount, id_action))
|
|
|
|
|
|
async def get_money(id_user: int) -> int:
|
|
"""Текущий баланс пользователя (SUM по журналу)."""
|
|
async with get_pool().connection() as conn:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute("SELECT bot.get_money(%s) AS money", (id_user,))
|
|
row = await cur.fetchone()
|
|
return int(row["money"] or 0) if row else 0
|
|
|
|
|
|
async def get_eggs(id_user: int) -> int:
|
|
"""Количество яичек пользователя."""
|
|
async with get_pool().connection() as conn:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute("SELECT bot.get_eggs(%s) AS eggs", (id_user,))
|
|
row = await cur.fetchone()
|
|
return int(row["eggs"] or 0) if row else 0
|
|
|
|
|
|
async def purification(id_user: int) -> dict[str, Any] | None:
|
|
"""Выполнить ежедневное очищение."""
|
|
async with get_pool().connection() as conn:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute("SELECT * FROM bot.purification(%s)", (id_user,))
|
|
return await cur.fetchone()
|
|
|
|
|
|
async def add_eggs(id_user: int) -> dict[str, Any] | None:
|
|
"""Получить яички (раз в сутки)."""
|
|
async with get_pool().connection() as conn:
|
|
async with conn.cursor() as cur:
|
|
await cur.execute("SELECT * FROM bot.add_eggs(%s)", (id_user,))
|
|
return await cur.fetchone()
|