398 lines
13 KiB
Python
398 lines
13 KiB
Python
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
from datetime import date
|
||
from math import ceil
|
||
|
||
from bot.db.repository import Budget, BudgetRepository, Expense
|
||
|
||
PERIOD_PAGE_SIZE = 8
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class BudgetStatus:
|
||
budget: Budget
|
||
today: date
|
||
days_left: int
|
||
total_spent: float
|
||
remaining: float
|
||
daily_limit: float
|
||
spent_today: float
|
||
remaining_today: float
|
||
is_over_daily: bool
|
||
is_over_budget: bool
|
||
is_expired: bool
|
||
selected: bool = False
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class PeriodExpensesPage:
|
||
budget: Budget
|
||
page: int
|
||
total_pages: int
|
||
total_count: int
|
||
total_sum: float
|
||
page_size: int
|
||
items: list[Expense]
|
||
|
||
|
||
class BudgetService:
|
||
def __init__(self, repo: BudgetRepository) -> None:
|
||
self._repo = repo
|
||
|
||
async def list_user_statuses(self, user_id: int) -> list[BudgetStatus]:
|
||
budgets = await self._repo.list_budgets_for_user(user_id)
|
||
selected_id = await self._repo.get_selected_budget_id(user_id)
|
||
return [
|
||
await self.get_status_for_budget(
|
||
budget,
|
||
selected=(budget.id == selected_id),
|
||
)
|
||
for budget in budgets
|
||
]
|
||
|
||
async def create_budget(
|
||
self,
|
||
user_id: int,
|
||
total_amount: float,
|
||
end_date: date,
|
||
*,
|
||
name: str = "Бюджет",
|
||
start_date: date | None = None,
|
||
is_active: bool = True,
|
||
select: bool = True,
|
||
) -> BudgetStatus:
|
||
start = start_date or date.today()
|
||
if end_date < start:
|
||
raise ValueError("Дата окончания не может быть раньше сегодняшнего дня")
|
||
if total_amount <= 0:
|
||
raise ValueError("Сумма бюджета должна быть больше нуля")
|
||
|
||
budget = await self._repo.create_budget(
|
||
user_id,
|
||
name=name,
|
||
total_amount=total_amount,
|
||
start_date=start,
|
||
end_date=end_date,
|
||
is_active=is_active,
|
||
select=select,
|
||
)
|
||
return await self.get_status_for_budget(budget, selected=select)
|
||
|
||
async def update_budget(
|
||
self,
|
||
user_id: int,
|
||
budget_id: int,
|
||
*,
|
||
name: str | None = None,
|
||
total_amount: float | None = None,
|
||
end_date: date | None = None,
|
||
start_date: date | None = None,
|
||
reset_expenses: bool = False,
|
||
) -> BudgetStatus:
|
||
existing = await self._repo.get_budget_for_user(budget_id, user_id)
|
||
if existing is None:
|
||
raise ValueError("Бюджет не найден")
|
||
|
||
next_end = end_date if end_date is not None else existing.end_date
|
||
next_start = start_date if start_date is not None else existing.start_date
|
||
if next_end < next_start:
|
||
raise ValueError("Дата окончания не может быть раньше даты начала")
|
||
if total_amount is not None and total_amount <= 0:
|
||
raise ValueError("Сумма бюджета должна быть больше нуля")
|
||
|
||
if reset_expenses:
|
||
await self._repo.clear_expenses_for_budget_id(budget_id)
|
||
|
||
budget = await self._repo.update_budget(
|
||
budget_id,
|
||
user_id,
|
||
name=name,
|
||
total_amount=total_amount,
|
||
start_date=start_date,
|
||
end_date=end_date,
|
||
)
|
||
selected_id = await self._repo.get_selected_budget_id(user_id)
|
||
return await self.get_status_for_budget(
|
||
budget,
|
||
selected=(budget.id == selected_id),
|
||
)
|
||
|
||
async def set_budget_active(
|
||
self,
|
||
user_id: int,
|
||
budget_id: int,
|
||
is_active: bool,
|
||
) -> BudgetStatus:
|
||
budget = await self._repo.set_budget_active(budget_id, user_id, is_active)
|
||
selected_id = await self._repo.get_selected_budget_id(user_id)
|
||
if not is_active and selected_id == budget_id:
|
||
# Prefer another active budget as selected.
|
||
others = await self._repo.list_budgets_for_user(user_id)
|
||
next_selected = next((b.id for b in others if b.is_active), None)
|
||
await self._repo.set_selected_budget_id(user_id, next_selected)
|
||
selected_id = next_selected
|
||
return await self.get_status_for_budget(
|
||
budget,
|
||
selected=(budget.id == selected_id),
|
||
)
|
||
|
||
async def select_budget(self, user_id: int, budget_id: int) -> BudgetStatus:
|
||
budget = await self._repo.get_budget_for_user(budget_id, user_id)
|
||
if budget is None:
|
||
raise ValueError("Бюджет не найден")
|
||
await self._repo.set_selected_budget_id(user_id, budget_id)
|
||
return await self.get_status_for_budget(budget, selected=True)
|
||
|
||
async def set_budget(
|
||
self,
|
||
user_id: int,
|
||
total_amount: float,
|
||
end_date: date,
|
||
start_date: date | None = None,
|
||
reset_expenses: bool = True,
|
||
name: str | None = None,
|
||
budget_id: int | None = None,
|
||
) -> BudgetStatus:
|
||
"""Backward-compatible: update selected/specified budget or create new."""
|
||
if budget_id is not None:
|
||
return await self.update_budget(
|
||
user_id,
|
||
budget_id,
|
||
name=name,
|
||
total_amount=total_amount,
|
||
end_date=end_date,
|
||
start_date=start_date,
|
||
reset_expenses=reset_expenses,
|
||
)
|
||
|
||
current = await self._repo.resolve_budget(user_id)
|
||
if current is None:
|
||
return await self.create_budget(
|
||
user_id,
|
||
total_amount,
|
||
end_date,
|
||
name=name or "Бюджет",
|
||
start_date=start_date,
|
||
)
|
||
return await self.update_budget(
|
||
user_id,
|
||
current.id,
|
||
name=name,
|
||
total_amount=total_amount,
|
||
end_date=end_date,
|
||
start_date=start_date,
|
||
reset_expenses=reset_expenses,
|
||
)
|
||
|
||
async def add_expense(
|
||
self,
|
||
user_id: int,
|
||
amount: float,
|
||
note: str | None = None,
|
||
spent_at: date | None = None,
|
||
budget_id: int | None = None,
|
||
) -> BudgetStatus:
|
||
if amount <= 0:
|
||
raise ValueError("Сумма траты должна быть больше нуля")
|
||
budget = await self._repo.resolve_budget(
|
||
user_id,
|
||
budget_id,
|
||
require_active=True,
|
||
)
|
||
if budget is None:
|
||
raise ValueError("Сначала задай бюджет: /budget")
|
||
|
||
day = spent_at or date.today()
|
||
if day > date.today():
|
||
raise ValueError("Нельзя добавить трату на будущую дату")
|
||
if day < budget.start_date or day > budget.end_date:
|
||
raise ValueError(
|
||
"Дата вне периода бюджета "
|
||
f"({budget.start_date.strftime('%d.%m.%Y')}–"
|
||
f"{budget.end_date.strftime('%d.%m.%Y')})"
|
||
)
|
||
|
||
await self._repo.add_expense(
|
||
user_id,
|
||
budget.id,
|
||
amount=amount,
|
||
note=note,
|
||
spent_at=day,
|
||
)
|
||
selected_id = await self._repo.get_selected_budget_id(user_id)
|
||
return await self.get_status_for_budget(
|
||
budget,
|
||
selected=(budget.id == selected_id),
|
||
)
|
||
|
||
async def expenses_on_date(
|
||
self,
|
||
user_id: int,
|
||
day: date,
|
||
budget_id: int | None = None,
|
||
) -> list[Expense]:
|
||
budget = await self._repo.resolve_budget(user_id, budget_id)
|
||
if budget is None:
|
||
raise ValueError("Сначала задай бюджет: /budget")
|
||
return await self._repo.list_expenses_on_date(budget.id, day)
|
||
|
||
async def today_expenses(
|
||
self,
|
||
user_id: int,
|
||
today: date | None = None,
|
||
budget_id: int | None = None,
|
||
) -> list[Expense]:
|
||
day = today or date.today()
|
||
return await self.expenses_on_date(user_id, day, budget_id=budget_id)
|
||
|
||
async def get_expenses_on_date_page(
|
||
self,
|
||
user_id: int,
|
||
day: date,
|
||
page: int = 0,
|
||
page_size: int = PERIOD_PAGE_SIZE,
|
||
budget_id: int | None = None,
|
||
) -> PeriodExpensesPage:
|
||
budget = await self._repo.resolve_budget(user_id, budget_id)
|
||
if budget is None:
|
||
raise ValueError("Сначала задай бюджет: /budget")
|
||
|
||
if page < 0:
|
||
page = 0
|
||
if page_size < 1:
|
||
page_size = PERIOD_PAGE_SIZE
|
||
|
||
items_all = await self._repo.list_expenses_on_date(budget.id, day)
|
||
total_count = len(items_all)
|
||
total_sum = sum(item.amount for item in items_all)
|
||
total_pages = max(1, ceil(total_count / page_size)) if total_count else 1
|
||
if page >= total_pages:
|
||
page = total_pages - 1
|
||
|
||
start = page * page_size
|
||
items = items_all[start : start + page_size]
|
||
return PeriodExpensesPage(
|
||
budget=budget,
|
||
page=page,
|
||
total_pages=total_pages,
|
||
total_count=total_count,
|
||
total_sum=total_sum,
|
||
page_size=page_size,
|
||
items=items,
|
||
)
|
||
|
||
async def get_period_expenses_page(
|
||
self,
|
||
user_id: int,
|
||
page: int = 0,
|
||
page_size: int = PERIOD_PAGE_SIZE,
|
||
budget_id: int | None = None,
|
||
) -> PeriodExpensesPage:
|
||
budget = await self._repo.resolve_budget(user_id, budget_id)
|
||
if budget is None:
|
||
raise ValueError("Сначала задай бюджет: /budget")
|
||
|
||
if page < 0:
|
||
page = 0
|
||
if page_size < 1:
|
||
page_size = PERIOD_PAGE_SIZE
|
||
|
||
total_count = await self._repo.count_expenses_for_budget(budget.id)
|
||
total_sum = await self._repo.sum_expenses_for_budget(budget.id)
|
||
total_pages = max(1, ceil(total_count / page_size)) if total_count else 1
|
||
if page >= total_pages:
|
||
page = total_pages - 1
|
||
|
||
items = await self._repo.list_expenses_for_budget(
|
||
budget.id,
|
||
limit=page_size,
|
||
offset=page * page_size,
|
||
)
|
||
return PeriodExpensesPage(
|
||
budget=budget,
|
||
page=page,
|
||
total_pages=total_pages,
|
||
total_count=total_count,
|
||
total_sum=total_sum,
|
||
page_size=page_size,
|
||
items=items,
|
||
)
|
||
|
||
async def list_budget_summaries(self) -> list[BudgetStatus]:
|
||
budgets = await self._repo.list_all_budgets()
|
||
return [await self.get_status_for_budget(budget) for budget in budgets]
|
||
|
||
async def undo_last_expense(
|
||
self,
|
||
user_id: int,
|
||
budget_id: int | None = None,
|
||
) -> tuple[BudgetStatus | None, float | None]:
|
||
budget = await self._repo.resolve_budget(user_id, budget_id)
|
||
target_id = budget.id if budget else None
|
||
deleted = await self._repo.delete_last_expense(user_id, target_id)
|
||
if deleted is None:
|
||
return None, None
|
||
status = await self.get_status(
|
||
user_id,
|
||
budget_id=deleted.budget_id,
|
||
)
|
||
return status, deleted.amount
|
||
|
||
async def get_status(
|
||
self,
|
||
user_id: int,
|
||
today: date | None = None,
|
||
budget_id: int | None = None,
|
||
) -> BudgetStatus:
|
||
budget = await self._repo.resolve_budget(user_id, budget_id)
|
||
if budget is None:
|
||
raise ValueError("Сначала задай бюджет: /budget")
|
||
selected_id = await self._repo.get_selected_budget_id(user_id)
|
||
return await self.get_status_for_budget(
|
||
budget,
|
||
today=today,
|
||
selected=(budget.id == selected_id),
|
||
)
|
||
|
||
async def get_status_for_budget(
|
||
self,
|
||
budget: Budget,
|
||
today: date | None = None,
|
||
*,
|
||
selected: bool = False,
|
||
) -> BudgetStatus:
|
||
day = today or date.today()
|
||
total_spent = await self._repo.sum_expenses_for_budget(budget.id)
|
||
spent_today = await self._repo.spent_on_date(budget.id, day)
|
||
remaining = budget.total_amount - total_spent
|
||
|
||
if day > budget.end_date:
|
||
days_left = 0
|
||
daily_limit = 0.0
|
||
remaining_today = 0.0
|
||
is_expired = True
|
||
else:
|
||
days_left = (budget.end_date - day).days + 1
|
||
remaining_at_day_start = remaining + spent_today
|
||
daily_limit = (
|
||
remaining_at_day_start / days_left if days_left > 0 else 0.0
|
||
)
|
||
remaining_today = daily_limit - spent_today
|
||
is_expired = False
|
||
|
||
return BudgetStatus(
|
||
budget=budget,
|
||
today=day,
|
||
days_left=days_left,
|
||
total_spent=total_spent,
|
||
remaining=remaining,
|
||
daily_limit=max(daily_limit, 0.0),
|
||
spent_today=spent_today,
|
||
remaining_today=remaining_today,
|
||
is_over_daily=spent_today > daily_limit and days_left > 0,
|
||
is_over_budget=remaining < 0,
|
||
is_expired=is_expired,
|
||
selected=selected,
|
||
)
|