61 lines
2.3 KiB
Python
61 lines
2.3 KiB
Python
from datetime import date
|
|
from sqlalchemy import select, func, Subquery, cast
|
|
from sqlalchemy.dialects.postgresql import TIMESTAMP
|
|
from models import PaymentRecord, Expense
|
|
from services.base import BaseService
|
|
from services.statistics.common import generate_date_range
|
|
|
|
|
|
class ExpensesStatisticsService(BaseService):
|
|
date_from: date
|
|
date_to: date
|
|
|
|
def _get_expenses_sub(self, model, date_column, amount_column) -> Subquery:
|
|
all_dates = generate_date_range(self.date_from, self.date_to, ["expenses"])
|
|
|
|
expenses = (
|
|
select(
|
|
func.sum(getattr(model, amount_column)).label("expenses"),
|
|
cast(getattr(model, date_column), TIMESTAMP(timezone=False)).label("date"),
|
|
)
|
|
.group_by("date")
|
|
.subquery()
|
|
)
|
|
|
|
expenses_with_gaps_filled = (
|
|
select(
|
|
all_dates.c.date,
|
|
(all_dates.c.expenses + func.coalesce(expenses.c.expenses, 0)).label("expenses"),
|
|
)
|
|
.join(expenses, all_dates.c.date == expenses.c.date, isouter=True)
|
|
.order_by(all_dates.c.date)
|
|
.subquery()
|
|
)
|
|
return expenses_with_gaps_filled
|
|
|
|
def _apply_expenses(self, deals_by_dates: Subquery, expenses_subquery: Subquery):
|
|
return (
|
|
select(
|
|
deals_by_dates.c.date,
|
|
deals_by_dates.c.deals_count,
|
|
deals_by_dates.c.revenue,
|
|
(func.coalesce(deals_by_dates.c.profit, 0) - func.coalesce(expenses_subquery.c.expenses, 0)).label(
|
|
"profit"),
|
|
(deals_by_dates.c.expenses + expenses_subquery.c.expenses).label("expenses"),
|
|
)
|
|
.join(expenses_subquery, expenses_subquery.c.date == deals_by_dates.c.date)
|
|
)
|
|
|
|
def apply_expenses(self, date_from: date, date_to: date, deals_by_dates: Subquery):
|
|
self.date_from, self.date_to = date_from, date_to
|
|
|
|
# Apply salary expenses
|
|
salary_expenses = self._get_expenses_sub(PaymentRecord, "start_date", "amount")
|
|
deals_by_dates = self._apply_expenses(deals_by_dates, salary_expenses)
|
|
|
|
# Apply additional expenses
|
|
additional_expenses = self._get_expenses_sub(Expense, "spent_date", "amount")
|
|
deals_by_dates = self._apply_expenses(deals_by_dates, additional_expenses)
|
|
|
|
return deals_by_dates
|